Mcp cn

Jianmu 的 MCP（Model Context Protocol）模块位于 jianmu/mcp/，提供了一套完整的三层架构来桥接外部 MCP 服务器与 Jianmu 的工具与技能生态。该模块以 FastMCP v2 为底层传输抽象，向上暴露客户端连接、工具提供者生命周期管理以及服务端构建能力，使 Jianmu Agent 能够无缝调用运行在独立进程或远程 HTTP/SSE 端点中的工具、资源和提示词。

三层架构总览

MCP 模块的核心抽象分为三个层次，每一层关注不同的集成关注点：

flowchart TB
    subgraph "构建层 — jianmu/mcp/server.py"
        MCPServer["MCPServer<br/>FastMCP 包装"]
        Builder["MCPServerBuilder<br/>链式构造器"]
        Builder -->|"build()"| MCPServer
    end

    subgraph "连接层 — jianmu/mcp/client.py"
        Config["MCPServerConfig<br/>子进程配置"]
        Client["MCPClient<br/>多传输客户端"]
        MCPTool["MCPTool<br/>远程工具包装"]
        ResTool["MCPResourceTool<br/>远程资源包装"]
        Config -->|"StdioTransport"| Client
        Client -->|"as_tools()"| MCPTool
        Client -->|"as_resource_tools()"| ResTool
    end

    subgraph "集成层 — jianmu/mcp/provider.py"
        Provider["MCPToolProvider<br/>基于 mcp.json 的工具提供"]
        Provider -->|"管理"| Client
    end

    subgraph "Jianmu 运行时"
        Swarm["SwarmRuntime"]
        ToolSet["ToolSet"]
        Agent["Agent"]
    end

    Swarm -->|"加载"| Provider
    ToolSet -->|"from_providers()"| Provider
    Agent -->|"execute()"| MCPTool
    Agent -->|"execute()"| ResTool

    style MCPServer fill:#e1f5fe
    style Client fill:#fff3e0
    style Provider fill:#e8f5e9

构建层 (server.py) 提供 MCPServer 和 MCPServerBuilder，用于将 Python 函数注册为 MCP 工具、资源和提示词并启动服务器。连接层 (client.py) 是核心——MCPClient 通过六种传输归一化策略连接到任意 MCP 端点，MCPTool 和 MCPResourceTool 将远端能力包装为 Jianmu 标准的 Tool 子类。集成层 (provider.py) 的 MCPToolProvider 从 mcp.json 配置文件读取服务器定义，管理客户端生命周期并暴露工具集合，可被 SwarmRuntime 或 ToolSet 直接消费。

MCPClient：多传输、多源归一化

MCPClient 是整个 MCP 模块的中枢，负责将多样化的服务器描述归一化为 FastMCP 传输对象，并提供惰性连接、双向上下文管理器支持和完整的 MCP 协议操作封装。

服务器源归一化策略

MCPClient.__init__ 接受五种形态的 server_source，内部通过 _prepare_server_source() 方法将其统一映射到 FastMCP 传输对象。下表总结了每种输入形态与对应的传输类型：

输入形态	示例值	解析结果	额外参数控制
MCPServerConfig	MCPServerConfig(command="node", args=["server.js"])	StdioTransport	env 合并自 config 或 constructor
Python 脚本路径 (.py 后缀)	"/path/to/server.py"	PythonStdioTransport	server_args, env, transport_kwargs
命令列表 (首元素为 python + .py 脚本)	["python", "server.py"]	PythonStdioTransport	server_args[2:] 追加
命令列表 (其他)	["node", "server.js"] / ["npx", "-y", "@scope/pkg"]	StdioTransport	server_args[1:] 追加
HTTP/HTTPS URL	"http://localhost:8000"	StreamableHttpTransport (默认) 或 SSETransport	transport_type 可强制 "sse"
FastMCP 实例	FastMCP("name")	直接使用，不做变换	—
配置字典	{"transport": "stdio", "command": "python", ...}	根据 transport 字段分发	transport_kwargs 合并

_prepare_server_source() 的分支逻辑按优先级逐个匹配。对于 MCPServerConfig，优先使用构造函数传入的 env 覆盖配置中的 env；对于命令列表形态，自动检测 python + .py 组合以选择 PythonStdioTransport。配置字典形态则根据 transport 键值（"stdio"、"sse"、"http"）分别构造对应传输对象，并支持 headers、auth、cwd 等扩展字段。

连接生命周期管理

