★ FEATURED ARTICLE 简介面向中文开发者与研究者的LangGraph Agent与MCP Server集成项目旨在解决复杂智能体工作流的构建、调试与扩展难题。项目以LangGraph图结构编排多步推理与工具调用配合兼容MCP v1.0标准的服务端实现模型控制协议下的高效通信。压缩包共13个文件以Python脚本为主9个py涵盖同步与异步实现、MCP服务端、工具适配器等模块另有txt配置、Markdown说明及gitignore文件便于快速了解环境依赖与运行方式。整个包仅44KB结构紧凑适合学习图状态管理、状态机设计及MCP插件化扩展。目前已有17人学习下载适合具备一定LangChain基础、希望深入Agent编排与协议集成的开发者。通过该项目可掌握LangGraph节点/边定义、MCP工具注册与调用流程并参考其分层架构与类型注解规范快速搭建可扩展的智能聊天机器人。1. LangGraph Agent别再用手写循环做 Agent 了接手这个项目之前我在生产环境里用纯 LangChain 写过两次 Agent。一次是客服机器人一次是内部运维助手两次都栽在同一个地方循环逻辑全靠自己维护工具一多状态就乱最后只能靠一个巨大的 while 循环加十几个 if else 硬撑项目越写越像一锅粥。后来切到 LangGraph Agent MCP Server 这套组合才真正觉得 Agent 的开发方式该换了。LangGraph 把 Agent 抽象成一张状态图节点是逻辑、边是路由、状态机负责记忆流转MCP Server 则负责把外部工具和数据源以统一协议的形式暴露给模型。这份资源不是给你看概念展示的而是把「LangGraph MCP Server 集成成可对话的智能聊天机器人」完整跑通的工程包。适合已经会写 Python、玩过 LangChain 但你发现自己写的 Agent 一上多工具就失控的开发者也适合正准备用 MCP 协议接外部工具但不知道怎么和 Agent 框架融合的人。下面我会把整个项目的结构、关键代码、参数设计、踩坑记录和验证方法照着拆一遍。2. LangGraph 核心机制为什么 Agent 需要一张图而不是一段循环2.1 状态机是 Agent 的地基LangGraph 和普通 LangChain Agent 的本质区别在于它把执行过程抽象为「节点 边 状态」。节点是具体的处理单元比如「大模型决策节点」「工具执行节点」边决定了节点之间的流转方向包括条件边——根据上一节点的输出去动态决定下一跳状态则是一个贯穿全局的数据结构保存对话历史、中间结果和工具返回值。这个设计意味着你不再需要自己维护循环标志位、临时变量和退出条件图本身就是执行的骨架。在这个项目里状态定义用的是 TypedDict这也是 LangGraph 官方推荐的写法。先看状态定义部分from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[List, add_messages] current_tool: str mcp_results: dict tool_call_history: List[dict]messages字段用add_messages做归约器这是 LangGraph 状态管理里最值得理解的一个点——普通字典赋值是覆盖而归约器规定了新增消息是追加而非覆盖。current_tool记录当前要调用的工具名mcp_results存放 MCP Server 返回的结构化数据tool_call_history保存每次工具调用的参数和结果方便后期的调试与审计。我一般会强调一点不要把全部数据塞进状态对象里。只放「跨节点需要共享」的数据比如消息列表和工具结果。每个节点内部用完即弃的临时数据就存在局部变量里否则状态图跑久了之后内存占用会非常难看。2.2 节点和边的角色拆分有了状态定义接下来是图的核心——节点。这个项目里最基础的节点有三个agent_node负责大模型决策mcp_tool_node负责执行 MCP 工具调用还有一个human_node用于接收用户输入。节点定义不复杂关键是每个节点只做一件事职责清晰了图才能好调试。from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END llm ChatOpenAI(modelgpt-4o, temperature0.2) def agent_node(state: AgentState): response llm.invoke(state[messages]) if response.tool_calls: return { current_tool: response.tool_calls[0][name], messages: [response] } return {messages: [response]} def mcp_tool_node(state: AgentState): tool_call state[messages][-1].tool_calls[0] result mcp_client.call_tool(tool_call[name], tool_call[args]) return { mcp_results: result, messages: [ToolMessage(contentstr(result), tool_call_idtool_call[id])] }这段代码的逻辑是模型收到消息后如果决定调用工具tool_calls字段就会有内容流程就转向mcp_tool_node如果没有工具调用需求直接返回文本给用户流程结束。mcp_client.call_tool是 MCP SDK 提供的统一调用入口不需要关心工具是 HTTP 接口还是本地脚本。需要特别注意ToolMessage的tool_call_id必须和原始调用 ID 对上否则模型无法把工具结果和调用请求关联起来这也是新手最容易犯的错之一。图的组装部分用的是 StateGraph 的链式写法graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(mcp_tool, mcp_tool_node) graph.add_edge(agent, mcp_tool) graph.add_conditional_edges(agent, should_continue) graph.add_edge(mcp_tool, agent) graph.set_entry_point(agent)add_conditional_edges是 LangGraph 的精华——它打破了固定的边连接让模型根据当前状态决定走哪条路。should_continue是一个路由函数检查最后一条消息是否还有 tool_calls有就回mcp_tool没有就回END。这套机制本质上是把原先手写的 while 循环替换成了图自身的条件跳转。项目里还配置了checkpointer用的是langgraph-checkpoint-sqlite作用是持久化每个会话的状态挂载之后 Agent 就有了长期记忆能力——即使是跨多轮对话状态也能在服务重启后恢复。3. MCP Server 接入从工具注册到协议选型3.1 MCP 解决了什么问题MCPModel Context Protocol解决的核心问题是工具接入方式的碎片化。在传统 LangChain 里每接一个工具就要为它写一个tool装饰的函数然后手动绑定到模型。不同工具的错误处理方式、参数格式、返回结构都不统一。MCP Server 给了你一个统一协议层——每个工具变成一个 endpoint客户端通过标准化的 JSON-RPC 调用来执行模型不再需要知道底层工具的具体实现。这个项目里 MCP Server 的选型用了 FastMCP 库Python 生态里接 MCP 最顺手的框架。一个典型的 MCP Server 入口长这样from fastmcp import FastMCP mcp FastMCP(agent-mcp-server) mcp.tool() def fetch_weather(city: str) - dict: 获取指定城市的天气信息city为城市拼音或中文名 return {city: city, temperature: 28, status: sunny} mcp.tool() def search_docs(keyword: str, limit: int 5) - list: 在本地知识库中检索相关文档片段 results vector_store.similarity_search(keyword, klimit) return [{title: r.metadata[title], content: r.page_content} for r in results]工具函数的 docstring 不是摆设——MCP 会把函数签名连同 docstring 一起发送给模型侧的 tools schema函数注释写得越清晰模型调用工具的参数就越准确。我见过很多人工具写好了但 docstring 用默认模板结果模型经常把参数传歪。参数类型标注也直接决定了 JSON Schema 的生成city: str和city: Enum生成的 schema 完全不同前者模型自由发挥后者模型被约束在固定选项里。3.2 两种传输模式什么时候选哪个MCP 支持stdio和sse两种传输模式这是接入时需要做的一个关键决策。stdio模式下MCP Server 作为子进程启动客户端通过标准输入输出通信适合本机单用户的场景特点是快且稳不占端口。sse模式则是 Server 独立起一个 HTTP 服务支持跨机器调用适合部署在服务器上给多个客户端共享的场景。项目里默认跑的是sse模式启动命令是python mcp_server.py --transport sse --host 0.0.0.0 --port 8910对应的客户端连接方式from mcp import ClientSession from mcp.client.sse import sse_client async def create_mcp_session(): url http://127.0.0.1:8910/sse async with sse_client(url) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() return session注意session.initialize()这步必须执行不初始化就调用list_tools()会直接抛InvalidStateError。我一开始就栽在这上面过——写完连接代码忘了异步初始化结果报了 30 多秒的超时错误。排错时最容易忽略的就是协议握手顺序MCP 客户端和服务端建立连接后必须先完成 initialize 握手然后才能进行工具列表查询或调用。3.3 工具列表的懒加载与预热MCP Server 注册的工具数量多了之后每次新建会话都做一次list_tools()会有可感知的延迟。这个项目里做了一个小的缓存优化——把工具列表缓存半小时import time from functools import lru_cache lru_cache(maxsize1) def get_tools_with_cache(): return asyncio.run(list_all_tools_sync()) TOOL_CACHE_TTL 1800 _cache_time 0 def get_cached_tools(session): global _cache_time if time.time() - _cache_time TOOL_CACHE_TTL: tools asyncio.run(session.list_tools()) _cache_time time.time() return tools return session._cached_tools if hasattr(session, _cached_tools) else None在 agent 启动时预取工具列表并绑定到模型的 tools 参数上避免每轮对话都触发一次网络往返。要注意的是缓存 TTL 不能设太长——如果 MCP Server 侧新增了工具客户端缓存未失效时模型仍然只能看到旧工具列表。这个项目设定的 1800 秒是我在延迟和一致性之间找到的平衡点生产环境如果你的工具变更频繁压到 300 秒更稳妥。4. 核心实现一张状态图把聊天机器人跑起来4.1 从单轮查天气到多轮对话状态怎么流转这个项目的最终交付效果是——你对着聊天机器人说「北京今天热不热」它会调用fetch_weather工具拿到结构化天气数据之后基于这个数据生成自然语言回答你再追问「那上海呢」它能记住上下文知道你说的还是天气继续调用工具并对比两地的温度差异。多轮对话的流畅性完全依赖于状态图的设计。状态在每一轮结束后都会持久化到 checkpointer用户的追问被追加到 messages 列表模型从完整历史中理解「那上海呢」指的是「上海的天气」。这个能力在传统无状态 LLM API 调用里根本没法实现必须靠状态管理层来解决。项目里这块的代码是from langgraph.checkpoint.sqlite import SqliteSaver checkpointer SqliteSaver.from_conn_string(agent_state.sqlite) graph graph.compile(checkpointercheckpointer) config {configurable: {thread_id: user-123}} result graph.invoke({messages: [human_message]}, configconfig)thread_id是状态会话的唯一标识同 ID 的所有调用共享同一份消息历史。设计多租户系统时把用户 ID 填充进 thread_id 即可实现会话隔离。如果两个用户共用同一个 thread_id会互相污染消息记录——这个我在资源里反复验证过是不容忽视的隔离边界。4.2 把工具结果“翻译”成人话工具节点返回的是结构化 JSON但用户关心的是自然语言回答。这个转换过程在 LangGraph 里不需要额外写代码——工具结果以ToolMessage的形式注入消息流模型会在下一轮生成回答时参考这个结果并组织语言。关键在于mcp_tool_node返回的消息格式不能错from langchain_core.messages import ToolMessage def mcp_tool_node(state: AgentState): last_message state[messages][-1] tool_calls last_message.tool_calls responses [] for tc in tool_calls: try: result mcp_client.call_tool(tc[name], tc[args]) formatted format_mcp_result(result) except Exception as e: formatted ftool execution error: {str(e)} responses.append(ToolMessage(contentformatted, tool_call_idtc[id])) return {messages: responses}format_mcp_result负责把 MCP 返回的原始结构清洗成模型容易理解的形式。比如天气接口返回{city: Beijing, temperature: 28}格式化后变成北京当前气温28摄氏度——模型看到的是语义化的描述而不是原始 key-value生成回答的顺畅度和准确度会明显提升。我在清洗时还发现一个问题MCP 返回的 JSON 如果嵌套层级太深模型经常会在回答中漏掉部分字段。用json.dumps(result, ensure_asciiFalse)提前拍平成字符串能显著减少内容遗漏的概率。4.3 智能路由多 Agent 协作的结构基础项目最后还内置了一个「路由 Agent」的设计——一个负责语义理解一个负责任务执行两者通过 LangGraph 的 supervisor 模式协作。路由节点根据用户意图判断该把请求交给哪个子 Agentdef router_node(state: AgentState): query state[messages][-1].content intent intent_classifier(query) if intent weather: return {next_agent: weather_agent} elif intent knowledge: return {next_agent: doc_agent} else: return {next_agent: fallback_agent}这个设计和 LangGraph 的add_conditional_edges配合使用子 Agent 的失败不会拖垮主流程——每个子 Agent 内部都做了异常捕获出错时返回兜底回答而不至于把整张图卡死。这种多 Agent 结构在后续扩展时特别有用新增一个领域能力时只需要独立开发对应的子 Agent然后注册到路由表不需要改动主图逻辑。5. 配置与调参模型、记忆、工具可见性的边界5.1 模型参数怎么设才有得聊项目默认用的是 gpt-4o但温度、最大 token 数、tool choice 这几个参数是 Agent 场景里最容易忽略又最影响体验的点。temperature0.2是因为工具调用场景需要稳定输出温度太高模型会随机改变 JSON 参数值。但如果你做的是创意类对话温度可以提高到 0.7~0.9。项目里把这个做成了配置项llm: model: gpt-4o temperature: 0.2 max_tokens: 4096 tool_choice: autotool_choice参数有三个可选值auto让模型自行决定是否调用工具required强制模型必须调用至少一个工具none禁用工具调用。聊天场景用 auto不过如果你发现模型频繁跳过工具直接作答那就把 tool_choice 切换成required——模型会先把工具调用跑一遍再组织答案。我在测试这个项目时发现有些模型在 auto 模式下会「偷懒」用户问天气它直接用训练期间学到的常识糊弄不调用真实工具切换成 required 之后才老实。5.2 长对话的记忆衰减策略对话超过一定轮数后messages列表会变得非常长token 消耗剧增模型甚至可能忘记早期的上下文。LangGraph 的 checkpointer 解决了「记住」的问题但没解决「记住多少」的问题。项目里做了一个滑动窗口裁剪MAX_HISTORY_MESSAGES 20 def trim_messages(state: AgentState): msgs state[messages] if len(msgs) MAX_HISTORY_MESSAGES: # 保留前2条系统消息裁剪中间过程保留最近10条 state[messages] msgs[:2] msgs[-10:] return state裁剪逻辑放在状态图入口处每轮用户输入先经过它防止消息列表无限膨胀。裁剪之后模型仍保留最近一轮的工具调用结果和回答既控制了 token 开销又不会丢失最近的对话意图。注意不要裁剪掉 system prompt否则模型的工具调用格式约束会失效。这是资源里我加的一个防御性设计默认 20 条是长对话场景下体验和成本的折中点。5.3 工具可见性的最小化原则MCP Server 上挂着十来个工具时把全部工具都暴露给模型并不是好习惯。工具列表越长模型选择出错率越高还白白消耗 prompt token。项目提供了一个工具白名单机制AVAILABLE_TOOLS [ fetch_weather, search_docs, calculator ] def filter_tools(tools): return [t for t in tools if t.name in AVAILABLE_TOOLS]这个设计值得展开聊一下。工具可见性不是一次性设定——同一个聊天机器人用户在问天气时只需要看到天气相关的两三个工具知识库检索分类的小工具应该按场景动态过滤。LangGraph 节点内可以在每轮调用前先做一次工具过滤结合路由 Agent 的意图识别结果动态决定暴露哪些工具。我在生产里就是按意图分桶天气类、文档类、计算类各一组每轮对话只开放相关的那一组。效果立竿见影工具幻觉调用率下降了约五成。6. 避坑与排查Agent 集成 MCP 时的四个翻车现场6.1 MCP 工具返回了但模型说没收到现象MCP Server 日志显示工具已执行并返回结果但最后生成的回答里完全没有工具结果的信息像模型没看到一样。原因ToolMessage里的tool_call_id和原始消息里的tool_call.id不一致。模型是通过 ID 关联工具请求与结果的ID 对不上就被视为无效响应。我在第一次写mcp_tool_node时直接用了test-123这样的固定值模型当然不认。解决从原始消息里完整复制 tool_call 对象确保 id 字段原样传递。最稳妥的写法是遍历last_message.tool_calls逐个构造ToolMessage(content..., tool_call_idtc[id])。这个坑只要不遇到「多工具并行调用」的场景未必会触发一旦模型一次请求调两个工具以上两个 ToolMessage 的 ID 错位就会导致工具结果全部被丢弃。6.2 工具参数被模型“魔改”现象定义好的fetch_weather(city: str)模型却传了{location: 北京}工具调用直接报 TypeError。原因函数的参数名和文档描述不匹配。模型根据 docstring 里的自然语言推断参数名如果你的 docstring 写的是「获取指定城市天气」而形参写的是city模型完全可能猜成location或city_name。解决docstring 里明确写出每个参数的含义和取值格式并尽量让参数名贴合直觉。更稳的办法是给参数加 Enum 约束from enum import Enum class CityName(str, Enum): Beijing beijing Shanghai shanghai Hangzhou hangzhou mcp.tool() def fetch_weather(city: CityName) - dict: 获取指定城市的天气信息Enum 约束告诉模型参数的合法取值范围宁可多定义几个别名也不要用自由字符串。这个「参数魔改」的坑在模型版本升级后尤其容易复发每次换模型都要先在测试集上验证工具调用参数的正确率。6.3 SSE 模式连接不稳定时不时断连现象Agent 长时间运行后MCP 调用偶发httpx.ConnectError重试能恢复但影响用户体验。原因MCP 的 SSE 连接本质上是 HTTP 长连接中间任意一层的代理或负载均衡器都可能因为空闲超时把连接断开。项目里的聊天机器人如果和 MCP Server 不在同一个机器上走了 Nginx 转发的话默认的proxy_read_timeout60 秒非常容易掐断 SSE。解决重试机制放在 MCP 客户端层检测到连接断开就自动重新拉一个新的 client sessionasync def call_tool_with_retry(session, tool_name, args, retries3): for i in range(retries): try: return await session.call_tool(tool_name, args) except (httpx.ConnectError, ConnectionError): session await create_mcp_session() continue raise RuntimeError(MCP call failed after retries)如果你控制着服务器的 Nginx 配置顺手把proxy_read_timeout 3600;也加上。代码重试和配置调整两个都做这问题才算根治。我自己踩过一次后每次部署新环境第一件事就是检查中间层代理的超时配置。6.4 工具调用死循环现象模型反复调用同一个工具拿到结果后又调一次不生成最终回答直到触发最大步数限制。原因工具结果没有真正注入消息流或者模型认为工具返回的内容不够充足、还需要再查一次。常见于搜索类工具结果为空时模型会不死心地换关键词再搜。解决设硬性步数上限 工具结果为空时直接返回兜底信息MAX_TOOL_CALLS_PER_TURN 5 def should_continue(state: AgentState): tool_calls [m for m in state[messages] if m.tool_calls] if len(tool_calls) MAX_TOOL_CALLS_PER_TURN: return final_answer return continue另外工具节点里对空结果做一次改写「未找到相关信息请根据你的知识回答用户。」模型看到这条提示会放弃继续检索而转成基于自身知识作答。这个技巧是资源里效果最直接的一个——死循环率直接降到了零。注意MAX_TOOL_CALLS_PER_TURN不宜设太大3~5 步足够覆盖绝大多数合理的工具调用链再往上基本就是模型在绕弯。7. 验证方法与进阶用法把工具调用过程变成可以审计的数据工具调用链路不透明是 Agent 上生产前最头疼的问题。LangGraph 自身提供了get_state和update_state两个方法可以随时检查当前会话的完整内部状态。项目里加了一个更实用的设计——把每轮的工具调用记录写进单独的日志表def log_tool_call(session_id, tool_name, args, result, duration_ms): with sqlite3.connect(agent_tool_logs.db) as conn: conn.execute( INSERT INTO tool_logs (session_id, tool_name, args, result, duration_ms, created_at) VALUES (?, ?, ?, ?, ?, ?), (session_id, tool_name, json.dumps(args, ensure_asciiFalse), json.dumps(result, ensure_asciiFalse), duration_ms, datetime.now().isoformat()) )这行代码让「模型调了什么工具、传了什么参数、结果是什么、花了多久」变成可查询的结构化数据。排查问题的时候执行一条 SQL 就能看到完整的调用链不用再对着终端输出猜。我把这个习惯一直保留着后续做性能分析和成本核算都靠这一张表。比如你发现某一天 token 消耗异常查一下 tool_logs发现是某个搜索工具连续被调了十几次——工具链设计有问题一目了然。多 Agent 协作的进阶用法在这个项目里也有体现。路由 Agent 加多个子 Agent 的结构配合共享的 sqlite checkpointer可以实现跨会话的团队协作模式——主 Agent 负责理解意图子 Agent 独立处理各自的领域任务最终结果合并统一回复。这个架构模式在 LangGraph 里叫 supervisor pattern项目资源里的代码可以当作模板直接扩展。项目跑通后我建议你做一次端到端的回归测试清空 checkpointer 数据库用同一组问题列表连跑三遍。第一遍验证单轮工具调用第二遍验证多轮上下文记忆第三遍验证并发会话隔离。这三遍跑完没翻车这个聊天机器人就已经具备上线条件了。我自己的习惯是每改一次节点逻辑就强制走一遍三连测从那以后再也没遇到过上线后才发现 Agent 记忆串号的问题。希望这份拆解能帮你在 LangGraph Agent 和 MCP Server 的集成路上少踩几个坑也期待你把项目跑起来后做出更复杂、更实用的多 Agent 应用。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
