LangGraph Agent 与 MCP Server 构建聊天机器人实战

简介：面向中文开发者与研究者，项目以LangGraph框架为核心构建Agent工作流，并通过MCP Server完成工具调用，重点解决多步骤推理、记忆管理、循环控制等复杂行为难以调试的问题，适合需要快速搭建高可用智能体应用的中高级Python工程师。项目采用强类型Pydantic状态模型与图结构逻辑，预置十余个中文友好型工具，同步与异步实现均有覆盖，能帮助读者低成本掌握LangGraph状态图编排与MCP协议通信机制。压缩包共13个文件，以9个Python源码为主，涵盖工具适配器、FastMCP服务端、同步/异步聊天入口等核心模块，并附2个文本配置、1个Markdown说明及.gitignore文件，整体仅44KB，结构紧凑无冗余。已有17人学习下载，虽体量小巧但五脏俱全，代码风格规范并配有测试与文档，可直接改造复用或作为学习范本。

1. 为什么是「LangGraph Agent + MCP Server」：这套组合解决聊天机器人的哪些问题

做聊天机器人最常见的一个尴尬是：模型很会聊天，但一问到「帮我查一下订单到哪了」「今天会议室空不空」「这个月组里花费超了没」，它就当场卡壳。早期做法是给模型塞一堆 JSON 格式的「工具说明」，再把工具函数写死在项目里，工具一多，接口对接、鉴权、参数校验全挤在一起，改一个得动一片。LangGraph Agent 把「模型决定调用哪个工具、工具结果再喂回模型」这个循环做成了可控的图结构，而 MCP Server 把工具本身变成标准协议服务，两者一拼，聊天机器人就从「会说话」变成「真会办事」。

这个项目标题对应的，正是一条目前最稳妥的落地路线：用 LangGraph 管状态、管循环、管记忆，用 MCP Server 管工具接入，最后包成一个能跑对话的 AI 聊天机器人。这套方案适合三类人：被工具调用循环逼疯的后端开发、想在公司内部系统上快速接 Agent 的算法工程师、以及刚看完一堆 langgraph 教程但不知道从哪下手的入门者。我下面按自己实践中验证过的顺序，从最小骨架开始讲，到接入协议、编排对话、最后把坑都摊开。

2. 先搭 LangGraph Agent 本体：状态、节点与最小可对话骨架

2.1 为什么用图而不是链：LangGraph 的核心是可控循环

大部分人对 Agent 的误解是「模型自己会循环」，实际 LangGraph 把循环画成了显式的图：模型节点做完推理，条件边检查输出里有没有工具调用，有就跳到工具节点，工具返回后回到模型节点再推理，直到不再调用工具才走到结束。这种设计的好处是每一步都看得见、能打断、能统计，出问题也不会让模型无限自我对话。

和 LangChain 的链式调用相比，图的差异在状态上。LangGraph 的状态是一个共享字典，每个节点往里写增量，不覆盖、可回溯。做聊天机器人时，最常见的状态就是消息列表，官方习惯用add_messages这个归约器做累积。下面是骨架代码，也是后面讲 MCP 接人的底子：

from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage from langchain_core.tools import tool class ChatState(TypedDict): messages: Annotated[list, add_messages] @tool def get_weather(city: str) -> str: """查询指定城市当前天气。""" return f"{city} 今天晴，25 度，适合外出。" tools = [get_weather] llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) llm_with_tools = llm.bind_tools(tools) def agent_node(state: ChatState): result = llm_with_tools.invoke(state["messages"]) return {"messages": [result]} builder = StateGraph(ChatState) builder.add_node("agent", agent_node) builder.add_node("tools", ToolNode(tools)) builder.add_edge(START, "agent") def route_after_agent(state: ChatState): last = state["messages"][-1] return "tools" if last.tool_calls else END builder.add_conditional_edges("agent", route_after_agent, {"tools": "tools", END: END}) builder.add_edge("tools", "agent") app = builder.compile()

这段代码的逻辑不复杂：agent_node把整段消息交给带工具绑定的模型，模型如果认为需要查天气，就会在返回的 AIMessage 里带上tool_calls；路由函数检查最后一条消息，有tool_calls就跳tools节点执行真实函数，执行完把 ToolMessage 追加到消息列表再回到agent；直到模型回答不再带工具调用，图才走到 END。get_weather上的装饰器是关键，docstring 就是模型的「工具说明书」，写得越具体，模型选错工具的概率越低。