MCPClient 实现了异步上下文管理器协议（__aenter__ → connect() / __aexit__ → close()），同时支持显式调用 connect() 和 close()。connect() 的核心策略是幂等——如果 self.client 已非空，直接返回缓存实例，避免重复连接。当连接失败时（Client.__aenter__ 抛出异常），会尝试调用 __aexit__ 清理资源，并将 self.client 和 self._context_manager 保持为 None，确保不会留下脏状态。

close() 方法对子进程传输做了特殊处理：在调用 FastMCP 上下文管理器的 __aexit__ 后，检查底层 transport 对象是否存在 close 方法，若存在则调用并等待 0.1 秒让子进程优雅退出，以此避免 "Event loop is closed" 的垃圾回收错误。

协议操作集

连接建立后，MCPClient 暴露六类 MCP 协议操作，每类操作内部先调用 connect() 确保传输就绪：

方法	MCP 协议语义	返回值处理
list_tools()	列出服务器工具	优先取 response.tools，回退到列表格式
list_resources()	列出服务器资源	优先取 response.resources，回退到列表格式
list_prompts()	列出服务器提示词	优先取 response.prompts，回退到列表格式
call_tool(name, arguments)	调用指定工具	兼容 call_tool_mcp（FastMCP v2 新 API）与 call_tool
read_resource(uri)	读取指定资源	直接委托 client.read_resource(uri)
get_prompt(name, arguments)	获取提示词内容	直接委托 client.get_prompt(name, arguments)
ping()	连通性探测	返回 True/False，不抛异常

工具与资源的批量包装

as_tools(allowlist) 和 as_resource_tools(allowlist) 是连接 MCP 工具与 Jianmu 生态的关键桥梁。as_tools() 先调用 list_tools() 获取全部工具定义，若指定 allowlist 则按名称（忽略大小写）过滤，然后将每个工具定义包装为 MCPTool 实例返回。as_resource_tools() 采用相同模式，将资源定义包装为 MCPResourceTool。

sequenceDiagram
    participant User
    participant MCPClient
    participant FastMCP Client
    participant MCP Server

    User->>MCPClient: as_tools(allowlist=["tool_a"])
    MCPClient->>MCPClient: connect() [幂等]
    MCPClient->>FastMCP Client: list_tools()
    FastMCP Client->>MCP Server: list_tools request
    MCP Server-->>FastMCP Client: [ToolDef, ToolDef, ...]
    FastMCP Client-->>MCPClient: response.tools
    MCPClient->>MCPClient: filter by allowlist<br/>wrap each as MCPTool(client, tool_def)
    MCPClient-->>User: List[MCPTool]

MCPTool 与 MCPResourceTool：远程能力本地化

这两个类是 MCP 远程工具与 Jianmu Tool 基类之间的适配层。它们继承 Tool（定义于 jianmu/tool/base.py），使得 MCP 工具可以直接被 ToolExecutor 节点调用、被 LLM 作为 function-calling schema 消费，或通过 ToolSet 统一管理。

MCPTool：Schema 清洗与结果归一化

MCPTool.__init__ 从原始 MCP 工具定义中提取 name、description、inputSchema、outputSchema 字段。关键的适配工作体现在两处：

Schema 清洗：OpenAI 兼容 API（以及部分其他 LLM 后端）拒绝 JSON Schema 元数据字段 $schema 和 additionalProperties。_clean_schema() 静态方法执行浅拷贝后移除这两个字段，确保生成的 function-calling schema 可被模型正确消费，避免 400 错误。

@staticmethod
def _clean_schema(schema: Any) -> Dict[str, Any]:
    if not isinstance(schema, dict):
        return {"type": "object"}
    cleaned = dict(schema)
    cleaned.pop("$schema", None)
    cleaned.pop("additionalProperties", None)
    return cleaned

输入参数智能映射：run() 方法实现了三种输入传递策略——（1）若传入 **kwargs，直接用作工具参数；（2）若 input 是字典，原样传递；（3）若 input 是非字典的单值且 inputSchema 仅定义了一个属性，自动将该值映射到该属性的键下。这种智能映射消除了调用方需要了解 MCP 工具内部参数结构的负担。

flowchart LR
    A["run(input='hello', kwargs={})"] --> B{"kwargs 非空?"}
    B -->|是| C["使用 kwargs 作为参数"]
    B -->|否| D{"input 是 dict?"}
    D -->|是| E["原样传递 input"]
    D -->|否| F{"inputSchema 仅有一个属性?"}
    F -->|是 key| G["映射为 {key: input}"]
    F -->|否| H["包装为 {'input': input}"]

结果归一化：执行结果的处理遵循优先级链——先检查 structuredContent（驼峰与下划线两种命名），再回退到 content 列表（逐个提取 text、data、blob），最后 str(result)。若结果标记了 isError，则提取错误文本返回。

