这段时间我一直在折腾一个 AI 智能体项目让 LangGraph 驱动的 Agent 同时去调内部工单系统、业务数据库和一个第三方天气服务。第一版我老老实实给每个系统写工具函数结果工具函数越写越胖鉴权、参数校验、错误处理全揉在一起换个系统就要大改。后来我把思路切到 MCPModel Context Protocol上用协议握手的方式把各个能力注册成标准 Server再统一接进 LangGraph 做多 Server 调用。这条链路从协议层到应用层完整跑通之后我的体会是MCP 真正的价值不是省事而是把工具的定义、发现、调用、鉴权这些横切逻辑从业务代码里彻底剥了出去。这篇文章不打算讲 PPT 式概念就记录我从协议握手、服务端实现到 LangGraph 多 Server 编排的完整过程包括我踩过的坑、看过的原始协议报文、以及最后总结出的生产落地建议。1. 为什么把 MCP 塞进 LangGraph一次选型复盘1.1 MCP 到底是什么一句话讲清楚MCP 是一个基于 JSON-RPC 2.0 的开放协议它的核心套路是把能力工具、资源、提示词做成可以被标准方式发现和调用的服务。你的 Agent 不需要提前知道某个系统有哪些接口只需要在通信时通过tools/list问一句你支持什么对方返回一份带 JSON Schema 的清单Agent 再按照这个 Schema 组织参数去调用。整个过程跟 USB 设备即插即用的逻辑很像——协议约定好了插上就能用。现在市面上大多数主流 AI 客户端和框架都在往这条路上靠。我项目里用到的 LangGraph 本身没有绑定 MCP但它可以用标准工具方式接入任何 MCP Server 暴露的工具。我在实际项目中同时跑了三个 Server一个负责订单查询一个负责工单流转还有一个负责外部天气和地图信息。每一次接入我都只需要关心这个 Server 的 URL 和鉴权方式业务侧的工具定义完全变成配置数据。1.2 和直接写工具函数相比MCP 解决了什么有人可能会问LangGraph 里直接写 Python 函数做工具再用bind_tools绑给模型不也挺好吗我在第一版也是这么干的但等系统扩到五六个外部依赖之后就发现问题了。直接写函数方案下每一个工具函数要处理连接外部系统的 SDK、token 刷新、超时重试、异常包装代码量会膨胀到完全不可控。更麻烦的是工具的输入输出 Schema 是散落在函数签名里的LLM 能不能正确生成参数完全依赖你写 docstring 时的心情。MCP 方案下这些规则全部由协议和 Server 端统一管理工具的 Schema 由服务端自动生成并随协议返回客户端只需要做一个通用执行器。我整理了一下两者的差异维度手写工具函数MCP Server工具发现手动维护工具列表协议自动发现参数校验靠函数签名和运行时判断标准 JSON Schema鉴权逻辑每个工具各写一遍Server 统一处理多客户端复用需要单独封装任何 MCP 客户端可直接连新增后端系统加函数、改绑定、改测试加 Server、注册 URL实际项目里最明显的变化是后端新增一个查询接口时我不再需要改 Agent 侧代码。MCP Server 加一个 toolAgent 下一次会话通过tools/list就会自动拿到新工具。1.3 我用的技术栈和版本组合这个项目的核心版本组合是Python 3.11全程异步mcpPython SDK1.x 版本FastAPI 作为 MCP Server 的承载框架LangChain 0.3 LangGraph 0.2 做 Agent 编排模型走 OpenAI 兼容接口本地也可以跑有一个非常重要的版本知识点mcpPython SDK 在 1.x 时代已经把 HTTP SSE 传输方式替换成了 Streamable HTTP。网上大量旧教程还在让你连/sse端点实际上新 SDK 默认走的是/mcp这个端点底层用 POST 做 JSON-RPC 通信长连接由客户端和服务端协商保持。我第一次照着旧教程配置结果服务起来之后客户端一直握手失败后来抓了报文才发现问题是端点路径完全不对。2. MCP 协议握手拆解从 initialize 到 tools/list2.1 三个核心方法initialize、initialized、tools/listMCP 协议通信有几个必须走的标准流程理解了它后面排查问题会轻松十倍。第一阶段是客户端发initialize请求向服务端声明自己要用的协议版本、自身客户端信息和能力声明服务端收到后返回自己的协议版本、能力声明Tools/Resources/Prompts 各支持哪些和服务器信息。第二阶段是客户端发notifications/initialized通知告诉服务端我已完成初始化可以进入正常工作状态。这个通知不需要服务端返回结果但协议要求必须有这个动作否则服务端可能拒绝后续请求。第三阶段才轮到真正的能力发现也就是tools/list。我最初用工具侧的黑盒方式接入对这三个阶段完全没概念一遇到协议版本不匹配就抓瞎。后来我手动构造 JSON-RPC 请求去调 server把每一阶段的响应报文打出来看瞬间就明白了整个握手机制后面再排查任何问题都有明确的方向。2.2 手动发一次握手看清协议长什么样我想用一个可复现的示例来说明。假设你的 MCP Server 已经跑在http://127.0.0.1:8000/mcp你可以直接用 httpx 模拟客户端发起 initializeimport httpx import json payload { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2025-03-26, capabilities: {tools: {}}, clientInfo: {name: debug-client, version: 1.0.0} } } resp httpx.post( http://127.0.0.1:8000/mcp, jsonpayload, headers{Content-Type: application/json, Accept: application/json, text/event-stream}, ) print(json.dumps(resp.json(), indent2, ensure_asciiFalse))正常情况下服务端会返回类似这样的结果{ jsonrpc: 2.0, id: 1, result: { protocolVersion: 2025-03-26, capabilities: { tools: {} }, serverInfo: { name: order-server, version: 0.1.0 } } }看到protocolVersion和服务端能力声明后再补一个notifications/initialized通知然后就可以调用tools/list。用这种方式把协议链路拆开看你会记住每个阶段的服务端行为和响应字段以后用 SDK 封装时再也不用盲猜。提示Streamable HTTP 服务端的响应可能是普通 JSON也可能是 SSE 流。手动调试时记得把Accept头设置成application/json, text/event-stream否则有些严格实现的 Server 会直接回 406。2.3 握手过程中的版本协商和能力声明协议版本是握手阶段最容易出问题的地方。我在不同机器上分别装过mcp0.9.x 和 1.x它们的默认协议版本不一样有的默认2024-11-05有的默认2025-03-26甚至更新版本里还可以协商到2025-06-18。协议设计上允许双方协商SDK 通常会在服务端和客户端都不支持对方版本时选一个两边都支持的最高版本。但如果你的 Server 是手写的或者走了某种网关做转发版本协商就可能失效。我遇到过 Server 端写死成2024-11-05而客户端 SDK 坚持用新版本最终初始化直接报错的情况。所以我的建议是不要依赖默认值在服务端和客户端显式确认协议版本。服务端用 FastMCP 时可以通过参数或配置设置版本客户端连接时也尽量显式指定。两个环境的 SDK 版本尽量保持一致这种问题就能在源头消失。能力声明也是一个关键点。initialize阶段返回的capabilities会告诉客户端这个 Server 支持哪些能力。注意即便 Server 不返回capabilities.tools客户端也可以通过tools/list去试探但规范上还是应该在初始化阶段声明。我在写自己的 Server 时会把tools、resources、prompts的能力都显式声明避免某些严格客户端在启动检测时误判。2.4 握手失败的排查清单结合我调试的经验握手失败通常离不开下面几类原因我放在一张表里方便你对照排查现象最常见原因排查手段连接被拒绝URL 路径错误或服务没起来用 curl 打一下/mcp确认路由存在initialize 返回 406Accept 头缺少text/event-stream检查客户端 headers协议版本报错服务端和客户端版本不匹配打印双方 protocolVersion手动对齐初始化成功后调用 tools 报错忘了发 initialized 通知检查通知是否已发送老教程示例跑不通还在用旧版 SSE endpoint换成 Streamable HTTP 新端点这里有一个我没有在文档里直接找到的细节Streamable HTTP 模式下服务端可以在 initialize 响应里带一个Mcp-Session-Id响应头后续请求需要带上这个会话标识。如果你的 Server 做了会话状态管理而客户端 SDK 版本较老没有正确处理这个头后面的工具调用就会全部失败但初始化看起来又是成功的。遇到这种情况直接看响应头是最快的定位方式。3. 服务端实战FastAPI 上跑一个 MCP Server3.1 为什么我不直接搬官方示例代码mcpSDK 里提供了FastMCP类几行代码就能声明工具并启动服务。官方示例看起来非常优美但直接搬进真实项目会有两个问题第一官方示例通常是一个独立启动的进程端口和生命周期由FastMCP自己管理。可实际项目中 MCP Server 往往是现有 FastAPI 服务的一部分需要和业务接口共享进程、共享数据库会话和配置中心。第二FastMCP封装层级较厚一旦出了协议层面的问题排查起来反而困难。所以我的做法是用 FastAPI 作为主体应用在应用内部挂载 MCP 的 Streamable HTTP 路由。这样既能对外提供标准 MCP 端点又能保留原有的 REST API 做调试和运维。两者共享一套配置和依赖注入体系省掉大量重复代码。3.2 服务端实现工具注册、Schema 生成与结果封装我用FastMCP来声明工具但显式让它跑在 FastAPI 的路由挂载上。一个简化版的服务端是这样的from mcp.server.fastmcp import FastMCP mcp FastMCP(order-server) mcp.tool() async def query_order(order_id: str) - dict: 查询订单状态返回订单基本信息、支付状态和当前物流节点。 # 真实项目里这里会走数据库或远程调用 return { order_id: order_id, status: shipped, payment: paid, logistics: 杭州转运中心 }FastMCP会根据query_order的函数签名和 docstring 自动生成工具的 JSON Schema。这里有个大坑参数的 type hint 不要用泛化的dict或Any否则 Schema 会变成一个无约束的objectLLM 根本不知道要传什么。我习惯的做法是为每个工具定义 Pydantic 模型作为参数结构这样inputSchema会包含详细的字段描述和必填约束。把FastMCP挂载到 FastAPI 的方式也比较直接from fastapi import FastAPI import uvicorn app FastAPI(titleinternal-mcp-gateway) # 根据 SDK 版本选择 streamable_http_app 或 sse_app app.mount(/mcp, mcp.streamable_http_app()) app.get(/healthz) async def healthz(): return {status: ok} if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)工具执行结果我建议只返回 JSON 可序列化的对象MCP 标准里tools/call的返回内容是content数组里面按类型区分text、image等。如果你直接返回一个 Python 对象SDK 会自动序列化但如果你希望结果带结构化的附加信息可以显式构造CallToolResult。我在实际项目中是让所有工具最终返回一个统一格式的 dict包含data、error和trace_id这样 Agent 侧拿到的文本天然就包含上下文。3.3 客户端适配把 MCP Tool 变成 LangGraph 认识的 ToolMCP 服务端跑通之后下一步是写客户端。我用的是mcpSDK 自带的客户端会话走 Streamable HTTP 连接。核心逻辑是初始化会话、拉取工具列表、把每个 MCP 工具转换成 LangChain 风格的BaseTool。一个可运行的转换器大致长这样from mcp.client.session import ClientSession from mcp.client.streamable_http import streamable_http_client from langchain_core.tools import BaseTool from pydantic import BaseModel, create_model def mcp_tool_to_langchain(mcp_tool, session: ClientSession): schema mcp_tool.inputSchema or {} fields {} for name, prop in schema.get(properties, {}).items(): # 这里粗映射真实场景建议按类型准确映射 annotation str if prop.get(type) integer: annotation int elif prop.get(type) boolean: annotation bool fields[name] (annotation, ...) args_model create_model(f{mcp_tool.name}Args, **fields) class MCPLangchainTool(BaseTool): name: str mcp_tool.name description: str mcp_tool.description or args_schema: type[BaseModel] args_model async def _arun(self, **kwargs): result await session.call_tool(mcp_tool.name, argumentskwargs) # 结果 content 可能是多种类型这里取 text texts [c.text for c in result.content if getattr(c, type, ) text] return \n.join(texts) return MCPLangchainTool()这里最值得注意的地方是MCP 工具的参数校验完全交给inputSchema而 LangChain 侧用 Pydantic 模型做同样的校验。如果两边 Schema 不一致LLM 生成的参数可能在运行时被拒。我的经验是不要手工转换 Schema直接让 Pydantic 模型由 MCP 的inputSchema动态创建最大程度保持两边一致。3.4 在 LangGraph Agent 节点里真正跑起来工具转换完成后LangGraph 侧就很简单了。把所有 Server 的工具合并成一个列表用bind_tools绑给模型然后在一个 Agent 节点里做循环推理。from langgraph.graph import StateGraph, MessagesState, END from langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4o, temperature0) tools await collect_all_mcp_tools() # 多个 Server 的 tools 合并 model_with_tools model.bind_tools(tools) async def agent_node(state: MessagesState): response await model_with_tools.ainvoke(state[messages]) return {messages: [response]}单工具场景下LangGraph 官方教程一般会建议你维护一个 tool_calls 循环判断模型返回里有没有tool_calls有就执行工具、把结果塞回消息列表然后继续让模型推理。多 Server 场景下这个循环逻辑没有本质变化但工具来源变多了工具名冲突、执行超时这些问题就会冒出来。这正好引出下一部分也是我这篇文章最想分享的实战内容。4. 多 Server 调用的架构设计与踩坑记录4.1 ServerRegistry连接、会话与工具缓存的统一管理多个 MCP Server 接入时绝对不能每个工具调用都现场新建连接。MCP 的连接建立包含协议握手即使本地握手也要走几轮 HTTP 请求拿个几十毫秒到几百毫秒是很正常的。如果 Agent 一次推理要调三个工具每个工具都现握手整体延迟会非常难看。我设计了一个轻量的ServerRegistry核心职责有三块维护每个 Server 的配置和懒加载连接、缓存每个 Server 拉取到的工具列表、统一提供工具元数据和运行时执行入口。class ServerRegistry: def __init__(self): self._servers {} # name - server config self._sessions {} # name - ClientSession self._tools {} # name - list[Tool] async def get_tools(self, name: str): if name not in self._tools: session await self._connect(name) mcp_tools await session.list_tools() self._tools[name] [mcp_tool_to_langchain(t, session) for t in mcp_tools] return self._tools[name]工具列表缓存的失效策略我用的是 TTL。因为 MCP Server 升级后工具集会变化但 Agent 会话不需要立刻感知所有变化我设置 60 秒过期既保证新鲜度又不频繁触发tools/list。按业务并发量来看这个策略很适合我那边的场景。4.2 工具名冲突前缀命名与兼容性多 Server 环境下最隐蔽的问题是工具名冲突。两个 Server 可能都注册一个叫search的工具合流之后后者会把前者覆盖Agent 实际调用的却是错误实现。这种问题在单测里很难发现因为单测通常不会同时加载两个 Server。我在注册阶段就对所有工具做统一改名规则是从 Server 名派生前缀用双下划线连接。比如jira-server里的search变成jira__searchdb-server里的search变成db__search。为什么用双下划线而不是冒号因为 OpenAI 兼容接口对工具名有严格的字符限制只允许字母数字、下划线和短横线冒号会让部分网关直接报校验错误。这个改名的动作要在工具绑定给模型之前完成而且改名后工具的description里最好保留一句原始工具名xxx方便 Agent 理解这个工具原本属于哪个系统也方便日志追踪。4.3 并发调用下的认证竞态和超时控制多 Server 接入后第二个大坑是认证竞态。我记得有次两个 Server 共用一套 OAuth 2.0 客户端凭据其中一个 Server 内部发现 token 过期后自动刷新刷新期间另一个 Server 拿着旧 token 去请求结果一个刷新动作把另一个 Server 的调用全部打挂。这个问题在单 Server 场景完全不会暴露只有并发调用时才会触发。解决办法有三个层次第一每个 Server 尽量使用独立凭据从源头隔离第二如果必须共享凭据就在 token 管理外层加锁保证同一时间只有一个刷新任务在跑第三请求发出前校验 token 的过期余量余量低于 10 秒就主动等待刷新完成再发起。超时控制同样要针对 Server 分层设置。工具发现tools/list这种低频操作我给了 5 秒超时工具执行tools/call这种可能涉及业务操作的请求我按服务类型给 30 到 60 秒。用asyncio.timeout包住调用避免一个慢 Server 把整个 Agent 的响应时间拖到无限长。4.4 业务可观测性日志、追踪与流式反馈多 Server 场景下一旦用户反馈结果不对排查难度比单工具高得多。我的做法是给每次 Agent 运行生成一个trace_id贯穿 MCP Server 调用全过程。每个 MCP 工具执行时都打结构化日志trace_id、Server 名、工具名、入参、出参、耗时、错误码。有了这些日志定位问题就像翻流水账一样直接。LangGraph 本身支持多种流式输出模式。我在生产环境用的是stream_modemessages这样模型推理过程中的每个 token、每次 tool_call、每条 tool_result 都能实时推给前端。前端可以渲染出正在调用订单服务这类提示用户感知会好很多。这一步虽然不直接解决技术问题但在实际业务中价值非常大——Agent 一卡就是十几秒没有反馈用户早就刷新页面了你后面的优化做得再好都没用。5. 压测结果、优化方向和部署建议5.1 一组实际压测数据项目联调完成之后我在内网环境做了一轮简单压测。三个 MCP Server工具总数 14 个用本地 SQLite 存储模拟业务数据机器是普通的开发笔记本。测试方式是固定 20 个问题每轮并发 10 个 Agent 会话我看三个指标单次 Agent 完整执行时长、工具平均执行时长、错误率。我观察到的典型数据如下场景P50P95错误率单 Server 单工具调用480ms720ms0.2%三 Server 各调一次2.1s3.4s1.1%并发 10 Agent每 Agent 三工具3.8s6.2s2.3%需要说明的是这个数据高度依赖模型推理时间和服务端业务逻辑。真正让我关注的是 P95 和错误率在不同场景下的变化。三 Server 并发时错误率从 0.2% 升到 2.3%主要来源不是模型而是连接复用不够和 token 刷新竞态。这直接驱动了我后面几个优化点的实施。5.2 四个立竿见影的优化点第一缓存工具发现结果。实测中tools/list的耗时占比不低尤其 Server 较多时每个 Server 都做一次列表拉取会显著拉长首轮 Agent 响应。我给工具列表加了一层 TTL 缓存后这部分开销基本降到了零。第二连接复用。MCP 会话尽量长周期保持每个 Server 全局维护一个或者按进程维护一个ClientSession而不是 Agent 运行时每次新建。这个改动对延迟的改善非常直观。第三延迟初始化。Agent 静态绑定了所有工具但实际业务中很多工具不会被用上。我的方案是第一次真正调用某个 Server 的工具时才建立连接后续走缓存。这样既不影响工具发现又能避免为闲置 Server 白白维持连接开销。第四健康检查与降级。每个 Server 每秒有健康检查连续失败超过阈值就把这个 Server 标记为降级状态。Agent 在绑定时过滤掉降级 Server 的工具避免模型选到一个已经挂了的能力导致整个任务失败。这个策略让系统在部分依赖不可用的情况下仍然能完成核心流程。5.3 生产环境部署的几个提醒多 Server 上生产时我建议特别注意几个和开发环境完全不同的点。MCP Server 如果走 HTTP 暴露前面必须加一层认证哪怕是内部网络也不能裸奔。我这边是在网关层统一校验请求头里的服务身份MCP 请求先过网关再转发到具体 Server。另外一个很容易被忽略的是超时和重试的幂等性。MCP 的tools/call有的操作天然幂等查询、读有的不是下单、写数据。如果客户端 SDK 在超时后自动重试会造成业务重复执行。我的做法是客户端不自动重试只在日志里标记需要重试的场景由 Server 端根据业务幂等键做去重。还有部署形态的问题。多个 Server 可以部署在同进程、同机房、跨机房决定因素不是 MCP 协议怎么走而是业务数据在哪里。MCP 协议对跨机房调用并没有特殊支持该走 RPC 内部网关就走网关该走消息队列就走队列协议层不需要为此做额外改动。最后我自己在实际操作中还有一个很深的体会不要过早优化握手延时。MCP 握手在长连接复用的前提下整体开销占比很低真正吃性能的是你 Server 里的业务逻辑和模型推理的并发控制。先把日志打全、把错误码对齐、把超时策略定清楚比一开始就追求微秒级延迟重要得多。这套链路后续要继续扩展的话我大概率会把高频业务工具合并进同一个 Server 进程减少跨进程开销同时保留 MCP 的协议边界让每个 Server 依然可以独立升级和故障隔离。