agent

《AI Agent 工程实战》10.1 MCP 客户端接入：minagent 作为 client 连本地 Server

本节把 MCP Server 当成工具源：写一个同步外壳的 MCPToolSource，连上本地 stdio Server、list_tools，把每个远程工具适配成 minagent 的 Tool 注册进 ToolRegistry，并讲清命名冲突、错误映射与连接生命周期；最后用假模型驱动 Agent 真跑一次 MCP 调用（Python 3.14.6 + mcp 2.3.0 实测）。

Leeting Yan 8 分钟阅读 3796 字

10.1 MCP 客户端接入：minagent 作为 client 连本地 Server

卷一的 minagent 里，工具是我们自己在 Python 进程内注册的：写个函数、包一个 Tool、丢进 ToolRegistry。这条路对自家代码够用，但 Agent 的能力边界一旦要往外扩，就会撞上「每个工具都得自己实现一遍」的墙——文件系统、Git、数据库、第三方 SaaS，难道都重写？MCP（Model Context Protocol）就是这道墙的缺口：它把「工具」从进程内的函数，变成任何实现了该协议的 Server 都能提供的远程能力。

本节把 ResearchCopilot 推进到「工具源可插拔」：新增 minagent/mcp_source.py，让 minagent 作为 MCP client 连上外部 Server，把对方暴露的工具适配成自己的 Tool。协议规范、Server 实现、传输、OAuth 不在本书范围，需要时直接看站内 MCP 专题 。

10.1.1 为什么是「客户端」，不是「再写一套工具」

先划清分工。MCP 世界里有两个角色：

角色谁扮演职责
Server外部进程暴露 tools / resources / prompts
Client我们的 Agent连接 Server、发现工具、发起调用

本节的落点是 client 侧：minagent 只负责「连上去、问它有哪些工具、把它们当自己的工具用」，至于 Server 内部怎么实现工具、用什么传输、怎么鉴权，一概不管。这带来一个关键的架构收益——Agent 的能力边界 = 已连接 Server 的工具并集，加一个 Server 就多一批能力，不改 Agent 一行代码。

这正好接上卷一的结论：ToolRegistry 只认 Tool（name / description / parameters / func）。只要能把远程工具翻译成这个形状，引擎完全感知不到工具的来路。所以本节的活本质是写一个适配器。

10.1.2 本地夹具 Server：只是测试夹具

演示要有个 Server 可连。我们用官方 SDK 自带的服务端辅助 API 起一个最小的本地 stdio Server。这里只是夹具——MCP Server 本身怎么写不在本书范围，站内 MCP 专题 有 44 篇专门讲它。

# fixture_server.py —— 本地夹具，仅用于演示，不是本书的教学重点
from mcp.server.mcpserver import MCPServer

mcp = MCPServer("research-fixture")

_CORPUS = {
    "mcp": "MCP 让 Agent 通过统一协议接入外部工具与资源。",
    "cache": "语义缓存按向量相似度命中历史回答，可省下重复推理。",
}

@mcp.tool()
def echo(text: str) -> str:
    """原样返回输入文本，用于连通性检查。"""
    return text

@mcp.tool()
def search_corpus(query: str, limit: int = 3) -> list[str]:
    """在本地语料里按关键词检索片段，返回最多 limit 条。"""
    hits = [v for k, v in _CORPUS.items() if query in k or query in v]
    return hits[:limit]

@mcp.tool()
def word_count(text: str) -> int:
    """统计文本的字符数。"""
    return len(text)

if __name__ == "__main__":
    mcp.run("stdio")

这里踩到第一个实测坑：mcp 2.3.0 把 FastMCP 改名成了 MCPServer。网上大量教程还写着 from mcp.server.fastmcp import FastMCP，在 2.x 下会直接 ModuleNotFoundError。这不是笔误，是 SDK 的大版本迁移，本机实测确认。

10.1.3 MCPToolSource：连接、发现、适配

客户端侧我们实现一个 MCPToolSource。它的形状很像卷一的 ToolRegistry，但多了一件难事：MCP 的官方客户端 API 是异步的，而 minagent 的 ToolRegistry.call 是同步的。这个「同步外壳套异步内核」的桥，是本节最值得看的部分。

# minagent/mcp_source.py
from __future__ import annotations
import asyncio, json, threading
from typing import Any

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from minagent.tools import Tool, ToolRegistry


class MCPToolSource:
    """把一个 MCP Server 暴露的工具适配成 minagent 的 Tool（同步外壳）。"""

    def __init__(self, name: str, command: str, args: list[str]) -> None:
        self.name = name
        self.command = command
        self.args = args
        self._loop: asyncio.AbstractEventLoop | None = None
        self._thread: threading.Thread | None = None
        self._session: ClientSession | None = None
        self._stop: asyncio.Event | None = None