MCPResourceTool：只读资源包装

MCPResourceTool 将 MCP 资源建模为只读 Tool。其 input_schema 固定为 {"type": "string"}，接受可选的 URI 覆盖参数。run() 方法优先使用传入的 URI 覆盖资源定义的默认 uri，然后调用 client.read_resource(uri) 并依次从 contents 列表中提取 text、data、blob 内容。

与 Tool 基类的协议兼容

MCPTool 和 MCPResourceTool 都重写了 run() 为异步方法，因此 Tool.execute() 的 _execute_local() 路径可以正确检测到协程并通过 _call_single 或 _call_kwargs await。它们同样继承了 as_node()（可包装为行为树节点）、to_schema()（生成 LLM function-calling 定义）等能力。

MCPToolProvider：配置文件驱动的生命周期管理

MCPToolProvider 是 MCP 模块与 Jianmu Tool Provider 协议之间的集成点。它实现了 BaseToolProvider 接口（initialize() → get_tools() → close()），使 MCP 工具可以像内建工具或自定义工具一样被 ToolSet.from_providers() 或 SwarmRuntime 加载。

mcp.json 配置格式

MCPToolProvider 从 mcp.json（路径可通过 config_path 参数自定义）读取服务器定义。配置文件遵循标准 MCP 配置格式：

{
  "mcpServers": {
    "memory": {
      "transport": "stdio",
      "command": "npx",
      "args": ["-y", "@anthropic-ai/mcp-server-memory"]
    },
    "weather": {
      "transport": "http",
      "url": "http://localhost:8000",
      "headers": {"Authorization": "Bearer token"}
    }
  }
}

每个 mcpServers 条目包含 transport（"stdio" / "http" / "sse"）以及对应传输类型所需的参数。Provider 内部为每个条目创建一个 MCPClient 实例并托管其生命周期。

初始化流程

initialize() 方法按以下顺序执行：加载 mcp.json → 提取 mcpServers 字典 → 遍历每个服务器条目 → 创建 MCPClient → 调用 connect() → list_tools() → 为每个工具调用 _build_tool_wrapper() → 将包装后的工具追加到 self._tools 列表。单服务器连接失败仅记录警告，不会阻止其他服务器初始化。

_build_tool_wrapper() 是核心工厂方法：它用 type() 动态创建一个匿名 Tool 子类（_MCPWrappedTool），其 run() 方法调用 client.call_tool(tool_name, arguments=kwargs) 并通过 _normalize_mcp_result() 统一结果格式。工具命名遵循 mcp_{server_name}_{tool_name} 约定，确保来自不同 MCP 服务器的同名工具不会冲突。

sequenceDiagram
    participant RT as SwarmRuntime
    participant P as MCPToolProvider
    participant C as MCPClient
    participant S as MCP Server

    RT->>P: initialize()
    P->>P: _load_config() → mcp.json
    loop 每个 mcpServers 条目
        P->>C: MCPClient(server_config)
        P->>C: connect()
        C->>S: 建立传输
        S-->>C: 连接成功
        P->>C: list_tools()
        S-->>C: [ToolDef, ...]
        C-->>P: 工具定义列表
        P->>P: _build_tool_wrapper(server_name, client, tool_def)
    end
    P-->>RT: 就绪

    RT->>P: get_tools()
    P-->>RT: List[Tool] (含 MCP 包装工具)

    RT->>P: close()
    P->>C: close() [每个 client]
    C->>S: 断开连接
    P->>P: 清空 _clients, _tools

结果归一化函数

_normalize_mcp_result() 和 _normalize_schema() 是两个模块级工具函数。前者处理 MCP 返回结果的结构差异——优先提取 structuredContent，回退到拼接 content 列表中各块的 text/data/blob。后者清理 JSON Schema 中的 $schema 元数据，与 MCPTool._clean_schema() 形成互补。

在 SwarmRuntime 中的集成

MCPToolProvider 在 Jianmu 运行时中的主要消费方是 SwarmRuntime（位于 jianmu/swarm/runtime/core.py）。在运行时初始化阶段，工具提供者按以下优先级合并：

1. SwarmToolProvider（swarm 专用的动态工具源）
2. 若指定了 mcp_config_path，加载 MCPToolProvider
3. 用户自定义的 tool_providers（构造函数参数）

合并采用 last-wins by tool name 语义——后加载的提供者可以覆盖先前的同名工具。这意味着 mcp.json 中的 MCP 工具默认优先级高于 swarm 内建工具，但用户自定义提供者拥有最高优先权。

