简介这是一份面向中文开发者与AI应用研究者的LangGraph Agent与MCP Server集成智能聊天机器人项目源码适合希望深入掌握图结构智能体工作流、工具调用与协议服务端实现的中高级学习者。资源包共13个文件以9个Python源码为主辅以2个txt配置说明、1个md文档与1个gitignore压缩包约44KB涵盖同步与异步两套Agent实现、MCP工具适配器及FastMCP服务端脚本。项目以LangGraph构建节点、边与条件分支状态对象采用强类型Pydantic模型并配中文语义字段MCP Server兼容v1.0标准支持JSON-RPC流式传输、工具注册与动态加载预置学术检索、地图查询、知识库语义检索等中文友好工具。读者可据此理解多步骤推理、记忆管理与工具调用的可追溯设计并参考分层架构、Docker Compose编排与测试规范快速搭建可扩展的智能体系统。目前已有17人学习。1. 从一张架构图说起LangGraph Agent 接上 MCP Server 到底解决了什么很多人第一次听到「LangGraph Agent 与 MCP Server 集成」脑子里浮现的是两个陌生名词的拼接。换个具体场景你写了一个能查数据库、能发消息、能读文件的聊天机器人工具函数越堆越多每加一个外部能力就要改一遍 Agent 代码、重新注册一遍工具、重新调一遍 prompt。三个月后这个 Agent 变成了一坨谁都不敢动的意大利面。MCP Server 要解决的就是这件事——把「工具从哪来」和「Agent 怎么用工具」彻底解耦。MCPModel Context Protocol定义了一套标准协议任何实现了这个协议的服务端都能被 Agent 以统一方式发现和调用工具不再硬编码在 Agent 里而是像插件一样挂载。LangGraph 则是把 Agent 的执行流程从「一问一答」升级成「有状态、可循环、可中断」的图结构让多步推理、条件分支、人工介入这些真实场景有了落脚点。两者结合你得到的是一个工具可插拔、流程可编排、状态可追踪的聊天机器人骨架。这套方案适合已经写过基础 Agent、被工具管理折磨过、想让机器人真正接入生产系统的开发者。如果你还在纠结 LangGraph 和 LangChain 的区别一句话LangChain 偏链式调用和组件封装LangGraph 偏状态机和图编排做复杂 Agent 时后者更顺手。2. 把 MCP Server 跑起来从协议握手到第一个工具暴露2.1 MCP 的通信模型与 LangGraph 的接入点MCP 的核心是一个客户端-服务端模型。服务端负责暴露三类能力tools可调用的函数、resources可读取的数据、prompts预置的提示模板。客户端负责连接服务端、拉取能力列表、发起调用。LangGraph 在这里扮演的是「客户端宿主」的角色——它不需要知道工具内部怎么实现只需要拿到工具的名称、描述和参数 schema然后把这些包装成 LangGraph 能识别的 ToolNode。通信层常见两种方式stdio标准输入输出适合本地进程和 SSEServer-Sent Events适合远程服务。本地开发我一般先用 stdio调试直观日志直接打在终端部署到多实例环境再换 SSE。这里有个容易翻车的点MCP 的 stdio 模式下服务端的 stdout 被协议占用你如果往 stdout 打调试日志协议直接解析失败。血泪经验——所有日志走 stderr。2.2 用 Python 写一个最小可用的 MCP Server下面是一个暴露「查询天气」和「发送通知」两个工具的 MCP Server 骨架。依赖官方 Python SDK安装命令先给上pip install mcp langgraph langchain-openai服务端代码# mcp_server.py import asyncio import json import sys from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent # 创建服务端实例名字会出现在客户端的能力列表里 app Server(demo-tools) app.list_tools() async def list_tools() - list[Tool]: 返回本服务端暴露的所有工具定义 return [ Tool( nameget_weather, description查询指定城市的当前天气, inputSchema{ type: object, properties: { city: {type: string, description: 城市名称} }, required: [city], }, ), Tool( namesend_notice, description向指定频道发送一条文本通知, inputSchema{ type: object, properties: { channel: {type: string}, message: {type: string}, }, required: [channel, message], }, ), ] app.call_tool() async def call_tool(name: str, arguments: dict) - list[TextContent]: 根据工具名分发实际逻辑 if name get_weather: city arguments[city] # 真实项目这里换成 API 调用示例直接返回 result {city: city, temp: 26, condition: 晴} return [TextContent(typetext, textjson.dumps(result, ensure_asciiFalse))] if name send_notice: # 日志必须走 stderrstdout 留给协议 print(f[notice] {arguments[channel]}: {arguments[message]}, filesys.stderr) return [TextContent(typetext, textsent)] raise ValueError(funknown tool: {name}) async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: asyncio.run(main())逻辑说明list_tools返回的inputSchema是 JSON Schema 格式LangGraph 侧会用它生成工具的参数校验call_tool是唯一入口按 name 分发。参数说明Server(demo-tools)里的名字只是标识不影响调用TextContent的type目前常用text返回结构化数据时把 dict 序列化成字符串即可。注意print的filesys.stderr这是 stdio 模式下不翻车的关键。2.3 在 LangGraph 里把 MCP 工具挂成节点LangGraph 侧需要做三件事启动 MCP 客户端子进程、拉取工具列表、把工具转成 LangChain Tool 并塞进 ToolNode。# agent.py import asyncio from langgraph.prebuilt import ToolNode, create_react_agent from langchain_openai import ChatOpenAI from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_core.tools import StructuredTool async def build_agent(): # 1. 以子进程方式启动 MCP Server params StdioServerParameters(commandpython, args[mcp_server.py]) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 2. 拉取工具列表 tools_resp await session.list_tools() lc_tools [] for t in tools_resp.tools: # 3. 把 MCP 工具包装成 LangChain 可识别的 Tool async def _call(_namet.name, **kwargs): res await session.call_tool(_name, kwargs) return res.content[0].text lc_tools.append(StructuredTool.from_function( coroutine_call, namet.name, descriptiont.description, args_schemat.inputSchema, )) llm ChatOpenAI(modelgpt-4o-mini, temperature0) agent create_react_agent(llm, lc_tools) return agent if __name__ __main__: agent asyncio.run(build_agent()) out asyncio.run(agent.ainvoke({messages: [(user, 北京天气怎么样)]})) print(out[messages][-1].content)逻辑说明stdio_client负责拉起子进程并建立双向通道session.list_tools()拿到的是 MCP 原生 Tool 对象字段和 LangChain Tool 不完全一致需要手动映射。参数说明args_schema直接传 MCP 的inputSchema即可LangChain 内部会做转换create_react_agent是 LangGraph 预置的 ReAct 图生产环境建议自己用StateGraph搭方便加中断和人工审核节点。这段代码跑通后你往 MCP Server 里加新工具Agent 侧一行不用改。3. 状态图编排让聊天机器人记住上下文、支持多轮工具调用3.1 用 StateGraph 替代 create_react_agent 的三个理由create_react_agent适合验证但真实项目里我几乎都会换成手写StateGraph。原因有三第一ReAct 的循环条件写死了你想在工具调用前插入人工确认节点做不到第二状态结构固定为 messages 列表想额外维护用户画像、会话摘要这些字段很别扭第三出错时的重试和降级策略没法细粒度控制。手写图虽然多写几十行但换来的是完全可控。LangGraph 的核心概念是 State共享状态、Node处理函数、Edge跳转条件。每个节点读 State、返回增量更新框架负责合并。这个模型对聊天机器人特别合适——消息历史、工具结果、中间推理都能放进 State随时可以 checkpoint 到数据库实现断点续聊。3.2 一个带工具调用和人工确认的完整图# graph_agent.py from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langgraph.checkpoint.memory import MemorySaver from langchain_core.messages import HumanMessage, AIMessage, ToolMessage class AgentState(TypedDict): # add_messages 是 reducer新消息追加而非覆盖 messages: Annotated[list, add_messages] need_confirm: bool def build_graph(llm, tools): tool_map {t.name: t for t in tools} def think(state: AgentState): LLM 推理节点决定是回复还是调工具 resp llm.invoke(state[messages]) return {messages: [resp]} def route(state: AgentState): 条件边有 tool_calls 就走工具否则结束 last state[messages][-1] if getattr(last, tool_calls, None): return tools return END async def run_tools(state: AgentState): 工具执行节点 last state[messages][-1] results [] for call in last.tool_calls: tool tool_map[call[name]] out await tool.ainvoke(call[args]) results.append(ToolMessage(contentstr(out), tool_call_idcall[id])) return {messages: results} g StateGraph(AgentState) g.add_node(think, think) g.add_node(tools, run_tools) g.set_entry_point(think) g.add_conditional_edges(think, route, {tools: tools, END: END}) g.add_edge(tools, think) # 工具执行完回到推理 return g.compile(checkpointerMemorySaver())逻辑说明think节点调用 LLMroute检查最后一条消息是否带tool_calls有就跳tools节点执行完再回到think形成循环直到 LLM 不再请求工具。参数说明add_messages是内置 reducer保证消息按顺序追加MemorySaver是内存级 checkpoint生产环境换成SqliteSaver或 Postgres 实现持久化thread_id在调用时通过 config 传入用于区分不同会话。调用方式config {configurable: {thread_id: user-001}} result await graph.ainvoke( {messages: [HumanMessage(content帮我查下上海天气然后通知 ops 频道)]}, configconfig, )同一个thread_id多次调用历史自动累积这就是多轮对话的底层机制。3.3 工具调用的错误处理与重试边界工具调用失败是常态——网络超时、参数错误、下游服务挂了。LangGraph 里我一般做三层防护第一层在工具函数内部 try/except把异常转成结构化的错误消息返回给 LLM让模型自己决定要不要重试或换方案第二层在run_tools节点加超时控制单个工具超过 10 秒直接返回超时消息避免整个图卡死第三层用 LangGraph 的retry策略对特定节点配置重试次数。from langgraph.pregel import RetryPolicy g.add_node(tools, run_tools, retryRetryPolicy(max_attempts2))注意重试只对幂等工具有意义。发消息、下单这类有副作用的工具重试前必须做去重否则用户会收到两条通知。这个坑我在生产环境踩过后来所有写操作工具都加了idempotency_key参数。4. 避坑与排查MCP 集成里最容易翻车的五个地方4.1 工具列表拉取为空现象Agent 启动后 LLM 说「我没有可用的工具」但 MCP Server 日志显示已启动。原因通常是session.initialize()没调用或者list_tools装饰器写成了同步函数。MCP SDK 要求app.list_tools()修饰的是 async 函数写成普通 def 不会报错但永远返回空。解决检查装饰器下的函数是否async def并在list_tools里加一行 stderr 日志确认被调用。4.2 stdio 模式下进程挂起无响应现象Agent 调用工具后一直等待没有返回也没有报错。原因是 MCP Server 里某处往 stdout 写了非协议内容客户端解析器卡住。常见触发点第三方库的 print、logging 默认输出到 stdout、异常堆栈被打到 stdout。解决在 Server 入口最前面重定向sys.stdout到 stderr或者配置 logging 的 handler 明确指向 stderr。这个问题的排查成本极高因为没有任何报错信息只能靠二分法注释代码定位。4.3 工具参数 schema 不兼容现象LLM 生成的参数格式正确但调用时报 validation error。原因是 MCP 的inputSchema用了 JSON Schema 的某些字段如$ref、oneOfLangChain 的转换层不支持。解决保持 schema 扁平只用type、properties、required、description这几个基础字段。嵌套对象尽量拆成多个平级参数枚举用enum而不是anyOf。4.4 多轮对话状态丢失现象第一轮工具调用正常第二轮 LLM 忘记了之前的工具结果。原因是thread_id每次调用都变了或者 compile 时没传 checkpointer。解决确认graph.compile(checkpointer...)已配置且每次ainvoke传入相同的thread_id。另外注意ToolMessage必须带tool_call_id否则 reducer 合并时消息顺序会乱。4.5 工具调用死循环现象LLM 反复调用同一个工具图一直在 think 和 tools 之间循环token 消耗飙升。原因是工具返回的结果 LLM 无法理解或者工具描述有歧义导致模型误判。解决在 State 里加一个tool_call_count字段超过阈值比如 5 次强制走 END 并返回兜底回复同时优化工具返回内容错误信息要明确写「此操作失败请勿重试改为告知用户」。5. 进阶技巧用自定义日志和可观测性把黑匣子打开MCP Server 的日志管理是很多人忽略的一环。默认情况下你只能看到 Agent 侧的调用记录工具内部发生了什么完全是黑匣子。我的做法是在 MCP Server 里建一个独立的 logger输出结构化 JSON 到 stderr每条日志带trace_id、tool_name、duration_ms、status四个字段。Agent 侧在调用工具前生成trace_id并通过参数传入这样一次请求从 LLM 推理到工具执行到结果返回全链路可以串起来。import logging, json, sys, time logger logging.getLogger(mcp.tools) handler logging.StreamHandler(sys.stderr) handler.setFormatter(logging.Formatter(%(message)s)) logger.addHandler(handler) logger.setLevel(logging.INFO) async def call_tool(name, arguments): trace_id arguments.pop(_trace_id, unknown) start time.time() try: result await dispatch(name, arguments) logger.info(json.dumps({ trace_id: trace_id, tool: name, duration_ms: int((time.time() - start) * 1000), status: ok, })) return result except Exception as e: logger.info(json.dumps({ trace_id: trace_id, tool: name, duration_ms: int((time.time() - start) * 1000), status: error, error: str(e), })) raise这套日志配合 LangGraph 的 checkpoint 记录排查问题时能精确到「第几轮对话、哪个工具、耗时多少、返回了什么」。另一个值得投入的点是给图加interrupt——在工具执行前暂停等人工确认后再继续。LangGraph 原生支持这个模式配置interrupt_before[tools]即可适合涉及资金、权限、对外发送这类高风险操作。验证集成是否健康我习惯跑一个冒烟脚本依次调用每个工具、检查返回结构、确认日志有对应记录、验证多轮对话状态正确累积。这个脚本进 CI每次改 MCP Server 或图结构都跑一遍比事后翻日志高效得多。做这类集成项目最深的体会是协议层的东西看着简单真正吃时间的是边界情况——进程生命周期、编码、超时、并发。把这些处理干净剩下的就是业务逻辑那才是真正值得花心思的地方。希望帮到你。本文还有配套的精品资源点击获取