MCP：让 Agent 对接外部工具生态

这篇笔记解决什么问题

自己给 Agent 写工具并不难：定义一个 Python 函数，描述好参数 schema，注册进工具表，模型就能调用。但只要 Agent 数量一多、要对接的外部系统一多，这条路很快就走不下去了——每个 Agent 都要重写一遍"查订单""搜文档""发消息"，外部系统改一个接口，所有 Agent 挨个改一遍。

MCP（Model Context Protocol）就是为这个场景设计的开放协议：把外部能力做成独立的 Server，任何支持 MCP 的 Agent 都能连上来发现工具、调用工具。这篇笔记讲清楚三件事：MCP 到底解决什么问题、它的三个核心概念怎么协作、以及在生产环境接入 MCP 时最容易踩的坑（连接复用与故障隔离）。读完你可以自己动手写一个最小的 Server + Client + Agent 闭环。

为什么需要一层协议

先想一个具体场景：公司里有客服 Agent 和运营 Agent，两个都需要"按手机号查会员信息"。

没有 MCP 时，两条路都不舒服：

• 各写各的：两个 Agent 各自实现一版查询工具，逻辑重复，字段口径还可能不一致。
• 抽公共库：好一些，但仅限同语言同技术栈，而且库升级后每个 Agent 都要重新发版。

有了 MCP，会员查询做成一个独立 Server，两个 Agent 通过协议连接它。这带来两个直接收益：

1. 能力复用：Server 写一次，所有 Agent 共用。后面新增第三个、第四个 Agent，接入成本只是加一行配置。
2. 升级无感：Server 端加一个"查询会员消费明细"的新工具，Agent 什么代码都不用改——下次连接时 tools/list 自然会多出一个工具，模型看到描述就会用。

一句话概括：MCP 解决的是外部能力的复用与标准化。如果你的 Agent 很小、工具很少、也不碰外部系统，直接写本地函数完全够用，不必为了用 MCP 而用 MCP。

三个核心概念

MCP 的架构可以拆成三块：Server 提供能力，Client 建立连接并发起调用，Transport 决定消息在两者之间怎么传。

Server：暴露三类能力

一个 MCP Server 可以对外暴露：

• tools：可执行的动作，带参数 schema，比如"搜索工单""创建日程"。
• resources：可读取的上下文数据，用 URI 定位，比如一份配置文件、一张报表。
• prompts：预置的提示词模板。

实际工程里九成以上的使用集中在 tools 上，resources 和 prompts 见得少，初学阶段可以先只关注 tools。

Client：三步走的会话

Agent 侧通过 MCP Client 与 Server 通信，核心就三个动作：

initialize   → 握手，交换协议版本和能力声明
tools/list   → 问 Server：你有哪些工具？拿到名字、描述、参数 schema
tools/call   → 按名字调用某个工具，拿回结果

底层消息格式是 JSON-RPC，但官方 SDK（Python/TypeScript 都有）已经封装好了，日常开发不需要手动拼消息。

Transport：stdio 与 streamable HTTP

传输层最常用两种：

• stdio：Client 把 Server 当子进程拉起来，通过标准输入输出通信。零网络配置，本地工具的首选。
• streamable HTTP：Server 独立部署成 HTTP 服务，适合团队内统一维护、多个 Agent 远程共享的场景。

本文示例用 stdio，换成 HTTP 只是改连接参数，上层代码不变。

关键认知：模型根本看不出区别

这是理解 MCP 最重要的一点：在模型眼里，MCP 工具和本地工具没有任何区别。

模型每轮收到的都是同一种东西——一组 function 定义，每个有 name、description、parameters（JSON Schema）。它按语义挑一个调用，返回 tool_call。至于这个工具背后是同进程的 Python 函数，还是另一个进程里的 MCP Server，甚至是大洋彼岸的一个 HTTP 服务，模型不知道也不关心。

真正发生变化的是 Agent runtime 的分发层。模型说"调这个工具"之后，runtime 要判断：这个名字对应本地函数？直接调。对应某个 MCP Server 的工具？走 Client 的 tools/call。工程差异集中在这一层：

动手：最小闭环

下面用一个"书签管家"场景走通全流程：Server 管理书签数据，Agent 通过 MCP 操作它。安装依赖：pip install mcp openai。

第一步：写一个 MCP Server

# servers/bookmark_server.py
import json
from pathlib import Path

from mcp.server.fastmcp import FastMCP

STORE = Path(__file__).parent / "bookmarks.json"
app = FastMCP("bookmark-box")


def _load() -> list[dict]:
    return json.loads(STORE.read_text("utf-8")) if STORE.exists() else []


def _dump(items: list[dict]) -> None:
    STORE.write_text(json.dumps(items, ensure_ascii=False, indent=2), "utf-8")


@app.tool(description="保存一条书签，需要标题和 URL")
def save_bookmark(title: str, url: str) -> str:
    items = _load()
    items.append({"title": title, "url": url})
    _dump(items)
    return f"已保存：{title}"


