feat: add agent MCP support

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

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()

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": "自定义识别词",

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

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。

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 会话展示附件。

app/schemas/types.py

@@ -265,6 +265,8 @@ class SystemConfigKey(Enum):
     NotificationSendTime = "NotificationSendTime"
     # AI智能体配置
     AIAgentConfig = "AIAgentConfig"
+    # AI智能体外部MCP服务器配置
+    AIAgentMcpServers = "AIAgentMcpServers"
     # 通知消息格式模板
     NotificationTemplates = "NotificationTemplates"
     # 通知中心清理时间

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 服务器并读取工具列表 |
 
 ### 插件补充接口
 

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 |

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