最近在做AI Agent落地项目遇到一个非常现实的问题项目里不止一个数据源和工具服务怎么让Agent在一个统一的协议下把多个服务都调起来而不是每个工具各写一套对接代码。最后我把方案落在了MCPModel Context Protocol上配合LangGraph做多Server调度把整个链路从协议握手到多节点调用完整打通了。这篇分享我不想只讲概念。MCP的文档和科普已经不少但真正动手接的时候坑基本都在细节里握手阶段的消息格式、capabilities协商、stdio和HTTP两种传输方式怎么选、多个Server的工具和资源如何隔离、LangGraph里应该把MCP调用放在哪个节点……这些才是决定项目能不能跑起来的关键。我尽量把整个过程的思考逻辑和踩坑记录都写出来给准备上MCP的团队一个参考。1. MCP解决了什么问题把工具调用从私有协议变成标准协议先说清楚MCP的定位不然容易把它和普通API网关搞混。MCP是Anthropic在2024年底开源的一套开放协议全称是Model Context Protocol它做的事情是给AI应用Host和外部工具/数据源Server之间定义一套标准化的通信方式。在MCP出现之前每个Agent要调用外部工具基本都得自己定义一套JSON格式的请求响应结构。比如你做一个代码审查Agent要读Git仓库、查JIRA、看CI日志就得分别对接三个系统的API每个系统还要处理认证、错误码、数据结构差异。更麻烦的是如果后面要换工具或者加一个新工具整个Agent的工具调用层又要改一遍。MCP的核心思路是把这层通信标准化。用一个类比来说HTTP协议让浏览器和Web服务器之间不用关心对方是什么技术栈MCP则是让AI应用和工具服务之间不用关心对方是什么框架。Host只认MCP协议Server只要实现MCP协议就能被接入中间的数据格式、调用方式、能力声明全部统一。在实际项目中MCP带来的直接好处很直观一套代码接所有工具。Agent侧只需要写一次MCP客户端逻辑后续加工具就是起一个新的MCP Server配置好连接信息Agent自动就能发现和使用它。工具能力的标准化描述。每个MCP Server启动时都会声明自己提供哪些工具Tool、哪些数据资源Resource、哪些提示模板PromptHost不用预先硬编码而是动态发现。打通本地与云端工具。MCP的传输层可以走stdio本地进程间通信也可以走Streamable HTTP远程服务一个协议同时覆盖本地脚本和远程API。我在这次项目里用MCP接了三类Server一个本地文件系统操作Server、一个Git操作Server、还有一个对接内网Wiki的远程Server。三个服务能力差异很大但接入LangGraph的方式完全一致代码复用率非常高。这就是协议标准化的价值。2. 协议握手全拆解initialize请求、能力协商与Initialized通知MCP的连接建立不像普通REST API那样直接发业务请求就行。它有一个完整的握手过程本质上是客户端和服务端互相确认你是谁、你能干什么、咱们按什么规则聊。2.1 握手的前置条件JSON-RPC 2.0 消息封装先看通信基础。MCP的所有消息都遵循JSON-RPC 2.0规范也就是每条消息都有一个固定的结构{ jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2024-11-05, capabilities: { roots: { listChanged: true }, sampling: {} }, clientInfo: { name: my-agent, version: 1.0.0 } } }这里需要注意的是jsonrpc字段必须固定是2.0id用于关联请求和响应。千万别小看这个基础格式我调试时就遇到过Server直接报Parse error原因是某个请求少带了jsonrpc字段——因为我在代码里封装消息时图省事把这个固定字段写死了没放进去。2.2 initialize第一次握手协商协议版本和能力集建立会话后客户端发出的第一条业务消息必须是initialize。这条消息的意思是我准备连接了我支持的协议版本是多少我具备哪些客户端能力请告诉我你的情况。服务端收到后会返回一条响应格式大致如下{ jsonrpc: 2.0, id: 1, result: { protocolVersion: 2024-11-05, capabilities: { tools: { listChanged: true }, resources: { subscribe: true, listChanged: true }, prompts: { listChanged: true } }, serverInfo: { name: filesystem-server, version: 0.1.0 } } }重点看protocolVersion。MCP的协议版本升级比较频繁早期有2024-10-07我这次用的是2024-11-05。握手时如果两端支持的版本不一致服务端通常会返回支持的最高版本客户端可以选择降级兼容或者直接断开。capabilities字段是握手阶段最需要仔细处理的。它声明了双方各自支持的功能范围服务端的capabilities有三类关键能力tools可调用工具、resources可读取的资源、prompts可用的提示模板。每个大类下面还可以有子能力比如listChanged表示当工具列表或资源列表变化时服务端会主动通知客户端。客户端的capabilities重要一点是roots它表示客户端允许服务端访问哪些本地目录或资源根路径。2.3 能力协商的实战细节别把能力声明当摆设握手阶段的能力协商不是走个过场它直接影响后面的调用行为。举个实际例子如果服务端没有声明resources能力客户端就不应该去调用resources/list否则服务端会返回Method not found。同样如果客户端没有声明roots能力服务端在请求访问本地文件时就不能依赖roots/list来确认可访问范围而只能靠自己配置的权限白名单。我最初犯过一个错在初始化时给客户端声明了roots能力但在实际请求时没有传对应的roots参数。结果服务端按握手时的约定直接尝试读取roots来校验路径权限导致文件操作全部失败。排查了很久才发现是握手时的能力声明与实际行为不一致。所以在设计MCP接入层的时候一个很重要的经验是能力声明一定要和实际请求逻辑保持一致。如果你不确定某个能力要不要声明宁可先不声明等确实需要了再加。因为MCP Server通常不会主动校验你一定实现了声明的能力但会基于你声明的能力来决定自己的行为。2.4 Initialized通知让Server知道我已经准备好initialize双向握手成功后客户端还需要发一条notifications/initialized通知。注意这条消息是通知不是请求所以它没有id字段服务端也不需要返回响应。{ jsonrpc: 2.0, method: notifications/initialized }这个通知的作用是告诉服务端握手阶段结束接下来我可以正常发送业务请求了。有的MCP Server库在实现时会等待这个通知后才开始接受tools/call等业务请求如果客户端发完initialize就直接调工具会遇到请求被挂起或直接拒绝。2.5 生命周期收尾正常关闭别硬断会话结束时客户端可以发送notifications/cancelled来取消正在进行的请求或者直接关闭底层连接。对于stdio传输关闭连接的方式就是结束子进程对于HTTP传输通常直接断开即可。但有一个细节值得注意如果Server端有正在执行的长时间任务比如一个正在跑的代码分析硬断连接会导致服务端留下孤儿进程。我在项目里加了优雅关闭逻辑Agent退出前先发一个取消通知再等2秒让Server清理资源最后才关闭连接。这一套下来本地跑几十个Server进程也不会残留僵尸进程。3. 传输方式选型stdio还是Streamable HTTP不能只看文档MCP支持的传输方式官方文档里主要提供两种stdio和Streamable HTTP。选哪一种直接决定了你的部署形态和性能表现。3.1 stdio进程即服务适合本地工具链stdio模式的本质是MCP Server作为一个子进程被客户端拉起双方通过标准输入输出流来传递JSON-RPC消息。在这种模式下通信参数不通过网络传递而是在进程启动时通过命令行参数或环境变量注入。比如我启动文件系统Servernpx -y modelcontextprotocol/server-filesystem /path/to/allowed/dir客户端收到这个启动命令后负责创建子进程然后把JSON-RPC消息写入子进程的stdin从子进程的stdout读取响应。stdio模式的优点很明显无需额外起服务进程、无需配置端口、安全性好工具与进程边界天然隔离。缺点是只能在本地使用且每次连接都需要拉起一个进程有启动开销。3.2 Streamable HTTP远程调用按URL连接当Server部署在远程机器上时就需要走Streamable HTTP。这种模式的特点是服务端是一个HTTP端点客户端通过标准HTTP POST请求来发送JSON-RPC消息。启动远程MCP Server后会得到一个HTTP端点比如http://internal-wiki-server:8080/mcp。客户端连接时传入这个URL并通过HTTP协议完成握手和后续调用。Streamable HTTP的注意点不少最头疼的是会话管理。某些实现中服务端会要求客户端在initialize请求中携带一个会话ID后续所有请求都必须带上这个ID否则服务端会拒绝。这个设计类似于登录鉴权里的Token机制。我踩过的坑是用Python的requests库连续发请求时因为封装层没有自动携带会话头导致每次请求都被当成新会话服务端状态全部丢失。3.3 我的选型思路本地优先、远程按需在LangGraph项目的实际架构里我的原则是本地工具文件操作、Git命令、本地脚本执行一律用stdio。原因很简单延迟低、免部署、环境隔离。而且LangGraph Agent跑在本地时直接拉起子进程的开销可以接受。远程服务内网Wiki、云端API、数据库信息查询用Streamable HTTP。这样Server可以部署在独立的服务器上资源占用不影响Agent主进程。选型时还有个容易被忽略的点混合部署。同一个Agent里多个Server可以一部分走stdio、一部分走HTTP两者并不冲突。LangGraph层面感知不到传输层的差异它只是通过统一的MCP客户端接口来调用工具。4. LangGraph里接多Server从单工具节点到多服务并行调度接下来是重头戏怎么在LangGraph的Agent架构里同时调用多个MCP Server。4.1 基础架构MCP Client放在Agent的Tool节点里LangGraph的核心设计是图Graph节点Node是执行单元边Edge是流转逻辑。在一个典型的ReAct风格的Agent里节点通常包括Agent节点负责和大模型交互、决定下一步动作和Tool节点负责执行工具调用。MCP Client的接入位置就是Tool节点。具体做法是在Agent初始化时创建多个MCP Client分别连接到不同的Server。每个Client启动后通过list_tools()方法拉取工具列表把这些工具包装成LangChain的BaseTool格式。把所有工具合并后绑定给Agent节点的大模型让模型在规划阶段就能看到全部可用工具。模型决定调用某个工具时LangGraph会把调用请求派发给对应的Tool节点由Tool节点找到匹配的MCP客户端执行调用。这个架构的关键在于工具到客户端的映射。合并工具列表后每个工具必须知道自己的归属Server。我用的是在工具对象上挂元信息的方式给每个工具打了一个server_name标签Tool节点根据标签路由到对应的MCP Client。4.2 多Server初始化与工具合并下面是一段简化版的初始化逻辑展示如何连接多个Server并合并工具列表import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools async def connect_server(server_name: str, command: str, args: list[str]): server_params StdioServerParameters(commandcommand, argsargs) client stdio_client(server_params) read, write await client.__aenter__() session ClientSession(read, write) await session.__aenter__() await session.initialize() tools await load_mcp_tools(session) for tool in tools: tool.server_name server_name return session, tools async def init_all_servers(): all_tools [] sessions [] # 文件系统Server fs_session, fs_tools await connect_server( filesystem, npx, [-y, modelcontextprotocol/server-filesystem, /tmp/workspace] ) sessions.append(fs_session) all_tools.extend(fs_tools) # Git操作Server git_session, git_tools await connect_server( git, npx, [-y, modelcontextprotocol/server-git, --repo, /tmp/workspace/repo] ) sessions.append(git_session) all_tools.extend(git_tools) return sessions, all_tools注意这里比较关键的一步是load_mcp_tools。它会把MCP Server返回的JSON Schema格式的工具定义自动转换成LangChain Tool对象。转换后工具的参数校验和调用逻辑都可以直接复用LangChain的标准接口。4.3 LangGraph状态管理与工具调用结果回传LangGraph的状态State管理在多工具调用场景下非常重要。MCP Server返回的结果可能是结构化数据JSON、文本也可能是资源文件路径比如Git操作生成的diff文件。我的状态设计思路是messages保存Agent与大模型交互的完整消息记录包括工具调用结果。tool_results一个字典按tool_call_id存储每个MCP工具调用的原始返回结果方便后续节点追溯。active_server当前正在执行操作的Server名称用于日志追踪和故障定位。工具调用完成后结果要格式化为ToolMessage写回状态否则Agent节点的大模型看不到执行结果就无法继续推理。这里有个实用技巧MCP Server返回的内容有时是JSON字符串LangChain的ToolMessage不能直接塞JSON对象要先转成字符串。但如果内容是文件路径或超长文本最好做截断处理否则LLM上下文很快被撑爆。4.4 多Server调用的实际运行效果在LangGraph里跑多Server的流程大致是用户提问 - Agent节点规划 - 决定调用哪个工具 - 派发对应Server执行 - 结果回传 - Agent继续判断是否还需要其他服务。比如这样一个任务检查工作区里的代码变更统计新增了哪些函数并生成一份变更摘要。Agent可能会先调用Git Server的git_diff工具拿变更列表再调用文件系统Server的read_file工具读取具体文件内容最后调用Python代码解析工具统计函数。整个过程需要多个Server协作LangGraph通过图节点的多次循环来实现。这里LangGraph的REPLReAct循环机制很关键它允许Agent节点在大模型判断还需要调用工具时多次触发Tool节点形成循环。每轮循环都会更新状态直到大模型认为信息足够、给出最终回答。5. 多Server场景下的真实踩坑工具名冲突、超时与上下文膨胀理论讲完这里集中记录几个我在实际项目中反复踩的坑。每个都是线上跑出来才发现的问题很有代表性。5.1 工具名冲突两个Server的同名工具互相覆盖不同的MCP Server很可能导出同名的工具。比如文件系统Server和Git Server都有一个list_files工具合并工具列表时后面的会把前面的覆盖掉结果模型调用list_files时指向了错误的Server。解决方式是给工具名加前缀类似于命名空间for tool in tools: tool.name f{server_name}__{tool.name}这样文件系统的工具变成filesystem__list_filesGit的工具变成git__list_files互不干扰。对应的在Agent节点提示词里要明确告知模型工具的全名否则模型根据语义可能只猜出半截名字。5.2 初始化握手超时Server启动慢导致连接失败stdio模式的Server启动时间波动很大。npx第一次拉包可能要几十秒后续启动才快。而MCP客户端默认握手超时通常是10秒很容易在第一次运行时直接超时。我的解决方案是在连接层做重试和超时配置async def connect_with_retry(connect_func, retries3, timeout30): for attempt in range(retries): try: return await asyncio.wait_for(connect_func(), timeouttimeout) except asyncio.TimeoutError: print(fConnection attempt {attempt 1} timed out) if attempt retries - 1: raise另外建议在项目文档里明确首次启动时先手动跑一遍npx拉取依赖不要让用户消耗首次握手超时的体验。5.3 上下文膨胀工具结果太大把LLM窗口挤爆MCP工具返回的内容有时非常大。比如Git Server返回的diff可能有几千行文件系统Server读取的大文件可能有上百万字符。LangGraph每轮循环都会把工具结果写进messages状态如果不对结果做裁剪多轮循环后上下文会指数级膨胀。我在工具结果回传前加了一层处理如果返回内容超过一定阈值比如5000字符只保留前2000字符和后1000字符中间用省略标记替代同时把完整结果存到tool_results字典里。这样LLM只看到摘要需要完整内容时再通过专门的工具按需读取。这一改动直接把Agent的可用轮次从3-4轮提升到10轮以上。5.4 错误处理的粗粒度与细粒度多Server调用时错误处理要注意粒度。最怕的是把所有Server的调用包在一个大try-except里一旦某个Server挂了整个Agent流程就断了。合理的做法是每个Server连接失败时先标记为不可用从工具列表中移除对应工具而不是直接让整个Agent崩溃。工具执行失败时把错误信息作为ToolMessage回传给LLM让它决定是重试、换方案还是告诉用户失败原因。LangGraph节点层面做兜底如果某轮循环所有工具都调用失败直接终止ReAct循环并输出错误说明避免死循环。def tool_node(state): tool_calls state[messages][-1].tool_calls results [] for call in tool_calls: try: result execute_tool(call[name], call[args]) results.append(ToolMessage(contentresult, tool_call_idcall[id])) except Exception as e: error_msg fTool {call[name]} failed: {str(e)} results.append(ToolMessage(contenterror_msg, tool_call_idcall[id])) return {messages: results}5.5 会话生命周期管理如果Agent是长期运行的服务比如一个常驻的聊天后端MCP连接不能每次请求都重新握手。一方面是握手开销大另一方面是某些Server的状态比如Git的repo指针、Wiki的登录态需要持续保持。我的做法是用一个全局的连接池按Server名称缓存ClientSession实例。同时加心跳机制定时发送ping请求保活。对于偶尔出现的连接断开实现自动重连逻辑。6. 进一步优化MCP的Resources与Prompts在LangGraph中的使用思路MCP不只是工具调用。它的Resources和Prompts能力在多Server场景下也值得利用起来。6.1 Resources让Server主动暴露上下文Resources定义的是可以被读取的数据源比如配置文件、数据库Schema、项目文档。和工具不同Resources不需要LLM决定调用而是可以由Host按需拉取。在LangGraph里我通常把Resources当成Agent启动时的上下文预加载器。比如连接Wiki Server后在Agent启动时读取一组常用文档资源把它们压缩后直接作为System Prompt的一部分。这比等用户提问再去拉文档更快也能减少工具调用轮次。6.2 Prompts把提示词工程下放到Server端MCP的Prompts是一种可复用的提示词模板Server可以定义analyze_code、generate_report这类标准Prompt。LangGraph Agent在特定节点需要特定类型的输出时可以直接从Server拉取对应Prompt并填充。这个方法的价值在于提示词和工具逻辑不再散落在Agent代码里而是跟着Server走。换个团队、换套Agent框架只要对接同一个MCP ServerPrompt行为保持一致。对于需要严格统一输出格式的团队项目非常实用。6.3 把MCP调用封装成LangGraph的子图如果项目比较大可以考虑把MCP工具调用封装成LangGraph的子图而不是简单的节点函数。子图的好处是可以独立定义状态、支持条件路由、方便复用。我当前的做法是每个MCP Server一套子图子图内部包含连接检查 - 工具分类 - 执行调用 - 结果格式化几个步骤。多个子图通过LangGraph的父图编排根据任务路由到不同子图。这比把所有逻辑堆在一个Tool节点里清晰得多也更容易扩展新的Server。这种子图方案的注意点是子图和父图的状态需要明确定义接口。MCP执行的结果要标准化成统一的结构比如带content_type和content字段否则父图下游的Agent节点拿到什么都得自己猜。7. 动手落地一个最小多Server LangGraph示例最后给一个可以直接跑的最小示例覆盖多Server初始化、工具合并、LangGraph节点编排的完整链路。import asyncio from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage from typing import TypedDict, Annotated from langchain_openai import ChatOpenAI from langchain_mcp_adapters.tools import load_mcp_tools from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client class AgentState(TypedDict): messages: list tool_results: dict # 1. 初始化MCP Server连接 async def init_mcp_server(command, args): server_params StdioServerParameters(commandcommand, argsargs) client stdio_client(server_params) read, write await client.__aenter__() session ClientSession(read, write) await session.__aenter__() await session.initialize() tools await load_mcp_tools(session) return session, tools async def setup(): fs_session, fs_tools await init_mcp_server( npx, [-y, modelcontextprotocol/server-filesystem, /tmp/workspace] ) git_session, git_tools await init_mcp_server( npx, [-y, modelcontextprotocol/server-git, --repo, /tmp/workspace/repo] ) # 工具名加前缀防止冲突 for tool in fs_tools: tool.name ffs__{tool.name} for tool in git_tools: tool.name fgit__{tool.name} all_tools fs_tools git_tools return all_tools # 2. 定义Agent节点 def agent_node(state: AgentState): llm ChatOpenAI(modelgpt-4o, temperature0) llm_with_tools llm.bind_tools(all_tools) response llm_with_tools.invoke(state[messages]) return {messages: [AIMessage(contentresponse.content, tool_callsresponse.tool_calls)]} # 3. 定义工具执行节点 def tool_node(state: AgentState): tool_calls state[messages][-1].tool_calls results [] for call in tool_calls: tool next(t for t in all_tools if t.name call[name]) result tool.invoke(call[args]) results.append(ToolMessage(contentstr(result), tool_call_idcall[id])) return {messages: results} # 4. 条件路由有工具调用就进工具节点否则结束 def should_continue(state: AgentState): last_message state[messages][-1] if last_message.tool_calls: return tools return end # 5. 构建LangGraph async def main(): global all_tools all_tools await setup() workflow StateGraph(AgentState) workflow.add_node(agent, agent_node) workflow.add_node(tools, tool_node) workflow.set_entry_point(agent) workflow.add_conditional_edges(agent, should_continue, {tools: tools, end: END}) workflow.add_edge(tools, agent) app workflow.compile() result await app.ainvoke({ messages: [HumanMessage(content统计工作区代码变更输出变更文件列表)], tool_results: {} }) print(result[messages][-1].content) asyncio.run(main())这段代码就是完整的多Server Agent骨架跑通之后按需加节点就行。我在项目里就是从这个版本一路扩展出来的。最后说一句实际操作的体会MCP LangGraph这套组合最大的价值不在于某一个功能多炫而是把工具的接入成本和切换成本降下来了。之前接一个工具要写几百行胶水代码现在无非是配一行Server启动命令。踩了几次坑之后我现在接新工具已经基本形成套路了。如果团队正在做多工具Agent建议直接上MCP早用早省心。