diff --git a/app/agent/__init__.py b/app/agent/__init__.py index 1f0e1a91..34ebed57 100644 --- a/app/agent/__init__.py +++ b/app/agent/__init__.py @@ -51,7 +51,9 @@ from app.agent.middleware.tool_selection import ToolSelectorMiddleware from app.agent.middleware.usage import UsageMiddleware from app.agent.prompt import prompt_manager from app.agent.runtime import agent_runtime_manager +from app.agent.mcp import agent_mcp_manager from app.agent.tools.factory import MoviePilotToolFactory +from app.agent.tools.impl.mcp import create_external_mcp_tools from app.chain import ChainBase from app.core.config import settings from app.core.event import eventmanager @@ -1041,6 +1043,7 @@ class MoviePilotAgent: settings.LLM_MAX_ITERATIONS, self._public_runtime_config_signature(runtime_config), agent_runtime_manager.current_signature(), + agent_mcp_manager.config_signature(), ) def _get_cached_agent( @@ -1097,6 +1100,39 @@ class MoviePilotAgent: allow_message_tools=False, ) + async def _initialize_mcp_tools(self) -> List: + """ + 初始化外部 MCP 工具列表。 + """ + return await create_external_mcp_tools( + session_id=self.session_id, + user_id=self.user_id, + channel=self.channel, + source=self.source, + username=self.username, + stream_handler=self.stream_handler, + agent_context=self._tool_context, + ) + + async def _initialize_subagent_mcp_tools(self) -> List: + """ + 初始化子代理可用的外部 MCP 工具列表。 + """ + return await create_external_mcp_tools( + session_id=self.session_id, + user_id=self.user_id, + channel=self.channel, + source=self.source, + username=self.username, + stream_handler=None, + agent_context={ + "user_reply_sent": False, + "reply_mode": None, + "should_dispatch_reply": False, + "is_admin": bool(self._tool_context.get("is_admin")), + }, + ) + async def _create_agent(self, streaming: bool = False): """ 创建 LangGraph Agent（使用 create_agent + SummarizationMiddleware） @@ -1126,6 +1162,7 @@ class MoviePilotAgent: # 工具列表 tools = self._initialize_tools() + tools.extend(await self._initialize_mcp_tools()) skills_middleware = SkillsMiddleware( sources=[str(agent_runtime_manager.skills_dir)], bundled_skills_dir=str(settings.ROOT_PATH / "skills"), @@ -1142,9 +1179,11 @@ class MoviePilotAgent: activity_log_tools = list( getattr(activity_log_middleware, "tools", []) or [] ) + subagent_tools = self._initialize_subagent_tools() + subagent_tools.extend(await self._initialize_subagent_mcp_tools()) subagent_middlewares, subagent_task_tools = create_subagent_middlewares( model=non_streaming_model, - tools=self._initialize_subagent_tools(), + tools=subagent_tools, stream_handler=self.stream_handler, ) max_tools = settings.LLM_MAX_TOOLS diff --git a/app/agent/mcp.py b/app/agent/mcp.py new file mode 100644 index 00000000..cfb0ee7b --- /dev/null +++ b/app/agent/mcp.py @@ -0,0 +1,600 @@ +"""Agent 外部 MCP 客户端与配置管理。""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import os +import re +import uuid +from dataclasses import dataclass +from typing import Any, Optional +from urllib.parse import urljoin + +from app.db.systemconfig_oper import SystemConfigOper +from app.log import logger +from app.schemas.agent import ( + AgentMcpServerConfig, + AgentMcpServerTestResult, + AgentMcpServerToolInfo, +) +from app.schemas.types import SystemConfigKey +from app.utils.http import AsyncRequestUtils + +MCP_PROTOCOL_VERSION = "2025-11-25" +MCP_CLIENT_NAME = "MoviePilot Agent" +DEFAULT_MCP_TIMEOUT = 30 +MCP_TOOL_NAME_PATTERN = re.compile(r"[^a-zA-Z0-9_]+") + + +@dataclass(frozen=True) +class AgentMcpToolSpec: + """已发现的外部 MCP 工具定义。""" + + server: AgentMcpServerConfig + name: str + agent_tool_name: str + description: str + input_schema: dict[str, Any] + + +def _normalize_identifier(value: str, fallback: str = "mcp") -> str: + """把服务器或工具名称转换为 Agent 工具可用的标识片段。""" + normalized = MCP_TOOL_NAME_PATTERN.sub("_", str(value or "").strip()) + normalized = re.sub(r"_+", "_", normalized).strip("_").lower() + if not normalized: + normalized = fallback + if normalized[0].isdigit(): + normalized = f"{fallback}_{normalized}" + return normalized[:64] + + +def _normalize_timeout(value: Any) -> int: + """规范化 MCP 连接和调用超时时间。""" + try: + timeout = int(value or DEFAULT_MCP_TIMEOUT) + except (TypeError, ValueError): + timeout = DEFAULT_MCP_TIMEOUT + return min(max(timeout, 1), 600) + + +def _normalize_string_dict(value: Any) -> dict[str, str]: + """规范化请求头和环境变量字典，移除空键。""" + if not isinstance(value, dict): + return {} + normalized: dict[str, str] = {} + for key, item in value.items(): + normalized_key = str(key or "").strip() + if not normalized_key: + continue + normalized[normalized_key] = str(item or "") + return normalized + + +def _normalize_input_schema(value: Any) -> dict[str, Any]: + """规范化 MCP 工具参数 Schema，保证至少是 object schema。""" + if not isinstance(value, dict): + return {"type": "object", "properties": {}, "required": []} + schema = dict(value) + schema.setdefault("type", "object") + schema.setdefault("properties", {}) + schema.setdefault("required", []) + return schema + + +def _build_agent_tool_name(server: AgentMcpServerConfig, tool_name: str) -> str: + """构造注入 Agent 的外部 MCP 工具名。""" + prefix = server.tool_prefix or f"mcp_{server.name or server.id}" + normalized_prefix = _normalize_identifier(prefix, fallback="mcp") + normalized_tool_name = _normalize_identifier(tool_name, fallback="tool") + if normalized_tool_name.startswith(f"{normalized_prefix}_"): + return normalized_tool_name + return f"{normalized_prefix}_{normalized_tool_name}"[:128] + + +def _jsonrpc_message(method: str, params: Optional[dict[str, Any]] = None, *, request_id: Optional[str] = None) -> dict: + """构造 JSON-RPC 2.0 消息。""" + payload = {"jsonrpc": "2.0", "method": method} + if request_id is not None: + payload["id"] = request_id + if params is not None: + payload["params"] = params + return payload + + +def _raise_for_jsonrpc_error(payload: Any) -> None: + """检查 JSON-RPC 响应错误并转换为运行时异常。""" + if isinstance(payload, dict) and payload.get("error"): + error = payload["error"] + if isinstance(error, dict): + message = error.get("message") or error + else: + message = error + raise RuntimeError(f"MCP JSON-RPC 错误: {message}") + + +def _extract_jsonrpc_result(payload: Any, request_id: str) -> Any: + """从 JSON-RPC 响应中提取 result 字段。""" + if not isinstance(payload, dict): + raise RuntimeError("MCP 响应不是有效 JSON 对象") + if payload.get("id") != request_id: + raise RuntimeError("MCP 响应 ID 与请求不匹配") + _raise_for_jsonrpc_error(payload) + return payload.get("result") + + +async def _iter_sse_events(response) -> Any: + """按 SSE 事件格式迭代响应流。""" + event_name = "message" + data_lines: list[str] = [] + async for raw_line in response.aiter_lines(): + line = raw_line.rstrip("\r") + if not line: + if data_lines: + yield {"event": event_name, "data": "\n".join(data_lines)} + event_name = "message" + data_lines = [] + continue + if line.startswith(":"): + continue + field, _, value = line.partition(":") + if value.startswith(" "): + value = value[1:] + if field == "event": + event_name = value or "message" + elif field == "data": + data_lines.append(value) + if data_lines: + yield {"event": event_name, "data": "\n".join(data_lines)} + + +def _parse_sse_text_response(text: str, request_id: str) -> Any: + """从非流式 SSE 文本响应中提取匹配请求的 JSON-RPC 结果。""" + event_name = "message" + data_lines: list[str] = [] + for raw_line in str(text or "").splitlines(): + line = raw_line.rstrip("\r") + if not line: + if data_lines: + payload = _load_sse_json_payload(event_name, "\n".join(data_lines)) + if isinstance(payload, dict) and payload.get("id") == request_id: + return _extract_jsonrpc_result(payload, request_id) + event_name = "message" + data_lines = [] + continue + if line.startswith(":"): + continue + field, _, value = line.partition(":") + if value.startswith(" "): + value = value[1:] + if field == "event": + event_name = value or "message" + elif field == "data": + data_lines.append(value) + if data_lines: + payload = _load_sse_json_payload(event_name, "\n".join(data_lines)) + if isinstance(payload, dict) and payload.get("id") == request_id: + return _extract_jsonrpc_result(payload, request_id) + raise RuntimeError("MCP SSE 响应中未找到匹配请求") + + +def _load_sse_json_payload(event_name: str, data: str) -> Optional[dict]: + """解析 SSE data 中的 JSON-RPC 消息。""" + if event_name not in {"message", "messages"}: + return None + try: + payload = json.loads(data) + except (TypeError, ValueError): + return None + return payload if isinstance(payload, dict) else None + + +class _StdioMcpSession: + """stdio MCP 会话，按一次操作生命周期启动外部进程。""" + + def __init__(self, server: AgentMcpServerConfig) -> None: + self.server = server + self.process: Optional[asyncio.subprocess.Process] = None + self.stderr_task: Optional[asyncio.Task] = None + + async def __aenter__(self) -> "_StdioMcpSession": + """启动 stdio MCP 子进程。""" + if not self.server.command: + raise RuntimeError("stdio MCP 服务器缺少启动命令") + env = os.environ.copy() + env.update(self.server.env or {}) + self.process = await asyncio.create_subprocess_exec( + self.server.command, + *(self.server.args or []), + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=env, + ) + self.stderr_task = asyncio.create_task(self._drain_stderr()) + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + """结束 stdio MCP 子进程。""" + if self.stderr_task: + self.stderr_task.cancel() + if not self.process: + return + if self.process.returncode is None: + self.process.terminate() + try: + await asyncio.wait_for(self.process.wait(), timeout=2) + except asyncio.TimeoutError: + self.process.kill() + await self.process.wait() + + async def _drain_stderr(self) -> None: + """持续读取子进程 stderr，避免缓冲区阻塞。""" + if not self.process or not self.process.stderr: + return + try: + while True: + line = await self.process.stderr.readline() + if not line: + break + logger.debug(f"MCP stdio[{self.server.name}] stderr: {line.decode(errors='replace').strip()}") + except asyncio.CancelledError: + return + + async def notify(self, method: str, params: Optional[dict[str, Any]] = None) -> None: + """发送不需要响应的 JSON-RPC 通知。""" + await self._write_json(_jsonrpc_message(method, params)) + + async def request(self, method: str, params: Optional[dict[str, Any]] = None) -> Any: + """发送 JSON-RPC 请求并等待响应。""" + request_id = uuid.uuid4().hex + await self._write_json(_jsonrpc_message(method, params, request_id=request_id)) + while True: + payload = await self._read_json() + if payload.get("id") == request_id: + return _extract_jsonrpc_result(payload, request_id) + + async def _write_json(self, payload: dict) -> None: + """写入一行 JSON-RPC 消息。""" + if not self.process or not self.process.stdin: + raise RuntimeError("stdio MCP 进程未启动") + data = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n" + self.process.stdin.write(data.encode("utf-8")) + await self.process.stdin.drain() + + async def _read_json(self) -> dict: + """从 stdout 读取一行 JSON-RPC 消息。""" + if not self.process or not self.process.stdout: + raise RuntimeError("stdio MCP 进程未启动") + timeout = _normalize_timeout(self.server.timeout) + while True: + line = await asyncio.wait_for(self.process.stdout.readline(), timeout=timeout) + if not line: + raise RuntimeError("stdio MCP 进程已退出") + try: + payload = json.loads(line.decode("utf-8")) + except ValueError: + logger.debug(f"忽略非 JSON MCP stdout 行: {line!r}") + continue + if isinstance(payload, dict): + return payload + + +class _HttpMcpSession: + """Streamable HTTP MCP 会话。""" + + def __init__(self, server: AgentMcpServerConfig) -> None: + self.server = server + self.session_id: Optional[str] = None + + async def __aenter__(self) -> "_HttpMcpSession": + """进入 HTTP MCP 会话。""" + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + """退出 HTTP MCP 会话。""" + return None + + async def notify(self, method: str, params: Optional[dict[str, Any]] = None) -> None: + """发送不需要响应的 JSON-RPC 通知。""" + await self._post(_jsonrpc_message(method, params), expect_response=False) + + async def request(self, method: str, params: Optional[dict[str, Any]] = None) -> Any: + """发送 JSON-RPC 请求并等待响应。""" + request_id = uuid.uuid4().hex + return await self._post( + _jsonrpc_message(method, params, request_id=request_id), + expect_response=True, + request_id=request_id, + ) + + async def _post( + self, + payload: dict, + *, + expect_response: bool, + request_id: Optional[str] = None, + ) -> Any: + """向 Streamable HTTP MCP 服务发送一条 JSON-RPC 消息。""" + if not self.server.url: + raise RuntimeError("HTTP MCP 服务器缺少 URL") + headers = { + "Accept": "application/json, text/event-stream", + "Content-Type": "application/json", + **(self.server.headers or {}), + } + if self.session_id: + headers["Mcp-Session-Id"] = self.session_id + response = await AsyncRequestUtils( + headers=headers, + timeout=_normalize_timeout(self.server.timeout), + content_type="application/json", + accept_type="application/json, text/event-stream", + http2=False, + ).post_res(self.server.url, json=payload, raise_exception=True) + try: + if not response: + raise RuntimeError("HTTP MCP 请求无响应") + response.raise_for_status() + session_id = response.headers.get("Mcp-Session-Id") + if session_id: + self.session_id = session_id + if not expect_response: + return None + content_type = response.headers.get("content-type", "").lower() + if "text/event-stream" in content_type: + return _parse_sse_text_response(response.text, request_id or "") + data = response.json() + return _extract_jsonrpc_result(data, request_id or "") + finally: + if response is not None: + await response.aclose() + + +class _SseMcpSession: + """旧版 HTTP+SSE MCP 会话。""" + + def __init__(self, server: AgentMcpServerConfig) -> None: + self.server = server + self.response = None + self.endpoint: Optional[str] = None + self._stream_manager = None + self._event_iterator = None + + async def __aenter__(self) -> "_SseMcpSession": + """打开 SSE 流并读取服务端回传的 POST endpoint。""" + if not self.server.url: + raise RuntimeError("SSE MCP 服务器缺少 URL") + self._stream_manager = AsyncRequestUtils( + headers={"Accept": "text/event-stream", **(self.server.headers or {})}, + timeout=_normalize_timeout(self.server.timeout), + accept_type="text/event-stream", + http2=False, + ).get_stream(self.server.url, raise_exception=True) + self.response = await self._stream_manager.__aenter__() + if not self.response: + raise RuntimeError("SSE MCP 连接无响应") + self.response.raise_for_status() + self._event_iterator = _iter_sse_events(self.response).__aiter__() + self.endpoint = await self._read_endpoint() + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + """关闭 SSE 流。""" + if self._stream_manager: + await self._stream_manager.__aexit__(exc_type, exc, tb) + + async def notify(self, method: str, params: Optional[dict[str, Any]] = None) -> None: + """发送不需要响应的 JSON-RPC 通知。""" + await self._post(_jsonrpc_message(method, params)) + + async def request(self, method: str, params: Optional[dict[str, Any]] = None) -> Any: + """发送 JSON-RPC 请求并等待 SSE 流上的响应。""" + request_id = uuid.uuid4().hex + await self._post(_jsonrpc_message(method, params, request_id=request_id)) + timeout = _normalize_timeout(self.server.timeout) + while True: + event = await asyncio.wait_for(self._event_iterator.__anext__(), timeout=timeout) + payload = _load_sse_json_payload(event.get("event", ""), event.get("data", "")) + if isinstance(payload, dict) and payload.get("id") == request_id: + return _extract_jsonrpc_result(payload, request_id) + + async def _read_endpoint(self) -> str: + """读取 SSE endpoint 事件中的 POST 地址。""" + timeout = _normalize_timeout(self.server.timeout) + while True: + event = await asyncio.wait_for(self._event_iterator.__anext__(), timeout=timeout) + if event.get("event") != "endpoint": + continue + endpoint = str(event.get("data") or "").strip() + if not endpoint: + continue + return urljoin(self.server.url, endpoint) + + async def _post(self, payload: dict) -> None: + """向 SSE 握手返回的 endpoint 发送 JSON-RPC 消息。""" + if not self.endpoint: + raise RuntimeError("SSE MCP endpoint 未初始化") + response = await AsyncRequestUtils( + headers={ + "Accept": "application/json", + "Content-Type": "application/json", + **(self.server.headers or {}), + }, + timeout=_normalize_timeout(self.server.timeout), + content_type="application/json", + accept_type="application/json", + http2=False, + ).post_res(self.endpoint, json=payload, raise_exception=True) + try: + if not response: + raise RuntimeError("SSE MCP POST 请求无响应") + response.raise_for_status() + finally: + if response is not None: + await response.aclose() + + +async def _open_mcp_session(server: AgentMcpServerConfig): + """根据配置创建对应的 MCP 传输会话。""" + transport = "http" if server.transport == "streamable_http" else server.transport + if transport == "stdio": + return _StdioMcpSession(server) + if transport == "sse": + return _SseMcpSession(server) + if transport == "http": + return _HttpMcpSession(server) + raise RuntimeError(f"不支持的 MCP 传输协议: {server.transport}") + + +class AgentMcpManager: + """管理 Agent 外部 MCP 服务器配置、工具发现和工具调用。""" + + def get_servers(self) -> list[AgentMcpServerConfig]: + """读取已保存的外部 MCP 服务器配置。""" + raw_servers = SystemConfigOper().get(SystemConfigKey.AIAgentMcpServers) or [] + if not isinstance(raw_servers, list): + return [] + servers: list[AgentMcpServerConfig] = [] + for raw_server in raw_servers: + try: + servers.append(self.normalize_server(raw_server)) + except Exception as err: + logger.warning(f"忽略无效的 Agent MCP 配置: {err}") + return servers + + async def save_servers(self, servers: list[AgentMcpServerConfig]) -> bool: + """保存外部 MCP 服务器配置。""" + normalized_servers = [self.normalize_server(server).model_dump() for server in servers] + return await SystemConfigOper().async_set( + SystemConfigKey.AIAgentMcpServers, + normalized_servers or None, + ) + + def normalize_server(self, value: Any) -> AgentMcpServerConfig: + """规范化单个 MCP 服务器配置。""" + if isinstance(value, AgentMcpServerConfig): + raw_server = value.model_dump() + elif isinstance(value, dict): + raw_server = dict(value) + else: + raise ValueError("MCP 服务器配置必须是对象") + + raw_server["id"] = str(raw_server.get("id") or uuid.uuid4().hex[:12]).strip() + raw_server["name"] = str(raw_server.get("name") or raw_server["id"]).strip() + raw_server["transport"] = str(raw_server.get("transport") or "stdio").strip() + raw_server["description"] = str(raw_server.get("description") or "").strip() or None + raw_server["command"] = str(raw_server.get("command") or "").strip() or None + raw_server["args"] = [str(item) for item in raw_server.get("args") or []] + raw_server["env"] = _normalize_string_dict(raw_server.get("env")) + raw_server["url"] = str(raw_server.get("url") or "").strip() or None + raw_server["headers"] = _normalize_string_dict(raw_server.get("headers")) + raw_server["timeout"] = _normalize_timeout(raw_server.get("timeout")) + raw_server["tool_prefix"] = str(raw_server.get("tool_prefix") or "").strip() or None + raw_server["require_admin"] = bool(raw_server.get("require_admin", True)) + return AgentMcpServerConfig.model_validate(raw_server) + + def config_signature(self) -> str: + """生成外部 MCP 配置签名，用于 Agent 图缓存失效。""" + payload = [server.model_dump() for server in self.get_servers()] + raw_text = json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str) + return hashlib.sha256(raw_text.encode("utf-8")).hexdigest() + + async def initialize_session(self, session) -> None: + """完成 MCP initialize 和 initialized 通知流程。""" + await session.request( + "initialize", + { + "protocolVersion": MCP_PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": { + "name": MCP_CLIENT_NAME, + "version": "1.0.0", + }, + }, + ) + await session.notify("notifications/initialized") + + async def list_server_tools(self, server: AgentMcpServerConfig) -> list[AgentMcpToolSpec]: + """连接单个 MCP 服务器并读取工具列表。""" + normalized_server = self.normalize_server(server) + session_manager = await _open_mcp_session(normalized_server) + async with session_manager as session: + await self.initialize_session(session) + result = await session.request("tools/list") + tools_payload = result.get("tools", []) if isinstance(result, dict) else [] + tool_specs: list[AgentMcpToolSpec] = [] + for item in tools_payload: + if not isinstance(item, dict) or not item.get("name"): + continue + tool_name = str(item["name"]) + tool_specs.append( + AgentMcpToolSpec( + server=normalized_server, + name=tool_name, + agent_tool_name=_build_agent_tool_name(normalized_server, tool_name), + description=str(item.get("description") or ""), + input_schema=_normalize_input_schema(item.get("inputSchema")), + ) + ) + return tool_specs + + async def list_enabled_tool_specs(self) -> list[AgentMcpToolSpec]: + """读取所有启用 MCP 服务器暴露的工具定义。""" + tool_specs: list[AgentMcpToolSpec] = [] + seen_names: set[str] = set() + for server in self.get_servers(): + if not server.enabled: + continue + try: + for spec in await self.list_server_tools(server): + if spec.agent_tool_name in seen_names: + logger.warning(f"跳过重复的 MCP Agent 工具名: {spec.agent_tool_name}") + continue + tool_specs.append(spec) + seen_names.add(spec.agent_tool_name) + except Exception as err: + logger.warning(f"读取 MCP 服务器 {server.name} 工具失败: {err}") + return tool_specs + + async def call_server_tool( + self, + server: AgentMcpServerConfig, + tool_name: str, + arguments: Optional[dict[str, Any]] = None, + ) -> Any: + """调用单个 MCP 服务器上的指定工具。""" + normalized_server = self.normalize_server(server) + session_manager = await _open_mcp_session(normalized_server) + async with session_manager as session: + await self.initialize_session(session) + return await session.request( + "tools/call", + { + "name": tool_name, + "arguments": arguments or {}, + }, + ) + + async def test_server(self, server: AgentMcpServerConfig) -> AgentMcpServerTestResult: + """测试 MCP 服务器连接并返回工具列表。""" + tool_specs = await self.list_server_tools(server) + tools = [ + AgentMcpServerToolInfo( + name=spec.name, + agent_tool_name=spec.agent_tool_name, + description=spec.description, + input_schema=spec.input_schema, + ) + for spec in tool_specs + ] + return AgentMcpServerTestResult( + success=True, + message=f"连接成功，发现 {len(tools)} 个工具", + tools=tools, + tool_count=len(tools), + ) + + +agent_mcp_manager = AgentMcpManager() diff --git a/app/agent/tools/impl/_system_setting_utils.py b/app/agent/tools/impl/_system_setting_utils.py index b9b131a1..38b138c7 100644 --- a/app/agent/tools/impl/_system_setting_utils.py +++ b/app/agent/tools/impl/_system_setting_utils.py @@ -59,6 +59,10 @@ SYSTEMCONFIG_SETTING_METADATA = { "group": "ai_agent", "label": "AI 智能体配置", }, + SystemConfigKey.AIAgentMcpServers.value: { + "group": "ai_agent", + "label": "AI 智能体外部 MCP 服务器", + }, SystemConfigKey.CustomIdentifiers.value: { "group": "custom_identifiers", "label": "自定义识别词", diff --git a/app/agent/tools/impl/mcp.py b/app/agent/tools/impl/mcp.py new file mode 100644 index 00000000..2fdf617a --- /dev/null +++ b/app/agent/tools/impl/mcp.py @@ -0,0 +1,98 @@ +"""外部 MCP 工具适配器。""" + +import json +from typing import Any, Optional + +from pydantic import PrivateAttr + +from app.agent.mcp import AgentMcpToolSpec, agent_mcp_manager +from app.agent.tools.base import MoviePilotTool +from app.agent.tools.tags import ToolTag + + +class McpExternalTool(MoviePilotTool): + """将外部 MCP 工具包装为 MoviePilot Agent 工具。""" + + name: str = "mcp_external_tool" + tags: list[str] = [ + ToolTag.Read, + ToolTag.Admin, + ] + description: str = "Call an external MCP tool configured for MoviePilot Agent." + args_schema: dict[str, Any] = {"type": "object", "properties": {}, "required": []} + require_admin: bool = True + + _spec: AgentMcpToolSpec = PrivateAttr() + + def __init__(self, spec: AgentMcpToolSpec, session_id: str, user_id: str) -> None: + 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) -> Optional[str]: + """根据 MCP 工具信息生成友好的提示消息。""" + return f"调用 MCP 工具: {self._spec.server.name}/{self._spec.name}" + + async def run(self, **kwargs) -> str: + """ + 调用外部 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): + 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 result.get("isError"): + return json.dumps(result, ensure_ascii=False, indent=2, default=str) + if isinstance(result, str): + return result + return json.dumps(result, ensure_ascii=False, indent=2, default=str) + + +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=None, + agent_context: Optional[dict] = None, +) -> list[McpExternalTool]: + """创建当前已启用的外部 MCP Agent 工具列表。""" + tools = [] + for spec in await agent_mcp_manager.list_enabled_tool_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) + tools.append(tool) + return tools diff --git a/app/api/endpoints/agent.py b/app/api/endpoints/agent.py index 74d2d328..5933764a 100644 --- a/app/api/endpoints/agent.py +++ b/app/api/endpoints/agent.py @@ -20,6 +20,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app import schemas from app.agent import MoviePilotAgent, ReplyMode, StreamingHandler, agent_manager from app.agent.llm.capability import AgentCapabilityManager +from app.agent.mcp import agent_mcp_manager from app.chain.message import MessageChain from app.chain.site import site_interaction_manager from app.chain.skills import skills_interaction_manager @@ -56,6 +57,78 @@ _WEB_AGENT_NOTICE_LISTENER_REGISTERED = False _WEB_AGENT_BACKGROUND_TASKS: set[asyncio.Task] = set() +def _ensure_superuser(user: User) -> None: + """校验当前用户是否为超级管理员。""" + if not getattr(user, "is_superuser", False): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Forbidden") + + +@router.get("/mcp/servers", summary="查询 Agent MCP 服务器配置", response_model=schemas.Response) +async def list_agent_mcp_servers( + current_user: User = Depends(get_current_active_user), +) -> schemas.Response: + """ + 查询 Agent 外部 MCP 服务器配置。 + """ + _ensure_superuser(current_user) + servers = agent_mcp_manager.get_servers() + enabled_count = len([server for server in servers if server.enabled]) + return schemas.Response( + success=True, + data={ + "servers": [server.model_dump() for server in servers], + "enabled_count": enabled_count, + "total_count": len(servers), + }, + ) + + +@router.post("/mcp/servers", summary="保存 Agent MCP 服务器配置", response_model=schemas.Response) +async def save_agent_mcp_servers( + request: schemas.AgentMcpServersSaveRequest, + current_user: User = Depends(get_current_active_user), +) -> schemas.Response: + """ + 保存 Agent 外部 MCP 服务器配置。 + """ + _ensure_superuser(current_user) + success = await agent_mcp_manager.save_servers(request.servers) + return schemas.Response( + success=success, + message="保存MCP配置成功" if success else "保存MCP配置失败", + ) + + +@router.post("/mcp/servers/test", summary="测试 Agent MCP 服务器", response_model=schemas.Response) +async def test_agent_mcp_server( + request: schemas.AgentMcpServerTestRequest, + current_user: User = Depends(get_current_active_user), +) -> schemas.Response: + """ + 测试 Agent 外部 MCP 服务器连接并读取工具列表。 + """ + _ensure_superuser(current_user) + try: + result = await agent_mcp_manager.test_server(request.server) + return schemas.Response( + success=result.success, + message=result.message, + data=result.model_dump(), + ) + except Exception as err: + logger.warning(f"测试 Agent MCP 服务器失败: {err}") + return schemas.Response( + success=False, + message=f"测试MCP服务器失败: {str(err)}", + data={ + "success": False, + "message": str(err), + "tools": [], + "tool_count": 0, + }, + ) + + class _WebAgentStreamingHandler(StreamingHandler): """ Web 前端专用流式处理器，将工具提示和文本统一回调给 SSE。 diff --git a/app/schemas/agent.py b/app/schemas/agent.py index 9e3b24dc..f756506b 100644 --- a/app/schemas/agent.py +++ b/app/schemas/agent.py @@ -1,7 +1,7 @@ """AI智能体相关数据模型""" from datetime import datetime -from typing import List, Optional, Union +from typing import Any, List, Literal, Optional, Union from langchain_core.messages import BaseMessage from pydantic import BaseModel, Field, ConfigDict, field_serializer @@ -57,6 +57,58 @@ class ToolResult(BaseModel): error: Optional[str] = Field(default=None, description="错误信息") +class AgentMcpServerConfig(BaseModel): + """Agent 外部 MCP 服务器配置。""" + + id: str = Field(..., description="服务器唯一 ID") + name: str = Field(..., description="服务器显示名称") + enabled: bool = Field(default=True, description="是否启用") + transport: Literal["stdio", "sse", "http", "streamable_http"] = Field( + default="stdio", description="MCP 传输协议" + ) + description: Optional[str] = Field(None, description="服务器说明") + command: Optional[str] = Field(None, description="stdio 启动命令") + args: list[str] = Field(default_factory=list, description="stdio 启动参数") + env: dict[str, str] = Field(default_factory=dict, description="stdio 环境变量") + url: Optional[str] = Field(None, description="HTTP/SSE MCP 入口地址") + headers: dict[str, str] = Field(default_factory=dict, description="HTTP 请求头") + timeout: int = Field(default=30, description="连接和调用超时时间（秒）") + tool_prefix: Optional[str] = Field(None, description="注入 Agent 的工具名前缀") + require_admin: bool = Field(default=True, description="是否仅管理员可调用") + + +class AgentMcpServersSaveRequest(BaseModel): + """Agent 外部 MCP 服务器保存请求。""" + + servers: list[AgentMcpServerConfig] = Field( + default_factory=list, description="MCP 服务器配置列表" + ) + + +class AgentMcpServerTestRequest(BaseModel): + """Agent 外部 MCP 服务器测试请求。""" + + server: AgentMcpServerConfig = Field(..., description="待测试的 MCP 服务器配置") + + +class AgentMcpServerToolInfo(BaseModel): + """Agent 外部 MCP 工具摘要。""" + + name: str = Field(..., description="原始 MCP 工具名称") + agent_tool_name: str = Field(..., description="注入 Agent 后的工具名称") + description: str = Field(default="", description="工具说明") + input_schema: dict[str, Any] = Field(default_factory=dict, description="工具参数 Schema") + + +class AgentMcpServerTestResult(BaseModel): + """Agent 外部 MCP 服务器测试结果。""" + + success: bool = Field(..., description="测试是否成功") + message: str = Field(default="", description="测试消息") + tools: list[AgentMcpServerToolInfo] = Field(default_factory=list, description="工具列表") + tool_count: int = Field(default=0, description="工具数量") + + class AgentChatAttachment(BaseModel): """ Agent 会话展示附件。 diff --git a/app/schemas/types.py b/app/schemas/types.py index 0d2e1912..6c50043e 100644 --- a/app/schemas/types.py +++ b/app/schemas/types.py @@ -265,6 +265,8 @@ class SystemConfigKey(Enum): NotificationSendTime = "NotificationSendTime" # AI智能体配置 AIAgentConfig = "AIAgentConfig" + # AI智能体外部MCP服务器配置 + AIAgentMcpServers = "AIAgentMcpServers" # 通知消息格式模板 NotificationTemplates = "NotificationTemplates" # 通知中心清理时间 diff --git a/docs/mcp-api.md b/docs/mcp-api.md index ec04893b..7e32cad0 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -67,6 +67,26 @@ MCP 使用系统配置中的 `API_TOKEN` 作为认证密钥，文档中的 API K } ``` +## 4.1 Agent 外部 MCP Client 配置 + +MoviePilot 的内置 Agent 也可以作为 MCP Client 连接外部 MCP 服务器，将外部工具注入到智能助手工具列表中。当前支持： + +- `stdio`：按配置的命令和参数启动本地 MCP 进程，通过标准输入输出交换 JSON-RPC 消息。 +- `sse`：连接旧版 HTTP+SSE MCP 服务，先读取 `endpoint` 事件，再向返回的 endpoint POST JSON-RPC 消息。 +- `http` / `streamable_http`：连接 Streamable HTTP MCP 服务，直接向配置 URL POST JSON-RPC 消息。 + +这些配置是管理员级 Agent 运行时配置，保存在 `SystemConfigKey.AIAgentMcpServers` 中。外部 MCP 工具默认要求管理员上下文调用，避免普通用户触发高权限外部工具。 + +### Agent MCP 配置接口 + +这些接口使用登录态鉴权，并要求当前用户为超级管理员。 + +| 方法 | 路径 | 说明 | +| :--- | :--- | :--- | +| GET | `/api/v1/message/agent/mcp/servers` | 查询已配置的外部 MCP 服务器列表 | +| POST | `/api/v1/message/agent/mcp/servers` | 保存外部 MCP 服务器列表 | +| POST | `/api/v1/message/agent/mcp/servers/test` | 测试单个外部 MCP 服务器，返回发现的工具列表 | + ## 5. 错误码说明 | 错误码 | 消息 | 说明 | @@ -133,6 +153,9 @@ FastAPI 异常响应保留 `detail` 字段，并在错误详情为文本时返 | POST | `/api/v1/system/setting/PLUGIN_MARKET/sync-wiki` | 管理员从 MoviePilot Wiki 的插件文档同步公开插件仓库清单，和本地 `PLUGIN_MARKET` 合并去重后写入配置 | | GET | `/api/v1/system/modulelist` | 查询已加载模块，保留 `name` 原始中文字段，并提供 `name_i18n` 和 `name_key` 给多语言前端展示 | | GET | `/api/v1/system/moduletest/{moduleid}` | 测试指定模块可用性，保留原 `message`，并在标准响应顶层返回 `message_i18n` | +| GET | `/api/v1/message/agent/mcp/servers` | 管理员查询 Agent 外部 MCP 服务器配置 | +| POST | `/api/v1/message/agent/mcp/servers` | 管理员保存 Agent 外部 MCP 服务器配置 | +| POST | `/api/v1/message/agent/mcp/servers/test` | 管理员测试单个 Agent 外部 MCP 服务器并读取工具列表 | ### 插件补充接口 diff --git a/skills/moviepilot-api/SKILL.md b/skills/moviepilot-api/SKILL.md index 7fcb9963..d7106d4c 100644 --- a/skills/moviepilot-api/SKILL.md +++ b/skills/moviepilot-api/SKILL.md @@ -470,6 +470,14 @@ All endpoints are under the base URL `{MP_HOST}`. Path parameters are shown as ` | GET | `/api/v1/mcp/tools/{tool_name}` | Get tool definition | | GET | `/api/v1/mcp/tools/{tool_name}/schema` | Get tool input schema | +### Agent MCP Client (3 endpoints) + +| Method | Path | Description | +|--------|------|-------------| +| GET | `/api/v1/message/agent/mcp/servers` | List external MCP servers configured for the built-in Agent. Superuser login required | +| POST | `/api/v1/message/agent/mcp/servers` | Save external MCP servers for the built-in Agent. Body: `{"servers":[...]}` | +| POST | `/api/v1/message/agent/mcp/servers/test` | Test one external MCP server and return discovered tools. Body: `{"server":{...}}` | + ### Webhook (2 endpoints) | Method | Path | Description | diff --git a/tests/test_agent_mcp.py b/tests/test_agent_mcp.py new file mode 100644 index 00000000..1c6b6c9c --- /dev/null +++ b/tests/test_agent_mcp.py @@ -0,0 +1,143 @@ +import sys +import textwrap + +import pytest + +from app.agent.mcp import AgentMcpManager, AgentMcpToolSpec +from app.agent.tools.impl.mcp import McpExternalTool +from app.schemas.agent import AgentMcpServerConfig + + +def _write_stdio_mcp_server(tmp_path): + """写入一个用于测试的最小 stdio MCP 服务。""" + server_path = tmp_path / "stdio_mcp_server.py" + server_path.write_text( + textwrap.dedent( + """ + import json + import sys + + TOOLS = [ + { + "name": "echo", + "description": "Echo input text.", + "inputSchema": { + "type": "object", + "properties": { + "text": {"type": "string", "description": "Text to echo"} + }, + "required": ["text"], + }, + } + ] + + for line in sys.stdin: + request = json.loads(line) + request_id = request.get("id") + method = request.get("method") + if request_id is None: + continue + if method == "initialize": + result = { + "protocolVersion": request["params"]["protocolVersion"], + "capabilities": {"tools": {}}, + "serverInfo": {"name": "Fake MCP", "version": "1.0.0"}, + } + elif method == "tools/list": + result = {"tools": TOOLS} + elif method == "tools/call": + args = request.get("params", {}).get("arguments", {}) + result = {"content": [{"type": "text", "text": args.get("text", "")}]} + else: + result = {} + print(json.dumps({"jsonrpc": "2.0", "id": request_id, "result": result}), flush=True) + """ + ), + encoding="utf-8", + ) + return server_path + + +@pytest.mark.anyio +async def test_stdio_mcp_server_lists_tools(tmp_path): + """stdio MCP 服务器应能被初始化并读取工具列表。""" + server_path = _write_stdio_mcp_server(tmp_path) + manager = AgentMcpManager() + server = AgentMcpServerConfig( + id="fake", + name="Fake MCP", + transport="stdio", + command=sys.executable, + args=[str(server_path)], + timeout=5, + ) + + tools = await manager.list_server_tools(server) + + assert len(tools) == 1 + assert tools[0].name == "echo" + assert tools[0].agent_tool_name == "mcp_fake_mcp_echo" + assert tools[0].input_schema["properties"]["text"]["type"] == "string" + + +@pytest.mark.anyio +async def test_stdio_mcp_server_calls_tool(tmp_path): + """stdio MCP 工具应能通过 tools/call 返回内容。""" + server_path = _write_stdio_mcp_server(tmp_path) + manager = AgentMcpManager() + server = AgentMcpServerConfig( + id="fake", + name="Fake MCP", + transport="stdio", + command=sys.executable, + args=[str(server_path)], + timeout=5, + ) + + result = await manager.call_server_tool(server, "echo", {"text": "hello"}) + + assert result == {"content": [{"type": "text", "text": "hello"}]} + + +def test_normalize_server_generates_runtime_defaults(): + """MCP 配置规范化应补齐默认值并清理空字段。""" + manager = AgentMcpManager() + + server = manager.normalize_server( + { + "id": "demo", + "name": "Demo", + "transport": "http", + "url": " https://example.com/mcp ", + "headers": {" Authorization ": "Bearer token", "": "ignored"}, + "timeout": "bad", + } + ) + + assert server.id == "demo" + assert server.url == "https://example.com/mcp" + assert server.headers == {"Authorization": "Bearer token"} + assert server.timeout == 30 + assert server.require_admin is True + + +def test_mcp_external_tool_uses_discovered_schema(): + """外部 MCP 工具应保留发现到的 JSON Schema。""" + server = AgentMcpServerConfig(id="fake", name="Fake MCP") + spec = AgentMcpToolSpec( + server=server, + name="echo", + agent_tool_name="mcp_fake_echo", + description="Echo input text.", + input_schema={ + "type": "object", + "properties": {"text": {"type": "string"}}, + "required": ["text"], + }, + ) + + tool = McpExternalTool(spec=spec, session_id="session-1", user_id="10001") + + assert tool.name == "mcp_fake_echo" + assert tool.args_schema["properties"]["text"]["type"] == "string" + assert tool.require_admin is True