桥的做法：在后台线程里跑一个专属事件循环，把 stdio_client 与 ClientSession 的整个生命周期关在同一个协程里（这是实测踩出来的硬约束——anyio 的 task group 必须在进入它的那个任务里退出，否则报 Attempted to exit cancel scope in a different task），主线程则用 run_coroutine_threadsafe 把每次调用投递进去、阻塞等结果。

    def connect(self, timeout: float = 10.0) -> None:
        self._loop = asyncio.new_event_loop()
        ready = threading.Event()

        def _run() -> None:
            asyncio.set_event_loop(self._loop)
            self._loop.run_until_complete(self._serve(ready))
            self._loop.close()

        self._thread = threading.Thread(target=_run, daemon=True)
        self._thread.start()
        if not ready.wait(timeout):
            raise TimeoutError(f"MCP server {self.name} 未在 {timeout}s 内就绪")

    async def _serve(self, ready: threading.Event) -> None:
        self._stop = asyncio.Event()
        params = StdioServerParameters(command=self.command, args=self.args)
        async with stdio_client(params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                self._session = session
                ready.set()
                await self._stop.wait()        # 保活，直到 close() 触发

发现工具就是把 list_tools 的结果翻译成 Tool。注意第二个实测坑：SDK 返回的工具对象字段是蛇形的 input_schema，不是网上教程里的 inputSchema——写错会静默拿到空 schema。

    def list_tools(self):
        fut = asyncio.run_coroutine_threadsafe(self._session.list_tools(), self._loop)
        return fut.result(10.0)

    def call_tool(self, name: str, arguments: dict[str, Any]):
        fut = asyncio.run_coroutine_threadsafe(
            self._session.call_tool(name, arguments), self._loop)
        return fut.result(30.0)

    def register(self, registry: ToolRegistry, namespace: str | None = None) -> list[str]:
        exposed: list[str] = []
        for t in self.list_tools().tools:
            name = f"{namespace}.{t.name}" if namespace else t.name
            registry.add(Tool(name=name, description=t.description or "",
                              parameters=t.input_schema, func=self._make_func(t.name)))
            exposed.append(name)
        return exposed

_make_func 是适配的核心：它返回一个闭包，Agent 调用时闭包转发 call_tool，把远程结果转成字符串。

    def _make_func(self, remote_name: str):
        def _call(**arguments: Any) -> str:
            if self._session is None:
                return json.dumps({"error": "mcp server not connected"}, ensure_ascii=False)
            try:
                result = self.call_tool(remote_name, arguments)
            except Exception as e:
                return json.dumps({"error": f"{type(e).__name__}: {e}"}, ensure_ascii=False)
            if result.is_error:
                text = "".join(getattr(c, "text", "") for c in result.content)
                return json.dumps({"error": text or "remote tool failed"}, ensure_ascii=False)
            return "".join(getattr(c, "text", "") for c in result.content)
        return _call

三个细节值得记：

  • 返回值永远是字符串，且永远不抛异常——这是卷一 ToolRegistry.call 的契约，远程工具也必须遵守，否则一次远端报错就会打断整个 Agent 循环。
  • result.content 是内容块列表，每块有 .text。文本工具直接拼起来；图片、资源引用等其它块类型要另做处理，本书只取文本。
  • is_error 是远程应用级错误（工具跑失败了），和「调用本身失败」（连接断了、超时）是两回事，前者读 content、后者进 except，两条路都归到同一个 {"error": ...}。

10.1.4 命名冲突：两个 Server 有同名工具

真实场景里连着好几个 Server，同名工具几乎必然发生——比如两个 Server 都提供 echo、都提供 search。而卷一的 ToolRegistry.add 是「后者覆盖前者」，若不管，先注册的会被静默挤掉。

我们的策略是给冲突方加命名空间前缀。再起一个夹具 Server，故意暴露同名的 echo：

# fixture_server_b.py —— 第二个夹具，故意与 A 同名
from mcp.server.mcpserver import MCPServer

mcp = MCPServer("analytics-fixture")

@mcp.tool()
def echo(text: str) -> str:
    """这个 echo 来自 analytics Server，与 research-fixture 的 echo 同名。"""
    return f"[analytics] {text}"

@mcp.tool()
def row_count(table: str) -> int:
    """返回某个表的行数（夹具里是假的固定值）。"""
    if table not in {"papers", "runs"}:
        raise ValueError(f"unknown table: {table}")
    return {"papers": 128, "runs": 40}[table]

if __name__ == "__main__":
    mcp.run("stdio")

两个 Server 都注册进同一个注册表，第二个带 namespace="analytics"：

a = MCPToolSource("research-fixture", PY, ["fixture_server.py"])
b = MCPToolSource("analytics-fixture", PY, ["fixture_server_b.py"])
a.connect(); b.connect()
reg = ToolRegistry()
print("A 注册:", a.register(reg))
print("B 注册:", b.register(reg, namespace="analytics"))
print("最终工具名:", [t["name"] for t in reg.schemas()])

实测输出：

A 注册: ['echo', 'search_corpus', 'word_count']
B 注册: ['analytics.echo', 'analytics.row_count']
最终工具名: ['echo', 'search_corpus', 'word_count', 'analytics.echo', 'analytics.row_count']

三种策略各有取舍：

策略做法优点缺点
覆盖后注册赢零改动静默丢工具，最难查的 bug
前缀命名空间server.tool无歧义、可溯源名字变长，模型要多记一段
拒绝启动检测到冲突直接报错逼你在配置期解决第三方 Server 不可控时太硬

我们选前缀，因为它是唯一「既不丢工具、又不需要人工介入」的方案。但要注意：前缀对模型是可见的，analytics.echo 会进 schema 的 name 字段，模型得靠描述理解它是哪个 Server 的能力——所以描述写清楚仍然重要（见 10.2）。

10.1.5 错误映射与生命周期

错误映射分两层，前面 _make_func 已经埋好：

未知远程工具: {"error": "Unknown tool: no_such_tool"}
连接已关闭:   {"error": "mcp server not connected"}

第一行是「远程 Server 不认这个工具名」——SDK 把它包成 is_error=True，我们从 content 里取到原文。第二行是「连接已经关了」——闭包开头那道 self._session is None 的检查挡下来的。两条路径都返回合法 JSON，Agent 循环照常继续。

生命周期要回答三个问题：

问题答案
连接何时建服务启动时一次性 connect()，不要每次调用都连
连接何时关进程退出时 close()，stdio_client 会终止子进程
断了怎么办调用返回 {"error": ...} 不中断；由上层按 7.2 的重试策略决定是否重连
    def close(self, timeout: float = 5.0) -> None:
        if self._loop is None:
            return
        self._loop.call_soon_threadsafe(self._stop.set)   # 让 _serve 退出 with 块
        self._thread.join(timeout)
        self._loop = None
        self._session = None

「断了返回错误而不是抛异常」这个选择，直接决定了 Agent 的韧性：一次瞬时断连只消耗一轮，模型看到 {"error": "mcp server not connected"} 后可以换个工具或告知用户，而不是把整个会话打断。真正要重连时，重连本身也是 connect() 一次而已——Server 是无状态的，重连成本很低。

10.1.6 真跑：假模型驱动的 Agent 通过 MCP 调工具

把工具注册进 ToolRegistry 后，Agent 完全不知道它们是远程的。用卷一的 FakeModel 驱动一次「问 MCP 是什么」：

model = FakeModel([
    [ToolCall(id="c1", name="search_corpus", arguments={"query": "mcp"})],
    "MCP 让 Agent 通过统一协议接入外部工具与资源。",
])
agent = Agent(model, registry, system="你是研究助理。")
result = agent.run("MCP 是什么？")
for s in result.steps:
    print(f"  [{s.index}] {s.kind}: {s.detail}")
print("final:", result.final)
print("模型看到的工具名:", [t["name"] for t in model.calls[0][1]])

实测输出（Python 3.14.6 + mcp 2.3.0，Server 为本地 stdio 夹具）：

注册的 MCP 工具: ['echo', 'search_corpus', 'word_count']
steps:
  [0] tool_call: search_corpus -> MCP 让 Agent 通过统一协议接入外部工具与资源。
  [1] final: MCP 让 Agent 通过统一协议接入外部工具与资源。
final: MCP 让 Agent 通过统一协议接入外部工具与资源。
模型看到的工具名: ['echo', 'search_corpus', 'word_count']

三步印证了适配是透明的：第 0 步 Agent 通过 MCP 真调了一次远程工具并拿到结果，第 1 步收敛；最后一行说明模型拿到的 schema 里，工具名和本地工具一模一样。从 Agent 的视角，MCP 工具和本地工具没有任何区别——这正是适配器的目标。

真实环境 / 真实模型下会怎样：真实 Claude 面对 MCP 工具时会主动看描述挑工具，但它看不到 Server 边界——analytics.echo 和 echo 在它眼里只是两个名字不同的工具，描述没写清就会挑错。另一个真实差异是延迟：本地 stdio Server 调用是毫秒级，远程 HTTP Server 可能是几百毫秒，Agent 一轮多调几个工具，延迟会叠加，这也是 8.2 背压与 14.1 延迟剖析要处理的。验收清单：

  • connect() 只建一次连接，不是每次调用都连
  • 工具 schema 的 input_schema 字段拼写正确（不是 inputSchema）
  • 远程报错和连接断开都返回 {"error": ...}，不抛异常
  • 多 Server 同名工具已按前缀消歧，无静默覆盖
  • close() 能终止 stdio 子进程，不留孤儿进程

阅读导航：上一节：9.3 断线重连：Last-Event-ID 续传不丢步 · 下一节：10.2 工具设计模式：读写分离、幂等、可观测 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「agent」更多文章

  1. 《AI Agent 高级编程》目录
  2. 《AI Agent 高级编程》11.3 全景回顾：从零到框架的完整路径与展望
  3. 《AI Agent 高级编程》11.2 真实 API 冒烟：假模型 vs 真模型量化差距