| """MCP 调试工作台服务实现。""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| import time |
| from datetime import datetime |
| from typing import Any |
|
|
| from django.core.cache import cache |
| from django.db.models import QuerySet |
| from django.utils import timezone |
|
|
| from apps.core.exceptions import PermissionDenied, ValidationException |
| from apps.core.security.scrub import scrub_for_storage |
| from apps.enterprise_data.models import McpWorkbenchExecution |
| from apps.enterprise_data.services.metrics_service import EnterpriseDataMetricsService |
| from apps.enterprise_data.services.provider_registry import EnterpriseProviderRegistry |
| from apps.enterprise_data.services.types import ( |
| DEFAULT_ALERT_AVG_LATENCY_MS_THRESHOLD, |
| DEFAULT_ALERT_FALLBACK_RATE_THRESHOLD, |
| DEFAULT_ALERT_MIN_SAMPLES, |
| DEFAULT_ALERT_SUCCESS_RATE_THRESHOLD, |
| DEFAULT_METRICS_WINDOW_SECONDS, |
| ) |
|
|
| _SAMPLE_CACHE_TTL_SECONDS = 24 * 60 * 60 |
| _MAX_SAMPLE_JSON_CHARS = 12000 |
|
|
|
|
| class McpWorkbenchService: |
| """供 Admin 调试页面使用的 MCP 工具编排服务。""" |
|
|
| def __init__( |
| self, |
| *, |
| registry: EnterpriseProviderRegistry | None = None, |
| sample_ttl_seconds: int = _SAMPLE_CACHE_TTL_SECONDS, |
| persist_history: bool = True, |
| enforce_superuser: bool = True, |
| metrics_service: EnterpriseDataMetricsService | None = None, |
| ) -> None: |
| self._registry = registry or EnterpriseProviderRegistry() |
| self._sample_ttl_seconds = max(60, int(sample_ttl_seconds)) |
| self._persist_history = bool(persist_history) |
| self._enforce_superuser = bool(enforce_superuser) |
| self._metrics = metrics_service or EnterpriseDataMetricsService( |
| window_seconds=self._read_registry_int("get_metrics_window_seconds", DEFAULT_METRICS_WINDOW_SECONDS), |
| alert_min_samples=self._read_registry_int("get_alert_min_samples", DEFAULT_ALERT_MIN_SAMPLES), |
| alert_success_rate_threshold=self._read_registry_float( |
| "get_alert_success_rate_threshold", |
| DEFAULT_ALERT_SUCCESS_RATE_THRESHOLD, |
| ), |
| alert_fallback_rate_threshold=self._read_registry_float( |
| "get_alert_fallback_rate_threshold", |
| DEFAULT_ALERT_FALLBACK_RATE_THRESHOLD, |
| ), |
| alert_avg_latency_ms_threshold=self._read_registry_int( |
| "get_alert_avg_latency_ms_threshold", |
| DEFAULT_ALERT_AVG_LATENCY_MS_THRESHOLD, |
| ), |
| ) |
|
|
| def list_providers(self) -> list[dict[str, Any]]: |
| descriptors = self._registry.list_providers() |
| return [ |
| { |
| "name": item.name, |
| "enabled": item.enabled, |
| "is_default": item.is_default, |
| "transport": item.transport, |
| "capabilities": item.capabilities, |
| } |
| for item in descriptors |
| ] |
|
|
| def describe_tools(self, *, provider: str | None = None, actor_is_superuser: bool = False) -> dict[str, Any]: |
| self._ensure_superuser(actor_is_superuser=actor_is_superuser) |
| selected_provider = self._registry.get_provider(provider) |
| provider_name = selected_provider.name |
| tools = selected_provider.describe_tools() |
| normalized_tools: list[dict[str, Any]] = [] |
| for tool in tools: |
| tool_name = str(tool.get("name", "") or "").strip() |
| if not tool_name: |
| continue |
| input_schema = tool.get("input_schema") |
| if not isinstance(input_schema, dict): |
| input_schema = {} |
| required = input_schema.get("required") |
| if not isinstance(required, list): |
| required = [] |
| sample = self._load_sample(provider=provider_name, tool_name=tool_name) |
| normalized_tools.append( |
| { |
| "name": tool_name, |
| "description": str(tool.get("description", "") or "").strip(), |
| "input_schema": input_schema, |
| "required_fields": [str(item) for item in required if isinstance(item, str)], |
| "sample": sample if isinstance(sample, dict) else None, |
| } |
| ) |
| normalized_tools.sort(key=lambda item: item["name"]) |
| return { |
| "provider": provider_name, |
| "transport": selected_provider.transport, |
| "tools": normalized_tools, |
| } |
|
|
| def execute_tool( |
| self, |
| *, |
| provider: str | None = None, |
| tool_name: str, |
| arguments: dict[str, Any], |
| actor_username: str = "", |
| actor_is_superuser: bool = False, |
| replay_of_id: int | None = None, |
| ) -> dict[str, Any]: |
| self._ensure_superuser(actor_is_superuser=actor_is_superuser) |
| normalized_tool = str(tool_name or "").strip() |
| if not normalized_tool: |
| raise ValidationException(message="tool_name 不能为空", code="INVALID_TOOL_NAME") |
| if not isinstance(arguments, dict): |
| raise ValidationException(message="arguments 必须为 JSON Object", code="INVALID_TOOL_ARGUMENTS") |
|
|
| selected_provider = self._registry.get_provider(provider) |
| started = time.perf_counter() |
| replay_of = self._resolve_replay_record(replay_of_id=replay_of_id) |
| try: |
| response = selected_provider.execute_tool(tool_name=normalized_tool, arguments=arguments) |
| duration_ms = int((time.perf_counter() - started) * 1000) |
|
|
| meta = { |
| "provider": selected_provider.name, |
| "tool": response.tool, |
| "duration_ms": duration_ms, |
| } |
| if response.meta: |
| meta.update(response.meta) |
| duration_for_metrics = int(meta.get("duration_ms", duration_ms) or duration_ms) |
| fallback_used = bool(meta.get("fallback_used", False)) |
| observability = self._metrics.record( |
| provider=selected_provider.name, |
| capability=f"workbench:{response.tool}", |
| success=True, |
| duration_ms=duration_for_metrics, |
| fallback_used=fallback_used, |
| ) |
| meta["observability"] = observability |
| masked_arguments = self._mask_payload(arguments) |
| masked_data = self._mask_payload(response.data) |
| masked_raw = self._mask_payload(response.raw) |
| masked_meta = self._mask_payload(meta) |
| payload = { |
| "provider": selected_provider.name, |
| "tool": response.tool, |
| "arguments": masked_arguments, |
| "meta": masked_meta, |
| "data": masked_data, |
| "raw": masked_raw, |
| } |
| self._store_sample( |
| provider=selected_provider.name, |
| tool_name=response.tool, |
| data=masked_data, |
| captured_at=timezone.now(), |
| ) |
| self._create_history( |
| provider=selected_provider.name, |
| tool_name=response.tool, |
| arguments=masked_arguments, |
| response_data=masked_data, |
| response_raw=masked_raw, |
| response_meta=masked_meta, |
| success=True, |
| error_code="", |
| error_message="", |
| duration_ms=duration_ms, |
| operator_username=actor_username, |
| replay_of=replay_of, |
| ) |
| return payload |
| except Exception as exc: |
| duration_ms = int((time.perf_counter() - started) * 1000) |
| self._metrics.record( |
| provider=selected_provider.name, |
| capability=f"workbench:{normalized_tool}", |
| success=False, |
| duration_ms=duration_ms, |
| fallback_used=False, |
| ) |
| self._create_history( |
| provider=selected_provider.name, |
| tool_name=normalized_tool, |
| arguments=self._mask_payload(arguments), |
| response_data={}, |
| response_raw={}, |
| response_meta={}, |
| success=False, |
| error_code=type(exc).__name__, |
| error_message=str(exc).strip() or type(exc).__name__, |
| duration_ms=duration_ms, |
| operator_username=actor_username, |
| replay_of=replay_of, |
| ) |
| raise |
|
|
| def replay_execution( |
| self, |
| *, |
| execution_id: int, |
| actor_username: str = "", |
| actor_is_superuser: bool = False, |
| ) -> dict[str, Any]: |
| self._ensure_superuser(actor_is_superuser=actor_is_superuser) |
| try: |
| record = McpWorkbenchExecution.objects.get(id=execution_id) |
| except McpWorkbenchExecution.DoesNotExist as exc: |
| raise ValidationException( |
| message=f"未找到执行记录: {execution_id}", |
| code="WORKBENCH_EXECUTION_NOT_FOUND", |
| errors={"execution_id": execution_id}, |
| ) from exc |
|
|
| arguments = record.arguments if isinstance(record.arguments, dict) else {} |
| payload = self.execute_tool( |
| provider=record.provider, |
| tool_name=record.tool_name, |
| arguments=arguments, |
| actor_username=actor_username, |
| actor_is_superuser=actor_is_superuser, |
| replay_of_id=record.id, |
| ) |
| payload["replay_of"] = record.id |
| return payload |
|
|
| def list_history( |
| self, |
| *, |
| provider: str | None = None, |
| tool_name: str | None = None, |
| limit: int = 20, |
| actor_is_superuser: bool = False, |
| ) -> list[dict[str, Any]]: |
| self._ensure_superuser(actor_is_superuser=actor_is_superuser) |
| queryset: QuerySet[McpWorkbenchExecution] = McpWorkbenchExecution.objects.all() |
| normalized_provider = str(provider or "").strip() |
| if normalized_provider: |
| queryset = queryset.filter(provider=normalized_provider) |
| normalized_tool = str(tool_name or "").strip() |
| if normalized_tool: |
| queryset = queryset.filter(tool_name=normalized_tool) |
| rows = queryset.order_by("-created_at")[: max(1, min(100, int(limit or 20)))] |
| return [ |
| { |
| "id": row.id, |
| "provider": row.provider, |
| "tool_name": row.tool_name, |
| "success": row.success, |
| "duration_ms": row.duration_ms, |
| "error_code": row.error_code, |
| "error_message": row.error_message, |
| "operator_username": row.operator_username, |
| "requested_transport": row.requested_transport, |
| "actual_transport": row.actual_transport, |
| "created_at": row.created_at.isoformat() if row.created_at else "", |
| "arguments": row.arguments if isinstance(row.arguments, dict) else {}, |
| "replay_of_id": row.replay_of_id, |
| } |
| for row in rows |
| ] |
|
|
| @staticmethod |
| def _mask_payload(payload: Any) -> Any: |
| return scrub_for_storage(payload) |
|
|
| def _ensure_superuser(self, *, actor_is_superuser: bool) -> None: |
| if not self._enforce_superuser: |
| return |
| if bool(actor_is_superuser): |
| return |
| raise PermissionDenied(message="仅超级管理员可使用 MCP 调试工作台", code="PERMISSION_DENIED") |
|
|
| def _read_registry_int(self, method_name: str, default: int) -> int: |
| getter = getattr(self._registry, method_name, None) |
| if not callable(getter): |
| return default |
| try: |
| value = int(getter() or default) |
| except Exception: |
| return default |
| return value if value > 0 else default |
|
|
| def _read_registry_float(self, method_name: str, default: float) -> float: |
| getter = getattr(self._registry, method_name, None) |
| if not callable(getter): |
| return default |
| try: |
| value = float(getter() or default) |
| except Exception: |
| return default |
| return value |
|
|
| def _create_history( |
| self, |
| *, |
| provider: str, |
| tool_name: str, |
| arguments: dict[str, Any], |
| response_data: Any, |
| response_raw: Any, |
| response_meta: dict[str, Any], |
| success: bool, |
| error_code: str, |
| error_message: str, |
| duration_ms: int, |
| operator_username: str, |
| replay_of: McpWorkbenchExecution | None, |
| ) -> None: |
| if not self._persist_history: |
| return |
|
|
| requested_transport = str(response_meta.get("requested_transport", "") or "").strip() |
| actual_transport = str(response_meta.get("transport", "") or "").strip() |
| McpWorkbenchExecution.objects.create( |
| provider=provider, |
| tool_name=tool_name, |
| arguments=arguments if isinstance(arguments, dict) else {}, |
| response_data=response_data if isinstance(response_data, (dict, list)) else {"value": str(response_data)}, |
| response_raw=response_raw if isinstance(response_raw, (dict, list)) else {"value": str(response_raw)}, |
| response_meta=response_meta if isinstance(response_meta, dict) else {}, |
| success=success, |
| error_code=str(error_code or "").strip(), |
| error_message=str(error_message or "").strip(), |
| duration_ms=max(0, int(duration_ms or 0)), |
| requested_transport=requested_transport, |
| actual_transport=actual_transport, |
| operator_username=str(operator_username or "").strip()[:150], |
| replay_of=replay_of, |
| ) |
|
|
| @staticmethod |
| def _resolve_replay_record(*, replay_of_id: int | None) -> McpWorkbenchExecution | None: |
| if not replay_of_id: |
| return None |
| try: |
| return McpWorkbenchExecution.objects.get(id=int(replay_of_id)) |
| except (TypeError, ValueError, McpWorkbenchExecution.DoesNotExist): |
| return None |
|
|
| def _load_sample(self, *, provider: str, tool_name: str) -> dict[str, Any] | None: |
| sample = cache.get(self._sample_cache_key(provider=provider, tool_name=tool_name)) |
| if isinstance(sample, dict): |
| return sample |
|
|
| record = ( |
| McpWorkbenchExecution.objects.filter(provider=provider, tool_name=tool_name, success=True) |
| .order_by("-created_at") |
| .first() |
| ) |
| if record is None: |
| return None |
| sample = { |
| "captured_at": record.created_at.isoformat() if record.created_at else "", |
| "data": record.response_data if isinstance(record.response_data, (dict, list)) else {}, |
| } |
| cache.set( |
| self._sample_cache_key(provider=provider, tool_name=tool_name), sample, timeout=self._sample_ttl_seconds |
| ) |
| return sample |
|
|
| def _store_sample(self, *, provider: str, tool_name: str, data: Any, captured_at: datetime) -> None: |
| sample = { |
| "captured_at": captured_at.isoformat(), |
| "data": self._truncate_data(data), |
| } |
| cache.set( |
| self._sample_cache_key(provider=provider, tool_name=tool_name), |
| sample, |
| timeout=self._sample_ttl_seconds, |
| ) |
|
|
| @staticmethod |
| def _truncate_data(data: Any) -> Any: |
| try: |
| serialized = json.dumps(data, ensure_ascii=False, sort_keys=True, default=str) |
| except Exception: |
| return data |
| if len(serialized) <= _MAX_SAMPLE_JSON_CHARS: |
| return data |
| return { |
| "_truncated": True, |
| "preview": serialized[:_MAX_SAMPLE_JSON_CHARS], |
| "original_length": len(serialized), |
| } |
|
|
| @staticmethod |
| def _sample_cache_key(*, provider: str, tool_name: str) -> str: |
| return f"mcp_workbench:sample:{provider}:{tool_name}" |
|
|