@app.tool(description="列出当前保存的全部书签")
def list_bookmarks() -> str:
    items = _load()
    if not items:
        return "书签库是空的"
    return "\n".join(f"{it['title']} -> {it['url']}" for it in items)


@app.tool(description="按关键词模糊搜索书签标题")
def find_bookmark(keyword: str) -> str:
    hits = [it for it in _load() if keyword.lower() in it["title"].lower()]
    return "\n".join(f"{it['title']} -> {it['url']}" for it in hits) or "没有匹配的书签"


if __name__ == "__main__":
    app.run(transport="stdio")

注意两个细节：@app.tool() 装饰器会把函数签名的类型标注自动转成 JSON Schema，save_bookmark(title: str, url: str) 在模型那边就是一个要求两个字符串参数的工具；另外 Server 不必从零造轮子——包一层已有的内部函数、SDK 或 HTTP API，是 MCP Server 最常见的写法。

第二步：用 Client 验证 Server

先不引入模型，单独确认协议链路是通的：

# probe_client.py
import asyncio

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

SERVER = StdioServerParameters(
    command="python",
    args=["servers/bookmark_server.py"],
)


async def main() -> None:
    async with stdio_client(SERVER) as (reader, writer):
        async with ClientSession(reader, writer) as session:
            await session.initialize()                        # 1. 握手

            listing = await session.list_tools()              # 2. 发现
            for t in listing.tools:
                print(f"{t.name}: {t.description}")

            outcome = await session.call_tool(                # 3. 调用
                "save_bookmark",
                {"title": "MCP 规范", "url": "https://modelcontextprotocol.io"},
            )
            print(outcome.content[0].text)


asyncio.run(main())

跑通这一步，你就亲手完成了一次 initialize → tools/list → tools/call 的完整会话。

第三步：接进 Agent 循环

把 MCP 工具翻译成 OpenAI 兼容接口的 tools 数组，让模型自己决定调什么。模型端点用环境变量配置，DeepSeek、Kimi、GLM 等任何 OpenAI 兼容服务都能跑：

# bridge_agent.py
import asyncio
import json
import os

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

llm = OpenAI(
    api_key=os.environ["LLM_API_KEY"],
    base_url=os.environ["LLM_BASE_URL"],   # 任意 OpenAI 兼容端点
)
MODEL = os.environ.get("LLM_MODEL", "your-model-id")
ALIAS = "bookmarks"                         # 本 Server 的命名前缀


def as_llm_tool(tool) -> dict:
    """MCP 工具定义 → OpenAI function 定义"""
    return {
        "type": "function",
        "function": {
            "name": f"mcp__{ALIAS}__{tool.name}",
            "description": tool.description or tool.name,
            "parameters": tool.inputSchema or {"type": "object", "properties": {}},
        },
    }


def unwrap(result) -> str:
    text = "\n".join(c.text for c in result.content if c.type == "text")
    return f"[工具报错] {text}" if getattr(result, "isError", False) else text


async def run(task: str) -> None:
    server = StdioServerParameters(
        command="python", args=["servers/bookmark_server.py"]
    )
    async with stdio_client(server) as (reader, writer):
        async with ClientSession(reader, writer) as session:
            await session.initialize()
            listing = await session.list_tools()

            llm_tools = [as_llm_tool(t) for t in listing.tools]
            # 前缀名 → 原始工具名，调用时要还原
            dispatch = {f"mcp__{ALIAS}__{t.name}": t.name for t in listing.tools}

            history = [
                {"role": "system", "content": "你是书签管家，所有书签操作必须通过工具完成。"},
                {"role": "user", "content": task},
            ]

            for _ in range(6):  # 轮数上限，防失控
                reply = llm.chat.completions.create(
                    model=MODEL, messages=history, tools=llm_tools
                )
                msg = reply.choices[0].message
                history.append(msg.model_dump(exclude_none=True))

                if not msg.tool_calls:          # 没有工具调用 = 最终回答
                    print(msg.content)
                    return

                for call in msg.tool_calls:
                    real_name = dispatch[call.function.name]
                    args = json.loads(call.function.arguments or "{}")
                    result = await session.call_tool(real_name, args)
                    history.append({
                        "role": "tool",
                        "tool_call_id": call.id,
                        "content": unwrap(result),
                    })


asyncio.run(run("把 Python 官网 python.org 存成书签，然后列出我现在有哪些书签"))

回看这段代码：Agent 主循环完全不关心工具在哪执行，它只做两件事——把工具定义交给模型、把模型点名的调用分发给正确的执行器。这正是前面说的"变化只发生在分发层"。

前缀命名：多 Server 必备

上面代码里 mcp__bookmarks__save_bookmark 这种命名不是装饰。一旦接入多个 Server，重名几乎必然发生——search、query、read_file 这类名字每个 Server 都爱用。直接把原始名丢给模型，后接入的会覆盖先接入的，或者模型调错对象。

约定 mcp__{server标识}__{工具名} 的好处有三个：

• 天然避免跨 Server 冲突；
• 分发层看一眼前缀就知道该转发给哪个 Server；
• 排查日志时能立刻定位工具来源。