2.2 带工具循环的验证对话：判断图是否正常，看返回的消息列表

图编好之后，最直接的验证方式就是跑一轮真实对话。注意我上面刻意没加任何记忆配置，等于每次调用都是独立的，这在调试阶段最干净。

result = app.invoke( {"messages": [HumanMessage(content="北京今天适合跑步吗？")]}, config={"recursion_limit": 10} # 限制循环次数，防止死循环 ) for msg in result["messages"]: role = type(msg).__name__ text = msg.content if isinstance(msg.content, str) else str(msg.content) print(f"[{role}] {text}")

跑出的结果里你会看到四个角色依次出现：HumanMessage、AIMessage（带 tool_calls）、ToolMessage（天气结果）、AIMessage（最终回答）。如果少了 ToolMessage，说明路由没跳过去；如果只有两轮就停了但问题没回答，多半是模型没触发工具调用——这时候应该先查bind_tools是否生效，而不是查图结构。recursion_limit参数是 LangGraph 在每个节点间跳转时都会计数的，默认值偏保守，工具步骤一多就容易误报超限，调参时优先考虑它而不是无脑加循环。

2.3 本地跑通的完整依赖清单

新手在这里最容易翻车的是包装不全。LangGraph 本身只是编排框架，模型调用要靠langchain-openai或langchain-anthropic这类包，工具调用底层需要langchain-core。我在本地环境的建议是装一整组，不要一个个补齐：

pip install langgraph langchain-openai langchain-core langchain-mcp-adapters mcp langgraph-checkpoint-sqlite

langchain-mcp-adapters是后面接 MCP 的桥，现在装了后面就不折腾。装完跑一遍上面的对话脚本，如果提示缺pydantic或anyio版本冲突，直接按报错升级即可，这类冲突在 mcp 依赖链里非常常见，别慌。这一章搭完，你已经有了一个具备工具调用能力的 Agent 骨架，下一步就是把工具从写死的函数换成 MCP Server 提供的远程服务。

3. MCP Server 接入：把外部能力变成 LangGraph 的 Tool

3.1 MCP 解决什么问题：一个协议统一所有工具接入

先厘清概念：MCP Server 本身不是聊天机器人组件，它是 Model Context Protocol（模型上下文协议）的服务端，对外暴露 tools、resources、prompts 三类能力。在 Agent 场景里，我们最常用的是 tools。LangGraph Agent 要调用远程工具，之前得自己写 HTTP 接口、定义鉴权、处理错误码；有了 MCP 之后，任何语言写的工具只要包一层 MCP Server，客户端就能用统一方式拿到「工具有哪些、参数长什么样、怎么调用」。

这里要区分两个角色：MCP Server 负责把真实业务能力（查订单、操作数据库、调内部 API）包装成标准工具；LangGraph 里的 Agent 是消费方。常见做法是让 LangGraph 后端充当 MCP 客户端，通过 stdio 子进程或 SSE over HTTP 连到 Server。选哪种传输取决于部署方式：本地调试用 stdio 最简单，跨机器部署用 SSE。我不会在这章展开复杂加密传输，那些留给第 5 章坑位讲。

3.2 写一个最小可用的 MCP Server：订单查询服务

为了演示完整链路，我写一个最简的订单查询 Server。实际项目里它对接的可能是你的 ERP 数据库，但核心样板完全一致：

from mcp.server.fastmcp import FastMCP mcp = FastMCP("order-service") @mcp.tool() def query_order(order_id: str) -> str: """按订单号查询订单当前状态与物流进度。""" # 真实项目里这里写数据库查询 return f"订单 {order_id} 已发货，预计 3 天后送达。" @mcp.tool() def cancel_order(order_id: str) -> str: """取消未发货的订单，已发货订单会返回失败原因。""" return f"订单 {order_id} 已发货，无法取消。" if __name__ == "__main__": mcp.run(transport="sse")

用 FastMCP 装饰器定义工具是最省事的写法，docstring 同样会被协议序列化成模型可读的描述。注意query_order和cancel_order是两个高度相关的操作，我刻意放在同一个 Server 里——这符合 MCP 的规划思路：按业务域划分 Server，而不是一个工具开一个服务，否则连接数量和鉴权配置都会爆炸。transport="sse"表示通过 HTTP 长连接提供服务，本地可以先改成stdio试跑，命令分别是python order_server.py。

3.3 用 langchain-mcp-adapters 把 MCP 工具拉进 LangGraph

有了 Server，接下来就是让 LangGraph 能发现并且调用这些工具。适配器提供load_mcp_tools，它会连上 Server，遍历协议暴露的工具列表，转成 LangChain 标准 Tool 对象，然后再丢给第 2 章的ToolNode。这里用异步方式连接，注意 LangGraph 节点内部也可以切到异步：

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 fetch_mcp_tools(): server_params = StdioServerParameters( command="python", args=["order_server.py"], env=None, # 需要传递环境变量时可以在这里补 ) read, write = await stdio_client(server_params).__aenter__() async with ClientSession(read, write) as session: await session.initialize() tools = await load_mcp_tools(session) return tools tools = asyncio.run(fetch_mcp_tools())

这里的参数含义要解释清楚：command和args决定客户端以什么方式拉起 Server 进程，stdio_client返回的是一个异步上下文，生命周期由后面的ClientSession管理。session.initialize()是和 Server 做能力协商，必须调用，否则load_mcp_tools拿不到工具列表。关键点在于拿到tools后，你要把它与本地函数工具合并再绑定给模型，load_mcp_tools返回的已经是对方模型可识别的对象，不需要额外转换。取回的工具会保留 docstring 描述，但有时描述写得烂，模型就识别不了，第 5 章会教你怎么强制改写。

all_tools = tools + [get_weather] llm_with_tools = llm.bind_tools(all_tools)

合并后重新绑定模型，再把ToolNode换成接收all_tools即可。这里要特别注意 MCP 工具名可能和本地工具重名，bind_tools一旦遇到重复名不会报错，但调用时会随机选一个，这是后面必踩的雷。

3.4 工具边界判断：哪些能力值得用 MCP 暴露，哪些别硬塞

MCP 不是万能胶。我一般按这三条标准筛选：第一，工具是「请求-响应」型而非持续流式传输，比如查库存、查订单、调审批接口，都适合；第二，工具独立性强，不依赖调用方内存状态，如果你需要在多个调用间保持会话，那不是工具该干的，应该放进 LangGraph 的记忆里；第三，工具延迟可控，如果一个工具要跑几十秒，Agent 来回调用的体验会极其糟糕，宁可改成异步任务提交再加一个查询结果的工具。这些边界决定你的 Agent 是「顺手」还是「卡顿」，提前划清能省掉后面一多半优化时间。

4. 把聊天机器人做到可用：记忆、任务分流与工具失败兜底

4.1 用 checkpointer 给 Agent 装上记忆：按会话隔离上下文

第 2 章的骨架有个致命缺陷：它不记得上一轮聊了什么。聊天机器人没有记忆就是一个残废，所以 LangGraph 提供了一个叫 checkpointer 的机制——它在每次节点执行后把图状态存下来，下次相同会话 id 进来时先恢复再继续。常见实现是 SQLite 保存，好处是可以落盘重启。

from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string("checkpoints.sqlite") as checkpointer: app = builder.compile(checkpointer=checkpointer) result = app.invoke( {"messages": [HumanMessage(content="那北京呢？")]}, config={"configurable": {"thread_id": "user-007"}} )

这段代码最关键的是thread_id。它相当于会话主键，同一个user-007的多轮对话会共享消息历史，换个 id 就是全新对话。我在实际项目里通常用用户 ID 拼上会话 ID 做 thread_id，而不是直接用用户 ID，这样既能多轮记忆，又能支持同一个用户开启多个独立话题。SqliteSaver.from_conn_string打开的连接会在退出 with 块时自动提交，注意别把compile放进 with 外——常见错误是有人拿到 checkout 对象后又立刻关了连接，导致再次调用时直接报「database locked」。

4.2 任务分流：闲聊和工具调用不该挤在一个 Agent 里

把天气查询、订单查询、闲聊混在一个 Agent 里是新手最容易犯的架构错误。模型会在「该聊闲天」的时候也去调用一次没必要的工具，既慢又费钱。我的做法是把对话路由提到最前面：先用一个轻量分类步骤判断用户意图，再决定走纯聊天子图还是工具子图。LangGraph 里可以用条件边在入口处做分流，逻辑非常直观：

def route_by_intent(state: ChatState): text = state["messages"][-1].content.lower() if "订单" in text or "取消" in text or "物流" in text: return "mcp_agent" return "free_chat" builder.add_conditional_edges(START, route_by_intent, { "mcp_agent": "mcp_agent", "free_chat": "free_chat", })

这个方案的关键在于两个子图（mcp_agent和free_chat）都是完整可独立编译的图，再作为节点挂到外层总图上。好处有二：纯闲聊路径不会因为误触发工具带来额外延迟；工具路径可以单独配更高的recursion_limit和更长的超时。关键词匹配是粗路由，够用且稳定；如果业务复杂，可以换成一个小分类模型。做多 Agent 编排时这种「总图套子图」的模式，比把所有逻辑平铺在一个图里好维护太多。

4.3 工具失败与重试：不让一次报错杀死整段对话

真实世界里 MCP Server 会挂、数据库会超时、第三方接口会限流。这些问题在聊天机器人里的表现就是：模型调用工具失败后，要么卡死在工具节点，要么直接给用户一段「我无法完成你的请求」的废话。LangGraph 的节点支持retry策略配置，对瞬时错误很有用：

from langgraph.pregel import RetryPolicy builder.add_node( "tools", ToolNode(all_tools), retry=RetryPolicy( max_attempts=2, retry_on=lambda exc: isinstance(exc, TimeoutError), ), )

retry_on参数后面接一个判断异常的 lambda，只有符合条件才重试。我的习惯是只对TimeoutError和网络抖动重试，业务性错误如「订单不存在」绝不重试，因为再试多少次结果都一样，白白放大延迟。重试耗尽后，ToolNode 会往上抛异常并中断整个图，所以外层调用还要包一层 try-except，捕获后返回固定的兜底文案。这是agent 开发里最容易忽略的一条：模型层、工具层、传输层任何一层出事，都得让用户感知到「系统正在处理」而不是体验一场静默失败。

5. 集成中的常见坑：从工具黑匣子到死循环

5.1 坑：MCP 工具返回结构解析失败

现象：接入 MCP Server 后，LangGraph 执行到工具节点报错，错误信息类似Expected ToolMessage, got list。

原因：MCP 协议里工具的返回是一个 content 数组，而load_mcp_tools在部分版本下返回的是原始结构，没有自动转成 LangChain 的字符串格式。模型在 ToolMessage 里读到的是一坨 JSON，而不是干净文本。

解决：拿到tools后不要直接绑定，先包一层转换函数，把 content 数组里的 text 拼出来。代码里这样写：

def normalize_output(tool): original = tool.func def wrapped(*args, **kwargs): raw = original(*args, **kwargs) if isinstance(raw, list): return "\n".join( item.get("text", str(item)) for item in raw if isinstance(item, dict) ) return raw tool.func = wrapped return tool tools = [normalize_output(t) for t in await load_mcp_tools(session)]

这个操作相当于在协议层和模型层之间加适配器，不要小看它，很多看起来「模型不听话」的诡异问题，根源就是工具返回格式让模型读不懂。

5.2 坑：多个 MCP Server 工具同名互相冲掉

现象：同时接了订单服务和天气服务，两个都暴露了query，模型调用时拿到的结果经常错乱，时而是天气时而是订单。

原因：bind_tools内部用名字做映射，重名时后注册的会覆盖先注册的，但覆盖顺序不一定是你预想的。

解决：加载后统一重命名。MCP 工具名里其实自带 Server 前缀，可以保留前缀策略，我建议直接改成「server名_工具名」的格式：

for t in tools: t.name = f"{t.metadata.get('server_name', 'mcp')}_{t.name}"

不管底层命名怎么变，模型侧永远看到唯一名字，这样才能在日志里追溯到底调的哪个源。血的教训：千万别为了「让模型更好理解」去简化工具名，工具名唯一性远大于可读性。

5.3 坑：加了记忆后thread_id报错或状态丢失

现象：配置好SqliteSaver后，多轮对话记忆时好时坏，甚至偶尔报ContextVar相关的序列化错误。

原因：状态字典里混进了模型对象或者文件句柄，SQLite 保存前序列化失败。另一个常见原因是compile用的是旧的图对象，新图没带 checkpointer 就跑了。

解决：保证ChatState里所有字段都能被json.dumps，模型对象永远不要塞进 state，只存消息 ID 或纯文本。编译后第一时间检查图对象是不是同一个实例。

app = builder.compile(checkpointer=checkpointer) print(app.checkpointer) # 必须输出对象地址，而不是 None

5.4 坑：Agent 陷入工具调用死循环

现象：对话既不报错也不结束，日志里模型反复调用同一个工具，像中了邪。

原因：工具结果没有给模型提供决定性信息，模型试图通过反复调用「撞」出答案；或是路由条件写反，工具节点永远跳回 agent 且 agent 永远再次触发调用。

解决：第一道防线是recursion_limit，在 invoke 时设置合理的上限，超过就强制终止并返回一个「需要人工介入」的兜底文案。第二道防线是加一个循环检测节点，统计相同工具调用出现的次数，超过 3 次就直接切换成纯文本回答。这个检测节点 10 行代码，但能救你一命。

5.5 坑：MCP 连接泄漏导致端口被占满

现象：跑完几轮对话后，系统出现大量孤儿进程或端口冲突，Connection refused频繁出现。

原因：stdio_client启动的子进程在调用结束后没有被正确关闭。尤其在 FastAPI 这类长期运行的服务里，每次请求新建连接却忘记释放。

解决：把连接生命周期交给应用的依赖注入去管理，而不是在每次对话时临时创建。

from contextlib import asynccontextmanager @asynccontextmanager async def mcp_lifespan(): server_params = StdioServerParameters(command="python", args=["order_server.py"]) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() yield session

所有对话请求共享这一个 session。对话高频时连接能复用，反之也不泄漏。这个做法我通常写在第 6 章的压测之前，因为不解决连接管理，压测一跑准炸。

6. 上线前验证与调优：并发压测、回归用例和安全边界

6.1 用一句话回归集验证工具调用

聊天机器人的验收不能靠「感觉回答得不错」。我会准备一组固定问题，覆盖每个工具的正常调用、拒绝场景、闲聊场景，并在每次改动后跑一遍自动比对。这个回归集其实就是一组(输入, 期望工具名)的断言表：

regression_cases = [ ("帮我查订单 A10023 的状态", "query_order"), ("取消订单 B99999", "cancel_order"), ("北京天气怎么样", "get_weather"), ("今天心情不太好", None), # 预期不调用工具 ] for text, expected_tool in regression_cases: result = app.invoke({"messages": [HumanMessage(content=text)]}, config={"configurable": {"thread_id": "regression"}}) actual_tool = None for msg in result["messages"]: if getattr(msg, "tool_calls", None): actual_tool = msg.tool_calls[0]["name"] assert actual_tool == expected_tool, f"{text} 期望 {expected_tool}，得到 {actual_tool}"

这套断言不校验回答文本，只校验是否调用了正确的工具，因为文本生成有随机性，工具调用则必须是确定行为。回归集是控住 Agent 不翻车的最有效工具，每改一次 prompt、每调一次工具描述，先跑它再聊别的。

6.2 并发与超时配置：压测前先想清楚三个参数

并发压测时最先暴露的往往不是模型性能，而是连接和队列。聊天机器人服务本质上受模型 API 速率限制，所以压测要重点看三个参数：单 Agent 实例的recursion_limit（限制单轮最大步数）、MCP 连接的复用策略（见 5.5）、以及外部请求超时时间。我通常把模型调用的超时设为 30 秒，工具调用 15 秒，超过就按失败处理。压测命令简单直接：

hey -n 200 -c 20 -m POST -d '{"message":"查一下订单 A10023"}' http://127.0.0.1:8000/chat

如果压测中 95 分位延迟超过 5 秒，优先看 MCP Server 的数据库查询耗时，而不是盲目换更强的模型。大多数聊天机器人的延迟瓶颈在工具链路，不在推理。

6.3 安全边界与权限收敛

接入 MCP 工具后，模型就有了真正执行操作的权限，比如查订单、取消订单、改配置。agent 安全在这里不是一句口号，而是要落实成两件事：第一，工具参数必须走白名单校验，不能让模型自由拼接任意输入；第二，写操作类工具要在 MCP Server 内部加用户级权限判断，不能用模型说什么就执行什么。我的习惯是在工具函数入口加一层require_role("user")的装饰器，然后在路由分流时就把「只读」和「写操作」分进不同子图，写操作额外要求用户二次确认。这样做即使模型被诱导输出恶意工具调用，到执行层也会被拦下来。

最后分享一个我踩了三次才改掉的教训：不要把模型 API Key 写在 MCP Server 的环境变量里，更不要把它打进传给前端的配置。有一次我在调试时把 Server 的 stdio 参数打印到了日志，结果整个 API Key 直接出现在日志文件里，当天晚上就被人刷了几百美元。后来我把所有 MCP Server 的敏感配置全部收进独立的.env文件并加入 gitignore，服务启动时统一加载，日志里对敏感字段打码。这个习惯后来救了我很多次。希望帮到你。

本文还有配套的精品资源，点击获取