File size: 3,673 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 | """Admin 调试工作台占位模型。"""
from __future__ import annotations
import json
from typing import Any, ClassVar
from django.db import models
from django.utils.translation import gettext_lazy as _
from apps.core.security.scrub import scrub_for_storage, scrub_text
class McpWorkbench(models.Model):
id: int
name = models.CharField(max_length=64, default="MCP Workbench")
class Meta:
managed = False
verbose_name = _("MCP 调试工作台")
verbose_name_plural = _("MCP 调试工作台")
class McpWorkbenchExecution(models.Model):
"""MCP 调试执行历史。"""
id: int
provider = models.CharField(max_length=64, verbose_name=_("Provider"))
tool_name = models.CharField(max_length=128, verbose_name=_("工具名称"))
arguments = models.JSONField(default=dict, blank=True, verbose_name=_("请求参数"))
response_data = models.JSONField(default=dict, blank=True, verbose_name=_("响应数据"))
response_raw = models.JSONField(default=dict, blank=True, verbose_name=_("原始响应"))
response_meta = models.JSONField(default=dict, blank=True, verbose_name=_("响应元信息"))
success = models.BooleanField(default=False, verbose_name=_("是否成功"))
error_code = models.CharField(max_length=64, blank=True, default="", verbose_name=_("错误码"))
error_message = models.TextField(blank=True, default="", verbose_name=_("错误信息"))
duration_ms = models.PositiveIntegerField(default=0, verbose_name=_("耗时(毫秒)"))
requested_transport = models.CharField(max_length=32, blank=True, default="", verbose_name=_("请求协议"))
actual_transport = models.CharField(max_length=32, blank=True, default="", verbose_name=_("实际协议"))
operator_username = models.CharField(max_length=150, blank=True, default="", verbose_name=_("操作人"))
replay_of = models.ForeignKey(
"self",
null=True,
blank=True,
on_delete=models.SET_NULL,
related_name="replayed_records",
verbose_name=_("重放来源"),
)
created_at = models.DateTimeField(auto_now_add=True, verbose_name=_("创建时间"))
class Meta:
verbose_name = _("MCP 调试执行历史")
verbose_name_plural = _("MCP 调试执行历史")
ordering: ClassVar = ["-created_at"]
indexes: ClassVar = [
models.Index(fields=["provider", "tool_name", "-created_at"]),
models.Index(fields=["success", "-created_at"]),
models.Index(fields=["operator_username", "-created_at"]),
]
def __str__(self) -> str:
status = "success" if self.success else "failed"
return f"{self.provider}:{self.tool_name}:{status}"
def save(self, *args: Any, **kwargs: Any) -> None:
self.arguments = self._sanitize_json(self.arguments)
self.response_data = self._sanitize_json(self.response_data)
self.response_raw = self._sanitize_json(self.response_raw)
self.response_meta = self._sanitize_json(self.response_meta)
if self.error_message:
self.error_message = scrub_text(self.error_message)
super().save(*args, **kwargs)
@staticmethod
def _sanitize_json(value: Any) -> Any:
scrubbed = scrub_for_storage(value)
try:
serialized = json.dumps(scrubbed, ensure_ascii=False, sort_keys=True, default=str)
except Exception:
return scrubbed
if len(serialized) <= 50000:
return scrubbed
return {
"_truncated": True,
"preview": serialized[:12000],
"original_length": len(serialized),
}
|