MCP 这个缩写最近在技术圈出现的频率越来越高但很多人第一次听到时的反应都是这又是什么新协议。简单说Model Context Protocol 是一套让 AI 应用与外部工具、数据源之间建立标准化通信的协议规范。它的核心价值在于把过去每个 AI 应用都要自己写一遍的调工具逻辑抽象成了一套统一的握手、发现、调用流程。你可以把它理解成 AI 世界里的 USB-C 接口——不管对面接的是数据库、文件系统还是某个内部 API只要双方都遵守 MCP 的约定就能即插即用。这篇文章面向的是已经对 LangGraph 有基本了解、想进一步搞清楚 MCP 协议底层握手细节以及如何在 LangGraph 中同时挂载多个 MCP Server 的开发者。我会从协议握手的实际报文流程讲起拆解 Client 与 Server 之间到底交换了哪些信息然后一步步搭建一个能同时调用多个 Server 的 LangGraph 工作流。中间会穿插我在实际调试中踩过的坑比如初始化顺序导致的超时、多 Server 场景下的工具名冲突、以及流式输出时的缓冲问题。如果你正在做 AI Agent 的工具集成或者单纯想搞明白 MCP 到底怎么运转这篇应该能帮你省下不少翻文档的时间。1. 协议握手阶段到底发生了什么1.1 从 TCP 连接到能力协商的完整链路很多人以为 MCP 的握手就是简单的连上了就能用实际上从传输层建立连接到真正可以调用工具中间要经过好几个阶段。我用一个实际抓包的过程来还原这条链路。第一步是传输层建立。MCP 支持多种传输方式最常见的是 stdio标准输入输出和 SSEServer-Sent Events。stdio 模式下Client 会以子进程的方式启动 Server双方通过标准输入输出流通信。SSE 模式下则是通过 HTTP 长连接。不管哪种方式传输层建立之后双方并不会立刻开始干活。第二步是初始化请求。Client 会发送一个initialize请求这个请求里包含几个关键字段protocolVersion表示 Client 支持的协议版本capabilities表示 Client 自身具备哪些能力比如是否支持 roots、是否支持 samplingclientInfo则是 Client 的名称和版本号。这个请求的本质是在说我是谁我支持什么你那边什么情况第三步是 Server 的初始化响应。Server 收到后会返回自己的protocolVersion、capabilities和serverInfo。这里的capabilities是重点它决定了后续 Client 能对这个 Server 做什么。比如 Server 返回了tools能力说明它提供了可调用的工具返回了resources说明它暴露了可读取的资源返回了prompts说明它提供了预定义的提示模板。第四步是初始化完成通知。Client 收到 Server 的响应后会再发一个notifications/initialized通知告诉 Server 我准备好了可以开始正式交互了。只有这个通知发出去之后后续的tools/list、tools/call等请求才会被 Server 正常处理。注意很多初学者在调试时发现tools/list返回空或者直接报错八成是因为跳过了notifications/initialized这一步。协议规范里明确要求这个通知必须先发但有些 Server 实现比较宽松不报错也不返回数据排查起来很费时间。1.2 能力协商中的字段含义与常见陷阱capabilities这个字段看起来简单但里面的门道不少。我拿一个实际的 Server 响应举例{ protocolVersion: 2024-11-05, capabilities: { tools: { listChanged: true }, resources: { subscribe: true, listChanged: true }, prompts: { listChanged: false } }, serverInfo: { name: example-server, version: 1.0.0 } }这里的listChanged表示当工具列表发生变化时Server 是否会主动发送通知。subscribe表示 Client 是否可以订阅某个资源的变更。这些字段看起来是可选的能力声明但实际上会影响 Client 的行为逻辑。比如如果listChanged为 trueClient 就应该监听notifications/tools/list_changed通知并在收到后重新拉取工具列表。常见的陷阱有三个。第一个是协议版本不匹配。Client 和 Server 声明的protocolVersion如果不一致有些实现会直接断开连接有些则会尝试降级兼容。我建议在 Client 侧做好版本判断遇到不匹配时给出明确的错误提示而不是让后续调用莫名其妙地失败。第二个陷阱是能力声明与实际行为不一致。我遇到过某个 Server 声明了resources能力但实际调用resources/list时返回的是空数组。这种情况通常是 Server 实现不完整导致的Client 侧需要做好空值处理不能假设声明了能力就一定有数据。第三个陷阱是serverInfo里的名称重复。当你同时连接多个 Server 时如果两个 Server 都叫example-server在日志里就很难区分是谁在响应。我的做法是在 Client 侧给每个 Server 分配一个本地别名日志里统一用别名加原始名称的格式输出。1.3 握手失败的排查思路握手失败的表现形式很多有的是连接直接断开有的是请求超时有的是返回了错误码但信息很模糊。我总结了一套排查顺序基本能覆盖大部分场景。先看传输层是否正常。stdio 模式下检查子进程是否成功启动可以手动在终端里执行 Server 的启动命令看是否有报错输出。SSE 模式下用 curl 直接请求 SSE 端点看是否能建立连接并收到初始事件。再看初始化请求是否发出。在 Client 侧打开调试日志确认initialize请求的完整报文。重点检查protocolVersion字段是否填写正确有些 Server 对版本号格式很敏感写成2024-11-5和2024-11-05可能结果完全不同。然后看 Server 的响应。如果 Server 返回了错误错误信息里通常会包含具体原因。我遇到过一个 Server 在收到不支持的协议版本时返回的错误信息是Unsupported protocol version但实际原因是它的版本判断逻辑写错了把2024-11-05误判成了不支持的版本。这种情况只能去看 Server 的源码或者提 issue。最后看notifications/initialized是否发送成功。这个通知是单向的Server 不会返回响应所以很容易被忽略。可以在 Client 侧加一个日志确认这个通知确实发出去了。2. LangGraph 中挂载单个 MCP Server 的最小实现2.1 为什么选择 LangGraph 作为编排层LangGraph 的核心优势在于它把 Agent 的执行流程建模成了图结构每个节点是一个操作边表示流转条件。这种模型天然适合处理调用工具、根据结果决定下一步这类逻辑。而 MCP 提供的是标准化的工具调用接口两者结合之后你可以把 MCP Server 提供的工具直接作为 LangGraph 图中的节点来使用。相比自己写一个循环来不断调用工具LangGraph 的好处是状态管理更清晰。每次工具调用的输入输出都会被记录在状态里方便回溯和调试。而且 LangGraph 支持条件边可以根据工具返回的结果动态决定下一步走哪个分支这在处理复杂任务时非常有用。另一个考虑是生态兼容性。LangGraph 本身是 LangChain 生态的一部分可以很方便地和其他组件比如各种 LLM、记忆模块、回调系统集成。如果你已经在用 LangChain 做开发迁移到 LangGraph 的成本很低。2.2 用 stdio 方式连接第一个 Server先从一个最简单的场景开始连接一个提供文件读取工具的 MCP Server并在 LangGraph 中调用它。这里我用 Python 来演示因为 LangGraph 的 Python 生态最成熟。首先安装依赖pip install langgraph langchain-mcp-adapters mcplangchain-mcp-adapters是 LangChain 官方提供的适配器它把 MCP 的工具转换成了 LangChain 的 Tool 格式这样就能直接在 LangGraph 里用了。接下来是连接代码from langchain_mcp_adapters.client import MultiServerMCPClient from langgraph.prebuilt import create_react_agent from langchain_openai import ChatOpenAI async def main(): client MultiServerMCPClient( { filesystem: { command: npx, args: [-y, modelcontextprotocol/server-filesystem, /tmp], transport: stdio, } } ) tools await client.get_tools() model ChatOpenAI(modelgpt-4o) agent create_react_agent(model, tools) result await agent.ainvoke( {messages: [{role: user, content: 列出 /tmp 目录下的文件}]} ) print(result)这段代码做了几件事创建了一个MultiServerMCPClient实例配置了一个名为filesystem的 Server通过 npx 启动。然后调用get_tools()获取所有工具传给create_react_agent创建 Agent最后执行一次调用。MultiServerMCPClient这个名字里的 Multi 很关键它天生就是为多 Server 场景设计的。即使你现在只连一个 Server后续要扩展成多个也很方便。2.3 工具发现与调用的实际流程当你调用client.get_tools()时底层发生的事情比表面看起来要多。Client 会依次对每个配置的 Server 执行握手流程然后调用tools/list获取工具列表最后把每个工具包装成 LangChain 的StructuredTool对象。这里有一个细节值得注意工具的名称会被加上 Server 名称的前缀吗答案是取决于适配器的实现。在我用的这个版本里工具名称保持原样不会自动加前缀。这意味着如果你连接了两个 Server而它们都有一个叫read_file的工具就会发生名称冲突。这个问题我在下一节会详细讲怎么处理。工具调用时的流程是Agent 决定调用某个工具LangGraph 把调用请求转给MultiServerMCPClientClient 根据工具名称找到对应的 Server发送tools/call请求等待响应然后把结果返回给 Agent。整个过程对 Agent 来说是透明的它不需要知道工具背后是哪个 Server。提示在调试时可以在 Client 侧打开日志查看每次tools/call的实际请求和响应报文。这对于排查工具返回结果不符合预期的问题非常有帮助。3. 多 Server 场景下的工具命名冲突与路由策略3.1 冲突是怎么产生的当你同时连接多个 MCP Server 时工具名称冲突几乎是一定会遇到的问题。比如你连了一个文件系统 Server 和一个数据库 Server两者都可能提供名为query或read的工具。在 LangGraph 中工具名称是唯一的标识符如果两个工具重名Agent 在调用时就无法区分该用哪一个。我实际遇到过一个更隐蔽的冲突两个 Server 都提供了search工具但一个是在本地文件里搜索另一个是在远程 API 里搜索。由于名称相同Agent 有时会调用错误的工具导致返回结果完全不符合预期。这种问题在测试阶段很难发现因为两个工具都能正常返回结果只是结果的内容不对。3.2 用命名空间前缀做隔离最直接的解决方案是给每个 Server 的工具加上命名空间前缀。MultiServerMCPClient本身不提供这个功能但我们可以自己包一层async def get_prefixed_tools(client, prefix_map): all_tools [] for server_name, prefix in prefix_map.items(): server_tools await client.get_tools(server_nameserver_name) for tool in server_tools: tool.name f{prefix}_{tool.name} all_tools.append(tool) return all_tools这样filesystemServer 的read_file就变成了fs_read_filedatabaseServer 的read_file变成了db_read_file冲突就解决了。但这样做有一个副作用Agent 看到的工具名称变长了可能会影响它选择工具的准确率。我的经验是前缀不要太长两到三个字母就够了比如fs、db、api。另外在工具的 description 里也要相应更新把前缀的含义解释清楚。3.3 基于路由表的动态分发另一种思路是不改工具名称而是在 Client 侧维护一张路由表记录每个工具名称对应哪个 Server。当 Agent 调用工具时Client 查表找到目标 Server再转发请求。这种方案的好处是工具名称保持原样Agent 的提示词不需要调整。坏处是实现复杂度更高而且如果两个 Server 的工具名称完全一样路由表本身就无法区分还是得靠前缀或者其他机制来解决。我个人的选择是混合策略对于名称不太可能冲突的 Server直接用原名对于可能冲突的加前缀。判断是否可能冲突的方法很简单把两个 Server 的工具列表拉出来看有没有交集。有交集就加前缀没有就不加。策略优点缺点适用场景统一加前缀实现简单彻底避免冲突工具名变长可能影响选择准确率Server 数量多、工具名重叠概率高路由表分发工具名保持原样实现复杂无法处理完全重名Server 数量少、工具名基本不重叠混合策略兼顾两者优点需要手动判断是否加前缀大多数实际场景3.4 多 Server 初始化顺序与超时处理多 Server 场景下初始化顺序和超时是另一个容易出问题的地方。MultiServerMCPClient默认是并发初始化所有 Server 的但如果某个 Server 启动很慢比如需要加载大模型或者连接远程数据库就可能导致整体超时。我遇到过一次这样的情况三个 Server 里有一个是连接远程 API 的网络延迟很高初始化花了将近 30 秒。而 Client 的默认超时是 10 秒导致这个 Server 一直被判定为初始化失败。解决方案是在配置里单独给这个 Server 设置更长的超时时间client MultiServerMCPClient( { slow_server: { command: python, args: [slow_server.py], transport: stdio, timeout: 60, }, fast_server: { command: npx, args: [-y, modelcontextprotocol/server-filesystem, /tmp], transport: stdio, }, } )另外要注意的是如果某个 Server 初始化失败默认行为是抛出异常整个 Client 创建过程都会失败。如果你希望某个 Server 失败不影响其他 Server可以在创建时捕获异常或者使用get_tools的raise_on_errorFalse参数如果适配器支持的话。4. 流式输出与状态同步的实战细节4.1 MCP 的流式响应机制MCP 协议本身支持流式响应但具体实现方式取决于传输层。stdio 模式下Server 可以通过标准输出持续发送数据块Client 逐块读取。SSE 模式下Server 通过事件流推送数据。在 LangGraph 中流式输出主要体现在两个层面一是 LLM 生成 token 的流式输出二是工具调用结果的流式返回。前者由 LangChain 的回调系统处理后者需要 MCP Client 和 LangGraph 配合。我实际测试下来工具调用结果的流式返回在 stdio 模式下工作得比较好因为标准输入输出本身就是流式的。但在 SSE 模式下如果 Server 没有正确实现事件推送可能会出现结果一次性返回而不是逐块返回的情况。4.2 在 LangGraph 中消费流式结果LangGraph 提供了astream_events方法来消费流式事件。下面是一个完整的示例async def stream_agent(agent, input_message): async for event in agent.astream_events( {messages: [{role: user, content: input_message}]}, versionv2, ): kind event[event] if kind on_chat_model_stream: content event[data][chunk].content if content: print(content, end, flushTrue) elif kind on_tool_start: print(f\n[调用工具: {event[name]}]) elif kind on_tool_end: print(f\n[工具返回: {event[data].get(output)[:100]}...])这段代码会实时打印 LLM 生成的 token以及工具调用的开始和结束事件。flushTrue很重要否则输出可能会被缓冲看起来像是一次性输出的。注意在 Jupyter Notebook 里运行时流式输出的效果可能和终端里不一样因为 Notebook 有自己的输出缓冲机制。建议在终端里测试流式效果。4.3 多 Server 场景下的状态一致性当你同时使用多个 Server 时状态一致性是一个需要关注的问题。比如 Agent 先调用 Server A 的工具创建了一个文件然后调用 Server B 的工具去读取这个文件如果 Server B 的视角里这个文件还不存在比如因为缓存或者同步延迟就会出错。我的做法是在 Agent 的提示词里明确说明工具之间的依赖关系让 LLM 知道某些操作有先后顺序。另外在 LangGraph 的状态里记录每个工具调用的结果这样后续的工具调用可以参考之前的结果。还有一个细节是错误处理。如果某个 Server 调用失败LangGraph 默认会把错误信息返回给 LLM让 LLM 决定下一步怎么做。这个机制大多数时候工作得很好但如果错误信息太模糊比如只是一个 Internal ErrorLLM 可能会陷入反复重试的循环。我建议在 Client 侧对错误信息做一层加工把关键信息提取出来再返回给 LLM。5. 调试与排错的几个关键手段5.1 打开 MCP 协议层的日志调试 MCP 相关问题时第一件事就是打开协议层的日志。langchain-mcp-adapters底层用的是mcp这个 Python 包它支持通过环境变量控制日志级别export MCP_LOG_LEVELdebug设置之后所有的握手请求、工具列表请求、工具调用请求和响应都会打印出来。我通常会把这些日志重定向到一个文件里方便后续搜索python my_agent.py 21 | tee mcp_debug.log然后在日志里搜索initialize、tools/list、tools/call这些关键词就能看到完整的交互流程。5.2 用 MCP Inspector 做独立验证如果日志看起来没问题但行为还是不对下一步是用 MCP Inspector 做独立验证。Inspector 是一个官方的调试工具可以单独连接 Server手动发送请求并查看响应。npx modelcontextprotocol/inspector启动后会打开一个 Web 界面你可以在里面配置 Server 的连接参数然后手动执行initialize、tools/list、tools/call等操作。这样可以排除 LangGraph 和适配器的影响确认问题是否出在 Server 本身。我用这个工具发现过好几个问题比如某个 Server 的tools/list返回的工具 schema 格式不对导致适配器解析失败。这种问题在 LangGraph 层面看就是工具列表为空很难定位到根因。5.3 常见错误码与对应处理下面这张表是我在实际调试中整理出来的常见错误和对应的处理方式错误表现可能原因处理方式连接后立即断开Server 启动失败或协议版本不匹配手动执行 Server 启动命令检查输出tools/list返回空未发送notifications/initialized检查 Client 实现确认通知已发送工具调用超时Server 处理时间过长或网络延迟增加超时时间检查 Server 日志工具名称冲突多个 Server 提供同名工具加命名空间前缀或使用路由表流式输出不生效传输层不支持流式或缓冲未刷新检查传输模式确认 flush 设置错误信息模糊Server 返回的错误描述不完整在 Client 侧加工错误信息5.4 性能优化的几个实际经验最后分享几个性能优化方面的经验。第一工具列表不需要每次调用都重新拉取可以在 Client 初始化时拉一次然后缓存起来。只有当收到list_changed通知时才重新拉取。第二对于频繁调用的工具可以考虑在 Client 侧做一层结果缓存。比如文件读取工具如果同一个文件在短时间内被多次读取可以直接返回缓存结果。但要注意缓存失效策略避免返回过期数据。第三多 Server 场景下如果某个 Server 的响应特别慢可以考虑给它设置独立的超时和重试策略避免拖慢整个 Agent 的响应速度。我在实际项目里把这三个优化都加上了Agent 的平均响应时间从 3 秒多降到了 1 秒以内。当然具体效果取决于你的场景和 Server 实现但思路是通用的。提示性能优化之前一定要先做 profiling确认瓶颈在哪里。我见过有人一上来就加缓存结果发现瓶颈其实在 LLM 调用上缓存根本没起作用。6. 从单 Server 到多 Server 的演进路径6.1 什么时候该拆分成多个 Server一开始用一个 Server 把所有工具都塞进去是最简单的做法但随着工具数量增加会逐渐遇到几个问题。一是启动变慢因为所有工具都要在同一个进程里初始化。二是职责不清一个 Server 里既有文件操作又有数据库操作维护起来很混乱。三是故障隔离差一个工具出问题可能影响整个 Server。我的经验是当工具数量超过 10 个或者工具明显属于不同领域比如文件、数据库、网络就应该考虑拆分。拆分的粒度不用太细按领域分就够了。比如文件相关的放一个 Server数据库相关的放一个 Server外部 API 相关的放一个 Server。6.2 拆分后的配置管理拆分成多个 Server 之后配置管理就变得重要了。我建议用一个统一的配置文件来管理所有 Server 的连接信息而不是硬编码在代码里servers: filesystem: command: npx args: [-y, modelcontextprotocol/server-filesystem, /data] transport: stdio timeout: 30 database: command: python args: [db_server.py] transport: stdio timeout: 60 env: DB_HOST: localhost DB_PORT: 5432然后在代码里读取这个配置动态创建 Client。这样做的好处是增加或删除 Server 不需要改代码只需要改配置文件。6.3 监控与可观测性多 Server 场景下监控每个 Server 的健康状态和调用情况很有必要。我通常会在 Client 侧记录每个 Server 的调用次数、平均响应时间、错误率等指标然后定期输出到日志或者监控系统。一个简单的实现方式是用装饰器包一层工具调用import time from functools import wraps def monitor_tool(func): wraps(func) async def wrapper(*args, **kwargs): start time.time() try: result await func(*args, **kwargs) duration time.time() - start print(f[{func.__name__}] 成功, 耗时 {duration:.2f}s) return result except Exception as e: duration time.time() - start print(f[{func.__name__}] 失败, 耗时 {duration:.2f}s, 错误: {e}) raise return wrapper这样每次工具调用都会输出耗时和结果状态方便快速定位问题。6.4 一个完整的端到端示例最后给一个完整的示例把前面讲的内容串起来。这个示例连接两个 Server文件系统和数据库给工具加了前缀支持流式输出并且带了监控import asyncio import time from langchain_mcp_adapters.client import MultiServerMCPClient from langgraph.prebuilt import create_react_agent from langchain_openai import ChatOpenAI async def main(): client MultiServerMCPClient( { fs: { command: npx, args: [-y, modelcontextprotocol/server-filesystem, /tmp], transport: stdio, }, db: { command: python, args: [db_server.py], transport: stdio, timeout: 60, }, } ) all_tools [] for server_name, prefix in [(fs, fs), (db, db)]: tools await client.get_tools(server_nameserver_name) for tool in tools: tool.name f{prefix}_{tool.name} all_tools.append(tool) model ChatOpenAI(modelgpt-4o) agent create_react_agent(model, all_tools) async for event in agent.astream_events( {messages: [{role: user, content: 读取 /tmp/test.txt 的内容然后存入数据库}]}, versionv2, ): kind event[event] if kind on_chat_model_stream: content event[data][chunk].content if content: print(content, end, flushTrue) elif kind on_tool_start: print(f\n[调用工具: {event[name]}]) elif kind on_tool_end: print(f\n[工具完成]) if __name__ __main__: asyncio.run(main())这个示例可以直接跑起来前提是你有对应的 Server 实现和 API key。跑通之后你可以根据自己的需求替换 Server 和工具逐步扩展成更复杂的 Agent。我在实际使用中发现MCP 加 LangGraph 这套组合最大的价值在于标准化。以前每接一个新工具都要写一堆适配代码现在只要 Server 实现了 MCP 协议Client 侧几乎不用改代码就能用。当然协议本身还在演进有些细节比如错误码定义、流式语义还不够完善但整体方向是对的。如果你正在做 AI Agent 的工具集成这套方案值得花时间研究一下。