providers: list[ToolProvider] = [SwarmToolProvider(self)]
if mcp_config_path:
    providers.append(MCPToolProvider(config_path=mcp_config_path))
providers.extend(tool_providers or [])

MCPServer 与 MCPServerBuilder：构建 MCP 端点

当 Jianmu 需要对外暴露 MCP 能力（而非消费外部 MCP 服务）时，MCPServer 和 MCPServerBuilder 提供了轻量级的 FastMCP 封装。

MCPServer：薄封装

MCPServer 在构造时创建 FastMCP(name) 实例，并通过 add_tool()、add_resource()、add_prompt() 三个方法注册 Python 可调用对象。run(transport="stdio", **kwargs) 委托给 FastMCP 的启动逻辑，支持 stdio 和 sse 两种传输模式。

server = MCPServer("my-server", description="我的 MCP 服务")
server.add_tool(lambda a, b: a + b, name="add", description="两数相加")
server.add_prompt(greet_fn, name="greet")
server.run(transport="stdio")

MCPServerBuilder：链式构造

MCPServerBuilder 提供流式 API，每个 with_tool() / with_resource() / with_prompt() 返回 self，最后调用 build() 获取 MCPServer 实例，或直接 run() 启动。

server = (MCPServerBuilder("demo")
    .with_tool(add_fn, name="add")
    .with_resource(read_fn, uri="file://data")
    .build())

两种构造方式的自动检测依赖 fastmcp 包（pip install fastmcp>=2.0.0），如果未安装会在首次调用时抛出明确的 RuntimeError。

完整示例：从 MCP 服务器到 Agent 调用

以下示例来自 examples/mcp/mcp_demo.py，展示了 MCP 集成的完整端到端流程——创建一个 ReAct Agent，连接基于 stdio 子进程的 MCP Memory Server，并通过标准 Jianmu 工具调用模式进行实体存储和知识图谱查询。

flowchart LR
    subgraph "环境准备"
        A1["pip install fastmcp>=2.0.0"]
        A2["npm install -g @anthropic-ai/mcp-server-memory"]
    end

    subgraph "连接建立"
        B1["MCPClient(server_source=['npx', '-y', '@modelcontextprotocol/server-memory'])"]
        B2["as_tools() → List[MCPTool]"]
        B1 --> B2
    end

    subgraph "Agent 组装"
        C1["ModelClient.resolve(preference=['openai'])"]
        C2["create_react_node(model_client, tools=mcp_tools, config=ReActConfig(...))"]
        C3["Agent(root, state_schema=ReActState)"]
        C1 --> C2 --> C3
    end

    subgraph "执行"
        D1["agent.run(input_data={...})"]
        D2["LLM → function-calling → MCPTool.run() → MCPClient.call_tool()"]
        D3["agent.state.final_answer"]
        D1 --> D2 --> D3
    end

    A1 --> B1
    A2 --> B1

关键代码路径：MCPClient 将 ["npx", "-y", "@modelcontextprotocol/server-memory"] 归一化为 StdioTransport；as_tools() 返回的 MCPTool 列表被直接传入 create_react_node() 的 tools 参数；ReAct 节点在 LLM 返回 function-calling 指令后，通过 ToolExecutor 调用 MCPTool.run()，最终委托 MCPClient.call_tool() 将请求通过子进程 stdin/stdout 发送到 MCP 服务器。

传输类型决策指南

选择哪种传输类型取决于 MCP 服务器的部署形态：

场景	传输类型	MCPClient 输入示例	适用条件
本地子进程（任意命令）	StdioTransport	["npx", "-y", "pkg"] 或 MCPServerConfig(command="node", ...)	服务器在本地可执行
本地 Python 脚本	PythonStdioTransport	"/path/to/server.py" 或 ["python", "server.py"]	需要在当前 Python 进程中管理子进程
远程 HTTP 服务	StreamableHttpTransport	"http://host:port"	服务器在远程或容器化部署
远程 SSE 流	SSETransport	"http://host:port" + transport_type="sse"	需要服务器推送事件
内存内 FastMCP 实例	直接传递	FastMCP("name")	测试或嵌入式场景

与其他模块的关联

MCP 集成位于 Jianmu 扩展与集成域中，与以下模块形成紧密协作：

• 工具抽象：MCPTool 继承 Tool，可被 ToolSet 管理、ToolExecutor 调用
• Swarm 运行时：MCPToolProvider 通过 mcp_config_path 参数注入
• 上下文构建器：MCP 工具的 to_schema() 生成的 JSON Schema 被拼入 LLM 上下文
• Guard 体系：MCP 工具调用同样受策略检查、预算控制和频率限制约束