做智能体应用的朋友最近问我:为什么模型调外部工具总是越调越乱。我的答复很简单——工具接入没有标准化,流程编排没有框架化。后来我把手头的一个 Demo 全部切到“MCP 协议对接外部工具 + LangGraph 做编排”这套组合上,整条链路一下就清晰了许多。这篇实战记录就围绕这套组合展开,从选型逻辑、核心机制,到一份能直接跑通的 Demo 代码,再到我实际踩过的坑,完整过一遍。适合正在做智能体工程化、想把工具调用做标准的开发者参考。
1. 方案选型:为什么是 MCP 协议 + LangGraph
1.1 从硬编码函数到协议化对接
我第一次做工具调用的时候,代码结构大概是这样的:一个tool_registry字典,里面放着get_weather、calculate这类函数,然后让模型从一堆工具描述里挑一个“函数名”,代码再按名字去调。工具只有两三个时很顺手,工具一多就开始乱——参数校验散落各处、错误处理写法不统一,最关键的是每个工具都跟当前模型应用深度耦合,换一个框架就要重写一遍对接逻辑。
MCP(Model Context Protocol,模型上下文协议)解决的是工具对接的标准化问题。MCP Server 统一暴露工具列表和调用端点,客户端通过协议动态拿到工具清单,再按统一的 JSON-RPC 2.0 格式发起调用。模型应用层不再关心工具是什么语言写的、部署在哪个进程,只需要理解同一套协议。这就好比以前每个家电都要配一个专用遥控器,现在统一成了一个标准遥控器,家电自己上报“我能做什么、怎么操作”。
我后来把工具服务全部包成 MCP Server,Agent 侧不再硬编码任何工具函数。新增能力时,我只需要往 MCP Server 里加一个工具定义,所有接入同一 Server 的智能体都能自动发现并调用它,这才是“复用”。
1.2 MCP 的工作方式与生命周期
MCP 的通信模型是客户端-服务端模型。客户端启动后,先通过initialize握手确定协议版本,然后发送tools/list拉取工具清单,真正执行时发送tools/call并携带工具名和参数。工具清单里每个工具由三部分组成:name、description、inputSchema(JSON Schema 格式的参数定义)。这三点直接决定了大模型能不能准确选中工具、会不会生成合法参数,所以工具描述和参数 schema 其实是整个协议链路里对效果影响最大的部分。
传输层一般有两种:stdio 和 HTTP+SSE。stdio 适合本地进程,Agent 启动一个 MCP Server 子进程,通过标准输入输出通信;HTTP+SSE 适合把工具服务部署到远端,多个客户端共享。我在 Demo 里用的是 stdio,调试简单,不用处理网络鉴权,等工具变多了再平滑迁移到网络传输。两种传输在tools/list、tools/call这些方法语义上完全一致,换传输层不会侵入业务代码。
1.3 LangGraph 在链路中的位置
MCP 解决“外部工具怎么被调用”,LangGraph 解决“调用流程怎么编排”。实际 Agent 场景不是简单的一问一答,而是多轮推理、条件分支、状态累积。我在早期用过朴素的 while 循环:模型响应里有tool_calls就执行工具,结果追加给模型,再继续循环。单线程跑通没问题,一旦要并发、要持久化、要支持断点续跑,循环版本完全撑不住。
LangGraph 的核心思路是把流程画成一张图。节点是计算单元,agent 节点负责和模型交互,tools 节点负责执行工具;边是流转方向;条件边根据当前状态决定下一步去哪。节点之间共享一个状态对象,比如消息列表。整个流程结构清晰,还能用 checkpoint 机制把状态落盘,中断之后可以恢复执行。这也是我最终放弃手写循环、选择 LangGraph 的根本原因。
2. 核心概念拆解:真正理解这套链路
2.1 LangGraph 的 State、Node 与 Edge
LangGraph 的顶层抽象是StateGraph。你需要先定义一个State类型,它可以是 TypedDict,描述节点之间传递的数据。最常见的字段是messages,通常声明为Annotated[list, add_messages],意思是每次节点返回的消息列表都以追加方式合并,而不是整体覆盖。这保证了多轮对话中历史消息不会丢。
节点是接收state、返回更新字典的异步函数。比如 agent 节点会调用模型,把返回的 AI 消息写进状态;tools 节点执行工具后把ToolMessage写进状态。边分普通边和条件边:普通边是固定的“A 执行完去 B”,条件边则是一个路由函数,它根据当前状态里的最后一条消息是否包含tool_calls来决定下一步是继续调工具,还是直接结束。
ToolNode是 LangGraph 预置好的工具执行节点,它专门读取 AI 消息里的tool_calls,逐一执行对应工具,并把结果包装成ToolMessage。也就是说,只要把 MCP 工具列表喂给ToolNode,模型发起的工具调用就会被自动转发执行,不需要自己写调用分发逻辑。
2.2 MCP 客户端与服务端的通信模型
在 Python 生态里,MCP 客户端通常用官方 SDK 配合stdio_client使用。stdio_client会启动一个 MCP Server 子进程,通过 stdin/stdout 与它交换 JSON-RPC 2.0 消息。ClientSession负责会话管理,首先要调initialize()完成握手,之后才能调用list_tools()或call_tool()。
这里有个非常重要的细节:因为通信走的是 stdout,MCP Server 里任何print()输出都会混入协议帧,轻则导致警告,重则直接握手失败。服务端所有日志都要写到 stderr 或文件,这是踩坑率最高的地方之一,后面我会专门展开。实际对接 LangGraph 时,可以用langchain-mcp-adapters里的load_mcp_tools(session)方法,它会把 MCP 工具列表自动转换成 LangChain 的 StructuredTool,直接喂给ToolNode或模型的bind_tools,省去大量的手写封装。
2.3 工具描述、入参 Schema 与模型推理的关系
模型是否每次都能正确调用工具,很大程度上不看代码逻辑,而看工具描述写得好不好。工具描述要像“给模型看的说明书”,要写明:工具是做什么的、参数含义是什么、返回内容长什么样、什么场景适合用它。描述含糊的代价是模型要么不调用,要么生成完全不合法的参数。
入参 Schema 上,应该尽量多给约束:参数默认值、枚举值、格式示例。我的经验是,凡是参数没有枚举约束的自由文本,模型翻车率都明显更高。比如get_weather(city: str),如果不在描述里写明“城市名必须是标准城市名”,模型可能会传一个口语化的表达式,导致工具拿到后解析失败。返回结果也一样,如果工具返回复杂嵌套 JSON,描述里说明结构,模型才能准确提取字段给用户。这也是我后来把工具描述当成产品文案来写的原因。
3. 实战实现:完整跑通一个工具对接 Demo
3.1 环境准备与依赖安装
先准备 Python 3.10 以上的环境,然后安装依赖:
pip install langgraph langchain-mcp-adapters mcp langchain-openailanggraph提供编排框架,mcp提供协议客户端和服务端能力,langchain-mcp-adapters负责把 MCP 工具转成 LangChain 工具,langchain-openai则是模型调用客户端。模型部分我建议选一个支持 OpenAI 兼容接口的模型服务,通过base_url和api_key传入,密钥用环境变量管理,不要写死在代码里。
安装完可以检查一下版本兼容性,这几个包迭代速度都很快,版本错位会导致某些函数签名对不上。用pip list | grep -E "langgraph|mcp|langchain"一拉就能看清楚。
3.2 编写 MCP 服务端
我用 FastMCP 来定义本地工具服务,它基于装饰器语法,非常直观。下面是两个模拟外部工具:天气查询和简单四则运算计算器。
# mcp_server.py from mcp.server.fastmcp import FastMCP import operator mcp = FastMCP("demo-tools") @mcp.tool() def get_weather(city: str) -> str: """查询指定城市的当前天气。city 参数必须是标准城市名,例如:栖云市、澜江市。 返回格式:城市、温度、天气状况。""" mock_data = { "栖云市": {"temperature": 23, "condition": "晴"}, "澜江市": {"temperature": 18, "condition": "小雨"}, } data = mock_data.get(city, {"temperature": 20, "condition": "未知"}) return f"{city}:{data['temperature']}℃、{data['condition']}" @mcp.tool() def calculate(expression: str) -> str: """计算形如"3 + 5"的简单四则运算表达式,操作数与运算符之间用空格分隔。""" parts = expression.split() if len(parts) != 3: return "表达式格式错误,请使用空格分隔,例如:3 + 5" a, op, b = parts ops = { "+": operator.add, "-": operator.sub, "*": operator.mul, "/": operator.truediv, } try: return str(ops[op](float(a), float(b))) except Exception as e: return f"计算失败:{e}" if __name__ == "__main__": mcp.run()注意几个点。第一,工具函数的 docstring 会直接影响生成工具的description,所以一定要写得像前面说的“给模型看的说明书”。第二,calculate里我刻意只支持空格分隔的简单表达式,避免引入eval带来的任意代码执行风险,真正的外部工具如果涉及系统命令,一定要做严格的参数校验和权限控制。第三,mcp.run()不带参数时默认走 stdio 传输,正好匹配 Agent 侧的stdio_client。
真实项目里,把函数体换成对应的 HTTP 调用、数据库查询即可,MCP 工具的定义方式不会变。
3.3 编写 LangGraph Agent 并加载 MCP 工具
Agent 这一侧的核心逻辑是:启动 MCP 客户端会话、加载工具、定义状态、搭图、跑通一次对话。完整代码如下:
# agent.py import asyncio from typing import Annotated, TypedDict, Literal from langchain_openai import ChatOpenAI from langchain_mcp_adapters.tools import load_mcp_tools from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client class AgentState(TypedDict): messages: Annotated[list, add_messages] async def run(): server_params = StdioServerParameters( command="python", args=["mcp_server.py"], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools = await load_mcp_tools(session) model = ChatOpenAI( model="YOUR_MODEL", base_url="YOUR_BASE_URL", api_key="YOUR_API_KEY", temperature=0, ).bind_tools(tools) async def agent(state: AgentState): response = await model.ainvoke(state["messages"]) return {"messages": [response]} builder = StateGraph(AgentState) builder.add_node("agent", agent) builder.add_node("tools", ToolNode(tools)) builder.add_edge(START, "agent") def route(state: AgentState) -> Literal["tools", "__end__"]: last = state["messages"][-1] return "tools" if getattr(last, "tool_calls", None) else "__end__" builder.add_conditional_edges( "agent", route, {"tools": "tools", "__end__": END}, ) builder.add_edge("tools", "agent") app = builder.compile() result = await app.ainvoke({ "messages": [{"role": "user", "content": "栖云市天气怎么样?顺便算一下 15 * 27 等于多少。"}] }) for msg in result["messages"]: if getattr(msg, "content", None) or getattr(msg, "tool_calls", None): print(f"[{msg.type}] {msg.content}") if __name__ == "__main__": asyncio.run(run())这段代码里最核心的链路是:load_mcp_tools(session)一次性把 MCP 工具列表拉出来,传给model.bind_tools(tools),让模型在推理时可以看到这些工具并生成tool_calls;ToolNode(tools)则负责实际执行。模型一旦在回复里给出工具调用,条件边就会把流程导向 tools 节点;工具结果返回后,流程再回到 agent 节点,让模型看到结果并生成最终回答。
如果你不想手动维护条件边,LangGraph 还提供了create_react_agent(model, tools),一行就能构建出同样的 agent+tools 循环。手动写的好处是可读性和可扩展性更强,比如将来想在工具执行和最终回答之间插入“文本总结节点”或“人工确认节点”,手动图改起来更方便。
3.4 运行效果与调用链路分析
确认两个文件在同一个目录下,运行:
python agent.py输出大致如下(具体内容取决于你选的模型):
[human] 栖云市天气怎么样?顺便算一下 15 * 27 等于多少。 [ai] tool_calls: [get_weather(栖云市), calculate(15 * 27)] [tool] 栖云市:23℃、晴 [tool] 405 [ai] 栖云市目前是晴天,23℃。另外,15 乘以 27 等于 405。你可能会发现,模型在一次回复里同时发起了两个工具调用,ToolNode会并行或顺序执行它们,然后把两条ToolMessage都追加到状态里,再回到 agent 节点。最后模型综合所有工具结果输出自然语言回答。整个循环之所以能维持多轮上下文,靠的就是状态里不断累积的messages列表。这也是 LangGraph 相比手写循环最有优势的地方——状态是显式的,流程是可见的。
4. 常见问题与排查技巧实录
4.1 服务端进程启动异常与 stdout 污染
MCP over stdio 遇到的第一大坑就是服务端日志污染 stdout。一旦服务端代码里有print("debug..."),这些内容就会混进协议帧,客户端解析时报错“Received unexpected message”或直接握手超时。我刚开始调试时花了一晚上才定位到,原来是一个被保留的调试输出。
排查方法是先单独测试 MCP Server。写一个最简单的 Python 脚本,建立会话后initialize并list_tools,确认协议本身是通的,再加到 LangGraph 里。服务端自己的日志统一走logging到 stderr,或者干脆写文件。另外,StdioServerParameters里的command和args要注意正确性,Python 环境不一致时还要考虑用sys.executable而不是裸的python命令,否则子进程起不来。
4.2 工具调用超时与异步阻塞
外部工具如果本身是慢接口,比如请求第三方服务,那就要格外小心超时。MCP Server 端如果写了同步的阻塞调用,整个 Agent 的事件循环会被卡住,表现就是请求很久没有响应。我的处理方式是:服务端内部所有 IO 操作都用带超时的客户端,比如httpx.AsyncClient(timeout=30);同时在客户端侧也给会话调用设置超时,超时异常会被捕获并转成一个可读的 ToolMessage,让模型知道工具没执行成功,可以给出“稍后重试”之类的回复。
实际开发中,我把“单次工具调用最长 30 秒”作为约定,超过就报错并留下重试信息,而不是无限等下去。这个默认值建议写进工具描述里,模型对超时行为也有预期。
4.3 工具返回结构与模型解析不兼容
MCP 工具的返回内容是一组content列表,元素可能是TextContent、ImageContent等。load_mcp_tools转成 LangChain 工具后,默认会把文本内容拼成字符串,模型拿到的是这段字符串。如果你的工具返回了一个无格式的 JSON 字符串,模型不至于看不懂,但二次解析时很容易出错,比如字段名大小写、嵌套层级。
我的经验是:让服务端直接返回“已经格式化好的文本”,不要考验模型的 JSON 解析能力。比如天气工具返回栖云市:23℃、晴,模型直接摘抄就能回答用户。如果确实需要返回结构化数据,务必在工具描述里写明完整示例,让模型照着格式理解。
4.4 调试建议与安全边界
调试时我建议分三层排查:第一层,直接运行 MCP Server 并用脚本调用工具,确认协议和函数本身没毛病;第二层,单独跑 LangGraph 图,打印每轮状态里messages的长度和类型,确认流程走到了哪个节点;第三层,再加真实模型和工具,看模型生成的tool_calls是否符合预期。这种分层定位法能快速区分是协议问题、流程问题还是模型理解问题。
安全方面,MCP 工具描述会被直接喂给模型,存在提示词注入的风险。恶意工具描述可能诱导模型忽略用户意图去执行危险操作。我的原则是:工具权限最小化,凡是能暴露给模型的能力都要做参数校验和输出过滤,尤其涉及文件、网络、命令执行的工具,必须加白名单和确认机制。工具调用日志也要记录完整参数,方便事后审计。
5. 对后续扩展最有用的一些习惯
5.1 先包 MCP 再谈复用
现在不管新增什么外部能力,第一件事不是写代码直接调它,而是先包一层 MCP Server,把工具能力描述清楚——它是什么、参数是什么、返回是什么、什么时候该用。一开始多花半小时写描述和 schema,后面无论模型怎么换、Agent 框架怎么换,工具侧都不需要动。我手头的 Demo 已经用这种方式接入了三四种外部服务,每次扩展智能体能力都只是加一个装饰器函数,体验非常顺。
5.2 这套组合还能往哪里扩展
跑通最小链路之后,值得继续加东西的方向按优先级排大概是:把 stdio 传输换成 HTTP+SSE,让工具服务可以远程复用;给 LangGraph 加 checkpoint 持久化,让长流程支持中断和恢复;把不同权限的工具拆成多个 MCP Server,按 Agent 职责挂载,做多智能体协作。每一步都是在现有框架上叠加,不需要推翻重来。
这整套方案最大的价值,不是某一个库有多神奇,而是把“模型想做事”和“系统能做事”之间的连接变成了标准协议,再用可靠的编排引擎把它串起来。我个人踩过几次“工具描述含糊导致模型瞎调参”的坑之后,现在把工具描述当成产品文案在写。这套习惯,比具体用哪个框架重要得多。