MoviePilot/app/agent/tools/impl/mcp.py

"""
外部 MCP 工具适配器。
"""

import hashlib
import json
from typing import TYPE_CHECKING, Any, Optional

from pydantic import Field, PrivateAttr

from app.agent.mcp import AgentMcpToolSpec, agent_mcp_manager
from app.agent.tools.base import MoviePilotTool
from app.agent.tools.tags import ToolTag

if TYPE_CHECKING:
    from app.agent.callback import StreamingHandler


class McpExternalTool(MoviePilotTool):
    """
    将外部 MCP 工具包装为 MoviePilot Agent 工具。

    :ivar _spec: 已绑定的外部 MCP 服务与具体工具调用契约
    """

    name: str = Field(
        default="mcp_external_tool",
        description="供模型调用的稳定工具名称",
    )
    tags: list[str] = Field(
        default=[ToolTag.Read, ToolTag.Admin],
        description="工具的能力和权限分类标签，初始化时会统一归一化",
    )
    description: str = Field(
        default="调用为 MoviePilot Agent 配置的外部 MCP 工具。",
        description="供模型理解工具用途、输入约束和调用边界的说明",
    )
    args_schema: dict[str, Any] = Field(
        default={'type': 'object', 'properties': {}, 'required': []},
        description="工具输入参数的验证模型或 JSON Schema",
    )
    require_admin: bool = Field(
        default=True,
        description="是否仅允许系统管理员调用",
    )

    _spec: AgentMcpToolSpec = PrivateAttr()

    def __init__(self, spec: AgentMcpToolSpec, session_id: str, user_id: str) -> None:
        """
        绑定已配置的 MCP 工具及其输入协议。
        """
        super().__init__(
            session_id=session_id,
            user_id=user_id,
            name=spec.agent_tool_name,
            description=spec.description or f"Call external MCP tool {spec.name} on {spec.server.name}.",
            args_schema=spec.input_schema,
            require_admin=spec.server.require_admin,
        )
        self._spec = spec

    def get_tool_message(self, **kwargs: object) -> Optional[str]:
        """
        根据 MCP 工具信息生成友好的提示消息。
        """
        return f"调用 MCP 工具: {self._spec.server.name}/{self._spec.name}"

    async def run(self, **kwargs: object) -> str:  # type: ignore[override]
        """
        调用外部 MCP 工具。

        :param kwargs: 传递给外部 MCP 工具的参数
        :return: MCP 工具返回内容
        """
        result = await agent_mcp_manager.call_server_tool(
            server=self._spec.server,
            tool_name=self._spec.name,
            arguments=kwargs,
        )
        return self._format_mcp_result(result)

    @staticmethod
    def _format_mcp_result(result: Any) -> str:
        """
        将 MCP tools/call 返回结构转换为 Agent 可读文本。
        """
        if isinstance(result, dict):
            if result.get("isError") is True:
                return json.dumps(result, ensure_ascii=False, indent=2, default=str)
            content = result.get("content")
            if isinstance(content, list):
                parts = []
                for item in content:
                    if not isinstance(item, dict):
                        continue
                    if item.get("type") == "text" and item.get("text") is not None:
                        parts.append(str(item["text"]))
                    elif item:
                        parts.append(json.dumps(item, ensure_ascii=False, default=str))
                if parts:
                    return "\n".join(parts)
        if isinstance(result, str):
            return result
        return json.dumps(result, ensure_ascii=False, indent=2, default=str)


def _mcp_binding_identity(spec: AgentMcpToolSpec) -> str:
    """
    生成不暴露端点配置的稳定 MCP 工具绑定身份。
    """
    payload = json.dumps(
        {
            "agent_tool_name": spec.agent_tool_name,
            "server_id": spec.server.id,
            "tool_name": spec.name,
        },
        ensure_ascii=False,
        allow_nan=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


async def create_external_mcp_tools(
    *,
    session_id: str,
    user_id: str,
    channel: Optional[str] = None,
    source: Optional[str] = None,
    username: Optional[str] = None,
    stream_handler: Optional["StreamingHandler"] = None,
    agent_context: Optional[dict[str, object]] = None,
    specs: Optional[list[AgentMcpToolSpec]] = None,
) -> list[McpExternalTool]:
    """
    创建当前已启用的外部 MCP Agent 工具列表。
    """
    tools: list[McpExternalTool] = []
    current_specs = specs
    if current_specs is None:
        current_specs = await agent_mcp_manager.list_enabled_tool_specs()
    for spec in current_specs:
        tool = McpExternalTool(spec=spec, session_id=session_id, user_id=user_id)
        tool.set_message_attr(channel=channel, source=source, username=username)
        tool.set_stream_handler(stream_handler=stream_handler)
        tool.set_agent_context(agent_context=agent_context)
        object.__setattr__(
            tool,
            "_agent_tool_source",
            f"mcp:{spec.server.id}",
        )
        object.__setattr__(
            tool,
            "_agent_tool_binding",
            _mcp_binding_identity(spec),
        )
        tools.append(tool)
    return tools