记得同时维护一张"前缀名 → (server, 原始名)"的映射表：给模型看的是加工过的安全名字，真正 tools/call 时必须用 Server 声明的原始名字。

生产要点一：连接复用

demo 里每次运行都新起子进程、握手、列工具、调用、退出，一次性任务无所谓。但生产 Agent 一轮对话可能调用三五次工具，如果每次调用都重启一遍 stdio Server（有些 Server 启动还要装依赖、连数据库、预热缓存），延迟会直接劝退用户。答案显然是复用连接——但这里有个隐蔽的深坑。

直觉做法是把 ClientSession 塞进全局字典缓存：

# 反面示例：看起来能跑，随时会炸
sessions[server_id] = session

问题在于 MCP Python SDK 的连接上下文基于 AnyIO 的 cancel scope，它对 asyncio Task 的归属很敏感：在哪个 Task 里进入的上下文，就必须在同一个 Task 里退出。而 Web 服务里每个请求往往是独立的 Task——A 请求创建的 session 被 B 请求拿去用、再被清理协程关掉，就会撞上：

RuntimeError: Attempted to exit cancel scope in a different task than it was entered in

正确的模式是：每个 Server 配一个长期 worker 协程，session 的创建、使用、销毁全部锁死在这个 worker 自己的 Task 里；外部调用方只通过队列投递请求、等待结果，绝不直接触碰 session：

import asyncio
from contextlib import AsyncExitStack


class ServerWorker:
    """一个 MCP Server 一个 worker：会话生命周期只属于本 Task。"""

    def __init__(self, params: StdioServerParameters):
        self.params = params
        self.inbox: asyncio.Queue = asyncio.Queue()
        self.online = asyncio.Event()
        self.tools: list = []

    async def start(self) -> None:
        self._task = asyncio.create_task(self._serve())
        await self.online.wait()

    async def _serve(self) -> None:
        async with AsyncExitStack() as stack:
            reader, writer = await stack.enter_async_context(
                stdio_client(self.params)
            )
            session = await stack.enter_async_context(
                ClientSession(reader, writer)
            )
            await session.initialize()
            self.tools = (await session.list_tools()).tools
            self.online.set()

            while True:
                job = await self.inbox.get()
                if job is None:
                    return          # 关闭信号：在本 Task 内退出上下文
                name, args, done = job
                try:
                    result = await session.call_tool(name, args)
                    done.set_result(unwrap(result))
                except Exception as exc:
                    done.set_result(f"[工具报错] {exc}")

    async def call(self, name: str, args: dict) -> str:
        done = asyncio.get_running_loop().create_future()
        await self.inbox.put((name, args, done))
        return await done

    async def stop(self) -> None:
        await self.inbox.put(None)
        await self._task

这样连接只建立一次，后续所有工具调用都复用同一个会话，Task 归属问题也不存在了。

生产要点二：故障隔离

MCP Server 是外部依赖，外部依赖就会挂：包没装好、API Key 过期、网络抖动。原则是：一个 Server 连不上，只损失它自己的工具，绝不能拖垮整个 Agent。

实现上就是在启动和列工具时兜底：

async def collect_tools(workers: dict[str, ServerWorker]) -> list[dict]:
    available = []
    for alias, worker in workers.items():
        try:
            await asyncio.wait_for(worker.start(), timeout=10)
        except Exception as exc:
            print(f"[warn] MCP server {alias} 不可用，跳过：{exc}")
            continue    # 只跳过这一个，其余照常
        available += [as_llm_tool(t) for t in worker.tools]
    return available

书签 Server 挂了，Agent 依然能用内置工具和其他 Server 的能力回答问题，最多在回答里说明"书签功能暂时不可用"——这比整个 Agent 启动失败体面得多。

三类能力怎么选

到这里，一个成熟 Agent 的能力体系通常有三种形态，它们是互补关系而非替代关系：

拿"周报 Agent"举例：用日历 MCP Server 拉本周日程，用内置工具读本地的工作日志文件，用 Skill 规定周报的结构和口吻。模型看到的始终是一组扁平的 tools，但工程上每类能力有清晰的边界和独立的生命周期。判断标准也简单：动作和项目绑死→内置工具；只是方法论不需要执行→Skill；能力来自外部系统或想给多个 Agent 复用→MCP。

动手清单

1. 把本文的 bookmark_server.py 跑起来，用 probe_client.py 完成一次 initialize → tools/list → tools/call，观察打印出来的 inputSchema 长什么样。
2. 给 Server 加一个 delete_bookmark(title: str) 工具，不改 Agent 一行代码重新运行 bridge_agent.py，验证"Server 升级、Agent 无感"。
3. 再写一个极简的第二个 Server（比如待办清单），让 Agent 同时挂两个 Server，亲手实现前缀命名和分发映射。
4. 故意把第二个 Server 的启动命令写错，验证你的故障隔离逻辑：Agent 应该照常启动，只是少一组工具。
5. 把 demo 的"每次调用即连即断"改造成 ServerWorker 模式，对比同一任务下两种方式的总耗时。

*本文为学习笔记，知识框架参考自叶小钗「生产级 Agent」系列课程；文字与代码示例为本站原创整理。*