File size: 10,162 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 | """MCP 调试工作台 Admin。"""
from __future__ import annotations
import json
from typing import Any
from django.contrib import admin, messages
from django.template.response import TemplateResponse
from apps.enterprise_data.models import McpWorkbench
from apps.enterprise_data.services import McpWorkbenchService
@admin.register(McpWorkbench)
class McpWorkbenchAdmin(admin.ModelAdmin[McpWorkbench]):
"""MCP tools 调试页。"""
def has_module_permission(self, request) -> bool: # type: ignore[override]
return bool(request.user and request.user.is_active and request.user.is_superuser)
def has_view_permission(self, request, obj: Any | None = None) -> bool: # type: ignore[override]
return bool(request.user and request.user.is_active and request.user.is_superuser)
def has_add_permission(self, request) -> bool: # type: ignore[override]
return False
def has_change_permission(self, request, obj: Any | None = None) -> bool: # type: ignore[override]
return False
def has_delete_permission(self, request, obj: Any | None = None) -> bool: # type: ignore[override]
return False
def get_model_perms(self, request) -> dict[str, bool]:
if not self.has_view_permission(request):
return {}
return {"view": True, "add": False, "change": False, "delete": False}
def changelist_view(self, request, extra_context: dict[str, Any] | None = None): # type: ignore[override]
service = McpWorkbenchService()
providers = service.list_providers()
provider_names = [str(item.get("name", "") or "").strip() for item in providers if item.get("name")]
default_provider = self._pick_default_provider(providers=providers)
selected_provider = self._pick_selected_provider(
request=request,
provider_names=provider_names,
default_provider=default_provider,
)
selected_tool_name = self._pick_selected_tool_name(request=request)
arguments_json = str(request.POST.get("arguments_json", "") or "").strip() if request.method == "POST" else "{}"
execution_result: dict[str, Any] | None = None
execution_error = ""
actor_username = str(getattr(request.user, "username", "") or "").strip()
actor_is_superuser = bool(getattr(request.user, "is_superuser", False))
if request.method == "POST" and request.POST.get("_action") == "execute":
try:
parsed_arguments = self._parse_arguments(arguments_json)
execution_result = service.execute_tool(
provider=selected_provider,
tool_name=selected_tool_name,
arguments=parsed_arguments,
actor_username=actor_username,
actor_is_superuser=actor_is_superuser,
)
arguments_json = self._to_pretty_json(parsed_arguments)
messages.success(request, "工具执行成功")
except Exception as exc:
execution_error = str(exc).strip() or type(exc).__name__
messages.error(request, f"工具执行失败: {execution_error}")
elif request.method == "POST" and request.POST.get("_action") == "replay":
execution_id = request.POST.get("execution_id", "")
try:
execution_result = service.replay_execution(
execution_id=int(str(execution_id or "").strip() or "0"),
actor_username=actor_username,
actor_is_superuser=actor_is_superuser,
)
selected_provider = str(
execution_result.get("provider", selected_provider) or selected_provider
).strip()
selected_tool_name = str(execution_result.get("tool", selected_tool_name) or selected_tool_name).strip()
arguments_payload = execution_result.get("arguments")
if isinstance(arguments_payload, dict):
arguments_json = self._to_pretty_json(arguments_payload)
messages.success(request, "重放执行成功")
except Exception as exc:
execution_error = str(exc).strip() or type(exc).__name__
messages.error(request, f"重放执行失败: {execution_error}")
load_error = ""
tools: list[dict[str, Any]] = []
resolved_provider = selected_provider
try:
described = service.describe_tools(provider=selected_provider, actor_is_superuser=actor_is_superuser)
tools = described.get("tools", [])
resolved_provider = str(described.get("provider", selected_provider) or selected_provider).strip()
except Exception as exc:
load_error = str(exc).strip() or type(exc).__name__
messages.error(request, f"加载工具失败: {load_error}")
selected_tool = self._pick_selected_tool(tools=tools, selected_tool_name=selected_tool_name)
if selected_tool is None and tools:
selected_tool = tools[0]
selected_tool_name = str(selected_tool.get("name", "") or "").strip()
if request.method != "POST" and selected_tool is not None:
arguments_json = self._to_pretty_json(self._build_default_arguments(selected_tool))
history = service.list_history(
provider=resolved_provider,
tool_name=selected_tool_name,
limit=30,
actor_is_superuser=actor_is_superuser,
)
context = {
"title": "MCP 调试工作台",
"opts": self.model._meta,
"providers": providers,
"selected_provider": resolved_provider,
"tools": tools,
"selected_tool": selected_tool,
"arguments_json": arguments_json,
"selected_tool_name": selected_tool_name,
"selected_tool_schema_json": self._to_pretty_json((selected_tool or {}).get("input_schema", {})),
"selected_tool_sample_json": self._to_pretty_json((selected_tool or {}).get("sample") or {}),
"execution_result_json": self._to_pretty_json(execution_result or {}),
"execution_error": execution_error,
"load_error": load_error,
"history": history,
"has_view_permission": self.has_view_permission(request),
"site_header": self.admin_site.site_header,
"site_title": self.admin_site.site_title,
}
return TemplateResponse(
request,
"admin/enterprise_data/mcp_workbench/change_list.html",
context,
)
@staticmethod
def _pick_default_provider(*, providers: list[dict[str, Any]]) -> str:
for item in providers:
if item.get("is_default") and item.get("name"):
return str(item["name"]).strip()
for item in providers:
if item.get("name"):
return str(item["name"]).strip()
return ""
@staticmethod
def _pick_selected_provider(*, request: Any, provider_names: list[str], default_provider: str) -> str:
selected = str(request.POST.get("provider", "") or "").strip()
if not selected:
selected = str(request.GET.get("provider", "") or "").strip()
if selected in provider_names:
return selected
return default_provider
@staticmethod
def _pick_selected_tool_name(*, request: Any) -> str:
selected = str(request.POST.get("tool_name", "") or "").strip()
if selected:
return selected
return str(request.GET.get("tool", "") or "").strip()
@staticmethod
def _pick_selected_tool(*, tools: list[dict[str, Any]], selected_tool_name: str) -> dict[str, Any] | None:
if not selected_tool_name:
return None
for item in tools:
if str(item.get("name", "") or "").strip() == selected_tool_name:
return item
return None
@staticmethod
def _parse_arguments(arguments_json: str) -> dict[str, Any]:
text = str(arguments_json or "").strip() or "{}"
try:
payload = json.loads(text)
except (TypeError, ValueError) as exc:
raise ValueError("参数 JSON 格式不正确") from exc
if not isinstance(payload, dict):
raise ValueError("参数必须是 JSON Object")
return payload
@staticmethod
def _build_default_arguments(tool: dict[str, Any]) -> dict[str, Any]:
schema = tool.get("input_schema")
if not isinstance(schema, dict):
return {}
properties = schema.get("properties")
if not isinstance(properties, dict):
return {}
required = schema.get("required")
required_set = set(required if isinstance(required, list) else [])
defaults: dict[str, Any] = {}
for field_name, field_schema in properties.items():
if not isinstance(field_name, str):
continue
if field_name not in required_set:
continue
if not isinstance(field_schema, dict):
defaults[field_name] = ""
continue
if "default" in field_schema:
defaults[field_name] = field_schema["default"]
continue
field_type = str(field_schema.get("type", "string") or "string")
if field_type == "number" or field_type == "integer":
defaults[field_name] = 0
elif field_type == "boolean":
defaults[field_name] = False
elif field_type == "array":
defaults[field_name] = []
elif field_type == "object":
defaults[field_name] = {}
else:
defaults[field_name] = ""
return defaults
@staticmethod
def _to_pretty_json(payload: Any) -> str:
try:
return json.dumps(payload, ensure_ascii=False, indent=2, default=str)
except Exception:
return str(payload)
|