从“连接失败”到“多Server编排”，我最近在项目里把MCP和LangGraph彻底跑通了一遍，踩了不少坑，也把协议握手那层窗户纸捅破了。这篇就把从协议层到应用层的完整链路写出来，重点说清MCP初始化握手的每个细节，以及如何在LangGraph里同时管理、调用多个MCP Server，不绕弯子，全是实操经验。
先说结论：MCP不是又一个API网关，它是AI应用和外部工具之间的“标准化插槽”。你不需要针对每个工具写一套私有适配逻辑，而是让工具实现统一的MCP协议，大模型就能像调函数一样去调用它们。而LangGraph这种编排框架，又给了你一个灵巧的“流程骨骼”，把多个MCP Server变成Graph里的不同工具节点。两个东西结合，能做出很灵活的Agent系统。这篇适合已经用过LangChain/LangGraph、想接入MCP但还没完全搞懂握手细节的开发者，也适合刚接触MCP、想从零搭一套多Server调用的新手。
1. 从一次“连接失败”说起：MCP到底是什么
1.1 MCP不是API网关，而是“AI应用的USB接口”
很多朋友第一次看到MCP（Model Context Protocol）会误以为它是另一个RPC框架或者API网关。但它的定位完全不同：API网关是把外部接口统一转发给客户端，而MCP是定义了大模型应用与外部“上下文工具”之间的通信协议。类比一下，USB接口的意义不在于传输速度，而在于统一了设备连接标准——鼠标、键盘、U盘都能插同一个口。MCP就是给AI应用开的这个“口”：数据库、文件系统、浏览器自动化、代码分析器、SQL Server，只要能实现MCP Server，就能被Agent直接调用。
实际项目里我见过最典型的困境：Agent需要同时查数据库、操作浏览器、调本地脚本，以前要给LangChain写三个自定义Tool，每个Tool都要处理鉴权、输入输出校验、错误重试。用MCP之后，这些Server各自实现tools/list和tools/call，LangGraph侧只负责注册和调用，管道化、解耦化都清晰得多。
1.2 MCP协议的核心构件：client、server、transport
MCP协议里有三个关键角色：MCP Client（通常运行在Agent/LLM应用侧）、MCP Server（暴露工具/资源/提示词的进程或服务）、Transport（通信方式）。当前主流Transport有两种：stdio和Streamable HTTP。stdio适合本地进程，Server和Client在同一个机器上，通过标准输入输出交换JSON-RPC消息；Streamable HTTP适合远程Server，走HTTP+SSE，比如你在服务器上部署一个MCP服务，供本地Agent远程调用。
我在LangGraph里两种都试过。本地调试用stdio最顺手，不担心端口冲突；部署到云上以后改成HTTP，因为多个Agent实例可能要共享同一个Server。选Transport的建议是：先搞清楚你的Server是“进程内工具”还是“独立服务”。如果是前者，stdio的启动和销毁成本更低；如果是后者，HTTP是必然选择。另外，MCP的消息格式严格遵循JSON-RPC 2.0，这意味着每个请求必须有id、method、params，响应要么是result要么是error，没有第三种形态。
2. 协议握手：MCP初始化到底发生了什么
2.1 握手顺序：initialize → initialized → tools/list → tools/call
这部分是我这次分享最想强调的。很多人调用MCP失败，根本原因是没按握手顺序来。MCP的会话建立不是一个TCP连接那么简单，它是一个带“能力协商”的JSON-RPC会话流程，完整顺序如下：
- Client发送
initialize请求，带上自己支持的protocolVersion、capabilities和clientInfo。 - Server返回
initialize响应，带上Server的protocolVersion、capabilities、serverInfo和instructions。 - Client发送
notifications/initialized通知，告诉Server“我确认初始化完成”。 - 之后Client才能发送
tools/list获取工具列表，再通过tools/call调用具体工具。
这个顺序不能乱。我见过有人跳过notifications/initialized直接发tools/list，某些严格实现的Server会直接关闭连接或返回错误。协议设计者的意图很明确：先用initialize做“能力协商”，双方确认版本兼容、支持哪些特性，再进入业务调用阶段。
下面我用Python的mcp库写一个最小Client端握手示意，让你直观感受消息往返：
import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def handshake_demo(): # 启动一个本地 MCP Server 进程（stdio） server_params = StdioServerParameters( command="python", args=["my_mcp_server.py"] ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # 第1步：initialize init_result = await session.initialize() print("协商版本:", init_result.protocolVersion) print("Server能力:", init_result.capabilities) # 第2步和第3步：notifications/initialized 由 ClientSession 内部自动发送 # 第4步：列出工具 tools_result = await session.list_tools() for tool in tools_result.tools: print("可用工具:", tool.name, "->", tool.description) # 调用工具 call_result = await session.call_tool( name="echo", arguments={"message": "hello mcp"} ) print("调用结果:", call_result.content) asyncio.run(handshake_demo())如果你用的是langchain-mcp-adapters这类库，握手逻辑已经被封装好了，但理解底层依旧很重要——排查问题的时候，你要能分辨是“握手失败”还是“业务调用失败”。
2.2 用一个最小服务端看握手细节
很多人只写Client，不看Server，导致对协议的理解缺了另一半。我建议你至少写一次MCP Server，哪怕只暴露一个echo工具。原因很简单：只有站在Server视角，你才知道initialize响应里的capabilities是怎么影响后续行为的。
下面是一个符合MCP规范的最小Server（stdio模式），用mcp.server.fastmcp的快捷方式：
from mcp.server.fastmcp import FastMCP mcp = FastMCP("DemoServer") @mcp.tool() def echo(message: str) -> str: """原样返回输入消息""" return f"echo: {message}" if __name__ == "__main__": mcp.run(transport="stdio")FastMCP会自动处理initialize协商、tools/list反射、tools/call分发。但如果你去看它生成的响应，会发现capabilities里包含tools字段，serverInfo包含名称和版本。协议版本则由双方协商取交集——如果Client只支持某个低版本，而Server只支持高版本，握手就会失败。所以你需要在代码里显式控制protocolVersion的兼容范围，而不是默认写死。
2.3 握手中的超时与心跳：别忽略protocolVersion
一个容易踩的坑是超时。MCP的握手不是本地函数调用，尤其Streamable HTTP模式下，Server可能因为负载高、网络抖动导致initialize响应超过了Client的超时阈值。我一开始用的是默认超时，结果在调用一个冷启动的Server时频繁报错“Connection closed”。后来把超时调到30秒，并在Server端做了预热，问题就消失了。
再说protocolVersion协商。有些MCP库更新很快，不同版本之间协议有细微变化。如果你在initialize里声明了一个很高的版本，但Server不支持，返回的protocolVersion会和你的不一致。此时一定要以Server返回的版本为准，并决定是否继续会话。我在一个旧版Server上遇到过这个问题：它能完成握手，但之后发tools/list会挂起，后来发现是版本不匹配导致的消息格式差异。稳妥做法是：Client和Server都固定一个已知兼容的版本号，比如"2024-11-05"，不要盲目用最新。
3. LangGraph 多 Server 调用：架构与设计
3.1 为什么要在LangGraph里接多个MCP Server
LangGraph本身是编排Agent工作流的框架，它把任务拆成节点（Node）和边（Edge），节点之间传递状态（State）。而MCP Server提供的是工具。两者结合，意味着你可以在Graph的某个节点里调用多个外部服务，比如：
- Server A（Playwright MCP）负责浏览器自动化，完成网页截图、表单填充。
- Server B（SQL Server MCP）负责查询业务数据库。
- Server C（文件系统MCP）负责读写本地文件。
一个典型的“数据采集Agent”可能是这样的流程：先用Playwright MCP打开目标页面，抽取数据；然后把数据清洗后交给SQL Server MCP写入数据库；最后用文件系统MCP生成一份报告。如果不用MCP，这三个工具各自需要一套调用方式；统一成MCP后，LangGraph里所有节点看到的都是“工具名+参数JSON”，这大大简化了状态流转设计。
不过“多Server”也意味着新的复杂性：多个Server的工具命名冲突、连接生命周期管理、错误隔离。这些必须在架构设计阶段就定好，否则写到一半会非常痛苦。
3.2 多Server调用的两种模式：顺序串联、并行路由
从Graph拓扑角度看，多Server调用可以抽象成两种模式：
顺序串联模式：一个节点的输出作为下一个节点调用的输入。比如“打开页面 -> 提取数据 -> 写入数据库”。在LangGraph里这对应一条线性边，每个节点内部只调用一个或一组Server。
并行路由模式：一个节点同时向多个Server发起调用，收集所有结果后聚合。比如“同时问SQL Server和文件系统MCP，找出业务数据和历史报告”。这种模式适合把独立的查询任务并行化，显著降低响应延迟。
在LangGraph里实现并行路由，通常用SendAPI或者把多个异步调用放进同一个节点。我自己更倾向于后者：在节点函数里用asyncio.gather同时调用多个tool，把结果汇总成一个状态块。这样逻辑更直观，也方便控制单个调用的超时。
要注意的是，LangGraph的State是链式传递的，但多Server调用的中间结果不一定都要塞进顶层State。你可以设计一个局部State或者在节点内部维护局部变量，只把最终需要传递给下游的结果写入全局State。否则State会越来越臃肿，最终影响序列化和内存占用。
3.3 连接管理与命名空间隔离
多Server连接管理是我认为最容易被低估的一块。每个MCP Server都是一个独立连接（stdio是独立进程，HTTP是独立会话），如果Graph里每个节点都新建连接、用完不关，很快会耗尽文件描述符或端口。
我建议在应用启动时统一初始化所有MCP Server连接，把连接句柄存在一个全局对象里，LangGraph节点通过工具名查找对应的Server。另外要给工具名加命名空间前缀，例如playwright_navigate、sql_query、file_read，避免不同Server暴露同名工具（比如都有read或write）时产生歧义。这个方案在MultiServerMCPClient里有个参数可以指定前缀，后面实操部分会演示。
还有错误隔离。假如SQL Server MCP挂了，不能让整个Graph崩溃。在每个节点调用工具时，用try/except捕获异常，并返回一个带有错误标记的结果，让LangGraph的路由逻辑决定是重试、降级，还是终止。多Server场景下，一个服务故障不应该拖垮整个Agent。
4. 实操：从零实现 LangGraph 调用两个 MCP Server
4.1 准备工作：安装依赖与启动本地服务
我们先搭建一个最小实验环境。你需要安装以下Python库：
pip install langgraph langchain-mcp-adapters mcplangchain-mcp-adapters是目前把MCP工具转换成LangChain Tool最方便的方式，内置了MultiServerMCPClient。我实验用的两个Server：一个是本地的DemoServer（stdio），另一个是远程的Playwright MCP（Streamable HTTP）。你可以先用任意两个MCP Server替代。
启动本地DemoServer，写文件demo_server.py：
from mcp.server.fastmcp import FastMCP mcp = FastMCP("DemoServer") @mcp.tool() def add(a: int, b: int) -> int: """两数相加""" return a + b @mcp.tool() def get_current_time() -> str: """返回当前时间字符串（模拟）""" from datetime import datetime return datetime.now().isoformat() if __name__ == "__main__": mcp.run(transport="stdio")另一个Server我用了远程的Playwright MCP（比如npx @playwright/mcp@latest），它会启动一个HTTP服务。你需要确认地址和端口，后面配置要填。
4.2 代码实现：动态加载多MCP Server
下面是核心代码，用MultiServerMCPClient同时加载两个Server：
import asyncio from langchain_mcp_adapters.client import MultiServerMCPClient from langgraph.graph import StateGraph, MessagesState from langgraph.prebuilt import create_agent async def build_mcp_tools(): # 本地stdio Server，通过命令行启动 # 远程HTTP Server，需要配置url和transport client = MultiServerMCPClient( { "demo": { "command": "python", "args": ["demo_server.py"], "transport": "stdio", }, "playwright": { "url": "http://127.0.0.1:8931/mcp", "transport": "streamable_http", "headers": {"Authorization": "Bearer your_token"}, } } ) tools = await client.get_tools() return tools, client注意，MultiServerMCPClient会自动为每个Server的工具加上命名空间前缀，比如demo_add、demo_get_current_time、playwright_navigate等，这样Graph里不会有歧义。
如果你要更细粒度地控制连接生命周期，也可以分别创建ClientSession再手动转换工具，但日常开发用这个Client足够了。我在生产环境里还做过一个变体：从配置文件读取Server列表，循环注册，这样扩展Server只需要改配置，不用改代码。
4.3 在Graph中编排调用链
有了tools，就可以构造LangGraph Agent了。我这次没有用复杂的自定义节点，而是用create_agent快速实现一个会自主调用工具的Agent：
async def run_agent(): tools, client = await build_mcp_tools() agent = create_agent( "openai:gpt-4o", # 替换成你实际可用的模型 tools=tools, state=MessagesState, ) result = await agent.ainvoke( {"messages": [("user", "请先调用demo的add工具计算 3+5，再调用playwright打开百度首页并把标题告诉我。")]} ) print(result["messages"][-1].content) # 关闭所有MCP连接 await client.__aexit__(None, None, None)实际上create_agent内部已经做好了工具调用的循环，Agent会根据任务描述自动选择工具。这里有两个关键点：
- 模型的工具调用能力要好，否则不会自动把参数转换成JSON Schema要求的格式。
- 多个工具的调用顺序由模型决定，但如果你想强制“先A后B”，最好显式写一个自定义节点，而不是让Agent自由发挥。
强制顺序的写法也很简单，比如：
from langgraph.graph import START, END def call_demo(state): # 假设已有会话对象 ... def call_playwright(state): ... graph_builder = StateGraph(MessagesState) graph_builder.add_node("demo_step", call_demo) graph_builder.add_node("playwright_step", call_playwright) graph_builder.add_edge(START, "demo_step") graph_builder.add_edge("demo_step", "playwright_step") graph_builder.add_edge("playwright_step", END)此时每个节点内部可以访问全局的MCP连接池，再通过asyncio.gather做并行调用的扩展。这个方案比完全交给Agent更可控，也更适合对流程有严格要求的业务场景。
4.4 参数与结果流转：别让JSON Schema坑了你
MCP Server的工具定义依赖于JSON Schema，但很多工具函数的参数类型/描述写得随意，导致模型生成的参数经常校验失败。我遇到过几个印象深刻的坑：
第一个坑是枚举值。比如工具要求mode只能是"fast"或"slow"，但描述里没写清楚，模型生成了"quick"，Server直接报错。解决办法是在函数docstring里明确给出可选值，并在Schema里加enum。
第二个坑是整数和字符串的混淆。a和b明明是整数，模型有时候会传字符串"3"，某些严格校验的库会拒绝。我一般会在Server端做一次宽松类型转换，比如int(a)。
第三个坑是结果太长导致上下文爆炸。MCP工具返回的内容可能是一大段网页文本或整个文件内容，如果直接塞进LangGraph的State，会大幅增加token消耗。建议在Server端就做截断、摘要或结构化压缩，只返回关键信息。比如SQL Server MCP返回查询结果时，加一个limit参数限制行数；文件系统MCP返回文件时，可以支持max_chars参数。
@mcp.tool() def query_users(limit: int = 10) -> str: """查询用户列表，默认只返回10行""" rows = get_users(limit=limit) return summarize(rows)这个“裁剪返回结果”的思路在多Server调用里尤其关键，不然Agent在多个工具间切换时，上下文会被无用信息填满，导致后续决策质量断崖式下降。
5. 常见问题与排查技巧实录
5.1 连接失败：initialize握手没完成就发请求
这是我在社区看到最多的问题。症状是调用tools/call时返回Connection closed或No response from server。根因大多是握手时序错误，尤其是用了自己封装Client的代码，忘记发送notifications/initialized。
排查技巧：
- 开启Debug日志，观察是否出现
initialize的请求和响应。如果只有请求没有响应，多半是Server进程没起来或端口不通。 - 如果是stdio模式，用命令行手动启动Server，看是否有报错输出。
- 如果是HTTP模式，用
curl先手动发一次initialize请求，确认Server的HTTP路径和请求格式无误。
我还遇到过一种特殊情况：某些HTTP Server需要自定义Header（如Authorization），你如果只传了URL没传Header，Server在握手阶段就会拒绝连接。所以检查Header配置永远在第一位。
5.2 调用超时：工具响应慢导致LangGraph整个任务卡死
LangGraph中如果某个MCP工具调用时间很长，比如Playwright打开一个重度页面要十几秒，Agent默认的交互超时可能不够。最终表现是整个任务一直转圈，没有任何错误返回。
解决方案有三个层级：
- 在Server端设置幂等和快速失败，比如给耗时操作加一个内部超时。
- 在Client端调用
call_tool时传入timeout参数，避免无限等待。 - 在LangGraph节点内部用
asyncio.wait_for包裹工具调用，超时后返回预设的错误消息，让Agent选择备选路径。
import asyncio async def safe_call_tool(tool, **kwargs): try: return await asyncio.wait_for(tool.ainvoke(kwargs), timeout=15) except asyncio.TimeoutError: return "TOOL_TIMEOUT"这样即使某个Server出问题，整个Graph也能继续走下去，而不是卡死。
5.3 工具名冲突：多个Server都有同名工具
前面提过命名空间前缀，这是最推荐的做法。但如果你从数据库配置文件里加载工具列表，可能会遇到前缀叠加错误，比如playwright_playwright_navigate。这种情况通常是因为在Server名和工具名里都包含了前缀字段。写配置时一定要确认MultiServerMCPClient的tool_name_prefix是全局配置还是每Server配置。
如果用的不是这个库，而是自己写工具注册表，建议在注册时用server_name + "__" + tool_name这种双下划线分隔，解析时再split，避免不同Server名字类似时产生歧义。
另外，LangGraph的create_agent对重复工具名是会报错的，它会要求所有工具名唯一。所以即使你觉得“同名没关系”，框架层面就不允许，必须在注册阶段解决。
5.4 认证与权限：多Server下的token管理
多个远程MCP Server通常对应不同的认证方式：有的用Bearer Token，有的用API Key，有的用局域网白名单。把它们统一写在Java配置文件里是可行的，但要注意安全性，别把token提交到代码仓库。
我推荐的做法是使用环境变量或本地的密钥管理服务，在启动Agent时读取并注入到连接配置中。尤其当你有多个Server，且不同Server的token权限范围不同时，最好为每个Server创建独立的只读或最小权限token，而不是用一个超级账号通吃。权限过度不仅危险，而且会在工具误操作时造成更大的破坏。
这里给一个参考配置示例（不要真用硬编码token）：
import os def load_server_config(): return { "playwright": { "url": os.getenv("PLAYWRIGHT_MCP_URL"), "transport": "streamable_http", "headers": { "Authorization": f"Bearer {os.getenv('PLAYWRIGHT_TOKEN')}" } }, "sqlserver": { "url": os.getenv("SQL_MCP_URL"), "transport": "streamable_http", "headers": { "X-API-Key": os.getenv("SQL_API_KEY") } } }多Server的认证策略尽量简单统一，能走同一个SSO网关就走网关，能在网络层隔离就在网络层隔离，不要让Agent代码里到处都是token分支判断。
6. 经验总结：这套架构还能怎么扩展
最后分享一点个人折腾下来的体会。
MCP + LangGraph的最佳组合不是“无脑让大模型自己选工具”，而是“把流程节点化，把工具协议化”。你完全可以把Playwright MCP、SQL Server MCP、IDA MCP、FastAPI/ollama这类本地模型服务全部注册进来，让LangGraph作为“总指挥”。但前提是你对每个Server的能力边界、超时行为、返回数据规模都心里有数。我在实际项目里习惯给每个MCP Server写一个“能力卡片”，写明工具列表、平均响应时间、最大返回长度、推荐调用场景。这个卡片不进代码，而是放在团队Wiki里，这样后来者接新Server时少踩很多坑。
另一个建议是，尽量让MCP Server做“一次性返回完整结果”的事情，而不是“流式半成品”。虽然MCP支持流式调用，但在LangGraph的状态模型里，流式结果需要额外处理。如果任务需要流式输出，不如让MCP Server把数据写到某个中间存储（比如Redis或本地文件），然后返回一个“任务已生成，结果在xxx”的字符串，再由另一个工具去读取。这样虽然多了一步，但状态流转更清晰，排查问题也更方便。
还有个小技巧：在开发阶段，给每个MCP工具名加一个debug_前缀，部署时再替换。这样你可以在生产环境旁边的测试环境里反复调用而不会污染真实工具索引。我就是靠这个方式，在切换Server版本时快速验证新老工具行为是否一致。
这篇文章没有急着让你立刻上生产。MCP协议还在快速演进，工具链也在变化，但核心的握手时序、连接生命周期、多Server命名隔离和错误处理思路是稳定的。把这些地基打好，后面无论LangGraph更新到哪个版本，你都能快速适配。
