LangChain + MCP + Agent 实战:从零搭建可落地的AI工具调用系统

发布时间:2026/9/4 14:30:09
LangChain + MCP + Agent 实战:从零搭建可落地的AI工具调用系统 这次我们来看一个非常具体的组合LangChain MCP Agent。如果你已经在折腾 AI Agent大概率会碰到两个问题一是模型本身不知道怎么操作外部工具二是外部工具越接越多以后代码变得乱七八糟。MCP 的价值在于把“模型接入工具”的方式统一成一套协议而 LangChain / LangGraph 是上层用来编排 Agent 逻辑的框架。两者结合起来才是目前生产环境中比较常见的 Agent 技术栈。这篇文章不打算讲太多概念直接从代码入手。你会看到如何用 Python 写一个最小的 MCP 工具服务如何把它注册进 LangChain让模型自动决定调用哪个工具加入 LangGraph 之后怎么控制多步骤流程常见的调试断点放在哪里以及真要发布到生产环境时要注意什么。文末还整理了一组 Agent 相关的面试题方便你对照检查掌握程度。如果你正准备做 Agent 开发、MCP Server 集成或者想理顺 LangChain 和 LangGraph 之间的关系这篇文章可以直接收藏。1. 核心概念速览先给一张速览表把三个名词的关系说清楚。能力项LangChainMCPAgent定位Agent 应用开发框架工具调用协议AI 应用的一种交互模式解决什么问题编排模型、提示词、工具调用逻辑让不同 AI 应用统一接入外部工具/数据让模型自主决策“下一步做什么”核心对象Chain、Agent、Retriever、MemoryTool、Server、ClientLLM Tools 执行循环典型 APIcreate_agent/AgentExecutorload_mcp_tools/ClientSessionmodel.invoke()适合场景快速构建 RAG、聊天助手、工作流跨平台共享工具避免重复开发需要多步推理、多工具调用的任务生产落地难度中低中中高关系上可以这么理解Agent 解决“模型怎么思考”LangChain 提供整套应用框架MCP 提供统一的工具标准。实际开发中你会先写若干个 MCP Server然后在 LangChain Agent 里加载这些工具最终得到一个能自主决策的 AI 应用。这里还要区分一个高频概念function calling 和 MCP。function calling 是模型层面的一种能力告诉你“这一步需要调用哪个函数、参数是什么”MCP 则是应用层面的工具接入协议负责把函数传进模型上下文、统一鉴权、统一日志。两者并不冲突MCP 底层依然依赖模型的 function calling 能力把工具请求解析出来。2. 适用场景与使用边界先说适合使用这套技术栈的场景。企业知识库助手需要同时查询数据库、搜索内部文档、读取第三方 APIMCP Server 统一暴露这些数据源。自动化运维 Agent可以把告警查询、日志检索、工单创建封装成多个 MCP 工具由 Agent 按需调用。研发效率工具把代码仓库操作、文档生成、代码审查命令封装成工具集成到 IDE 或内部工具平台。数据分析 Agent让模型自主选择 SQL 查询、Python 解释、图表生成等工具典型的是最近流行的“AI 数据分析师”。也有不那么合适的场景。单个工具的简单调用直接写个if分支处理函数调用即可不需要引入完整的 Agent 和 MCP。流程固定、不允许模型自由决策的步骤比如订单支付前的强校验建议用普通代码控制流程不要交给 Agent 做“自主判断”。实时性要求极高的场景也要谨慎。MCP Agent 的链路比单次 HTTP 请求多多了工具注册、模型推理和循环调用延迟通常在数秒级。合规边界建议通过 MCP 接入内部系统前先做权限最小化设计不让模型获得任意命令执行权限。涉及用户敏感数据、企业机密时优先走私有化部署不把原始数据发送给第三方大模型服务。工具执行前需要记录审计日志尤其是写操作、删除操作和外部请求。建议先读后写写操作必须二次确认。3. 环境准备与前置条件本文的示例代码使用 Python 3.10依赖管理建议用uv或venv。你需要以下依赖。pip install langchain langchain-openai langgraph langchain-mcp-adapters mcp[cli]其中langchain核心框架。langchain-openaiOpenAI 兼容模型的接入包。langgraph基于图的 Agent 编排框架LangChain 官方推荐的生产级 Agent 运行时。langchain-mcp-adapters把 MCP 工具转换为 LangChain 工具的关键适配器。mcp[cli]MCP Python SDK用于编写和调试 MCP Server。如果你没有 OpenAI 的 API Key也可以使用兼容 OpenAI 接口的本地模型服务比如通过 vLLM 或 Ollama 启动的本地服务。只要模型支持 function calling 或 tool calling就可以继续往下走。本地模型服务举例from langchain_openai import ChatOpenAI llm ChatOpenAI( modelqwen2.5:14b, base_urlhttp://127.0.0.1:11434/v1, api_keyollama, temperature0, )这里要注意一个问题很多本地小模型虽然支持 OpenAI 兼容接口但 tool calling 能力不稳定。建议先选择 14B 以上参数量的模型或者直接用大厂 API不然 Agent 很容易出现“调用了不存在的参数”这类问题。4. 手写一个最小的 MCP 工具服务为了理解 LangChain 接入 MCP 的工具流程我们先用 FastMCP 写一个最简单的 MCP Server。这个服务器提供两个函数一个做加法一个返回当前时间。4.1 MCP Server 最小实现# mcp_server_demo.py from mcp.server.fastmcp import FastMCP mcp FastMCP(namedemo-tools) mcp.tool() def add_two_numbers(a: float, b: float) - float: 计算两个数字的和。 return a b mcp.tool() def get_current_city() - str: 返回当前城市的示例信息演示无参数工具。 return 北京 if __name__ __main__: mcp.run()mcp.run() 默认会启动 stdio transport。所谓 stdio transport就是通过标准输入输出与客户端通信。Agent 主进程会启动一个子进程来运行这个脚本然后通过 stdin / stdout 与它交换 JSON-RPC 消息。这个文件写完以后可以先单独启动验证一下。不过直接运行没有太多可观察输出因为 stdio 模式下数据是走管道传输的不会打印在控制台。为了验证功能可以加临时日志或者用 MCP Inspector 这类调试工具连接。# 使用 mcp 官方调试工具 mcp dev mcp_server_demo.py如果你希望 MCP Server 以 HTTP SSE 方式暴露给远程客户端可以改为if __name__ __main__: mcp.run(transportsse)生产环境中通过 HTTP 暴露时通常不会直接使用 SDK 自带配置而是由 Nginx 或 API 网关做转发与鉴权。关于这点后面生产部署小节再展开。4.2 为什么不直接用普通 Python 函数很多初学者会疑惑既然只是两个函数直接 import 进 LangChain 不就行了吗为什么还要包一层 MCP普通函数的问题是只能用在一个项目里。如果公司里已经有多个团队分别写了数据库查询工具、搜索工具、告警工具每个团队语言不同、接口风格不同、鉴权方式不同你在 LangChain 里接 A 团队的工具是一套代码接 B 团队的工具又得重写一套。MCP 统一了“工具描述、参数 Schema、调用返回”的格式。LangChain 里只要写一次适配器后续所有团队的工具都可以无差别加载。这就是 LangChain 官方推出 langchain-mcp-adapters 的动机不是在 LangChain 里新发明一套工具规范而是去适配 MCP 这个标准。5. 在 LangChain 里注册 MCP 工具并驱动 Agent5.1 用适配器加载 MCP 工具现在我们写一个客户端脚本连接上面创建的 MCP Server并把工具转换成 LangChain 可识别格式。# langchain_client_demo.py import asyncio from langchain_mcp_adapters.tools import load_mcp_tools from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client server_params StdioServerParameters( commandpython, args[mcp_server_demo.py], ) async def main(): 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) for tool in tools: print(f工具名称: {tool.name}) print(f工具描述: {tool.description[:100]}) print(f参数 Schema: {tool.args}) print(- * 50) asyncio.run(main())预期输出应该能看到两个工具add_two_numbers和get_current_city。这里有一个很关键的设计MCP Tool 在被转换成 LangChain 工具后工具名会保留 MCP Server 里定义的名称。也就是说工具命名规范和参数描述在 MCP Server 端就要写好不要指望 LangChain 侧帮你优化。5.2 创建一个能调用工具的 Agent我们接下来创建一个create_tool_calling_agent形式的 Agent。这个 Agent 会在模型的 function calling 能力驱动下自动判断是否需要调用 MCP 工具。import asyncio from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain_mcp_adapters.tools import load_mcp_tools from langchain_openai import ChatOpenAI from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client server_params StdioServerParameters( commandpython, args[mcp_server_demo.py], ) prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手可以调用工具来回答问题。), (human, {input}), (placeholder, {agent_scratchpad}), ]) async def main(): 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) llm ChatOpenAI( modelgpt-4o-mini, temperature0, ) agent create_tool_calling_agent(llm, tools, promptprompt) executor AgentExecutor(agentagent, toolstools, verboseTrue) result await executor.ainvoke({ input: 请计算 12.5 和 7.3 的和然后把结果用中文告诉我。 }) print(result[output]) asyncio.run(main())如果一切正常你会看到控制台先输出模型生成的 tool call 请求然后真正执行add_two_numbers最后输出回答。这里要特别提醒一个重点temperature参数。Agent 场景下推荐设为 0 或接近 0。因为模型需要准确生成结构化 tool call JSON温度过高会导致函数名拼错、参数类型不对。如果你发现模型经常“发明工具”先检查两件事temperature 是否太高工具描述是否产生歧义。5.3 使用 MultiServerMCPClient 同时接入多个服务实际开发时不会只接一个 MCP Server。LangChain 官方还提供一个MultiServerMCPClient支持一次初始化多个 MCP 连接比如同时接入公司内部的搜索 MCP 和数据库 MCP。from langchain_mcp_adapters.client import MultiServerMCPClient client MultiServerMCPClient( { search: { command: python, args: [search_mcp_server.py], transport: stdio, }, database: { url: http://127.0.0.1:8000/sse, transport: sse, }, } ) tools client.get_tools()多个 MCP 工具集中到一个列表之后直接传入 AgentExecutor 即可。工具数量增加以后必须注意工具名冲突。比如两个 MCP Server 里都有get_user_info工具直接加载会导致 LangChain 工具注册冲突随机只识别其中一个。建议在 MCP Server 端就用前缀规范化命名。6. 用 LangGraph 编排更复杂的 Agent 流程LangChain 团队目前的推荐是生产环境尽量用 LangGraph 而不是老式AgentExecutor。原因很直接AgentExecutor 是一个封装好的黑盒循环逻辑基本固定LangGraph 允许你自定义图谱把每个步骤拆分成节点方便控制条件分支、人工审批、重试策略。6.1 LangGraph 和 LangChain 的区别面试和实际选型时都会问这个问题。简单讲LangChain 是更宽泛的应用开发框架核心抽象是 Chain、Tool、Retriever、Memory。LangGraph 是 LangChain 生态中的有向图编排框架专门解决 Agent 状态流、循环和分支控制。同一个 LangChain 项目里可以不加 LangGraph 完成简单任务但一旦 Agent 有分支判断、循环限制、人工介入LangGraph 就是更合适的选择。6.2 LangGraph Agent 节点示例下面用 LangGraph 写一个简单流程模型决定是否需要工具如果需要就执行 MCP 工具然后把结果返回给模型如果不需要就直接输出。from typing import Literal from langchain_mcp_adapters.client import MultiServerMCPClient from langchain_openai import ChatOpenAI from langgraph.graph import END, START, StateGraph from langgraph.prebuilt import ToolNode from typing_extensions import TypedDict llm ChatOpenAI(modelgpt-4o-mini, temperature0) class AgentState(TypedDict): messages: list def create_tools(): client MultiServerMCPClient( { demo: { command: python, args: [mcp_server_demo.py], transport: stdio, }, } ) return client.get_tools() def should_continue(state: AgentState) - Literal[tools, __end__]: last_message state[messages][-1] if last_message.tool_calls: return tools return __end__ def call_model(state: AgentState): messages state[messages] response llm.invoke(messages) return {messages: [response]} def build_agent(tools): tool_node ToolNode(tools) workflow StateGraph(AgentState) workflow.add_node(agent, call_model) workflow.add_node(tools, tool_node) workflow.add_edge(START, agent) workflow.add_conditional_edges( agent, should_continue, { tools: tools, __end__: END, }, ) workflow.add_edge(tools, agent) return workflow.compile()这个示例的核心是should_continue这个条件边。每次模型返回后检查最后一条消息是否带tool_calls。如果有就进入工具执行节点执行完成后回到模型节点让它基于工具结果继续生成。如果模型认为不需要更多工具调用就结束流程。LangGraph 的好处是把循环逻辑显式地展示出来而不是藏在 AgentExecutor 内部。调试时可以准备在任意节点前插入断点或者用一个“人工确认”节点拦截敏感工具调用。7. 调试技巧Agent 开发里最耗时间的不是写代码而是排流程问题。下面是我在实战中比较常用的一套调试手段。7.1 先验证 MCP 工具本身在接入 LangChain 之前先单独确认 MCP Server 能把工具暴露出来。如果客户端都加载不到工具后续 Agent 肯定也调不了。验证方法有两种第一种使用 MCP 自带调试工具。mcp dev mcp_server_demo.py第二种用 Python 快速连接并打印工具列表。7.2 开启 verbose 和回调日志在 agent executor 上开启 verbose通常能看每步输入输出。但要更细的追踪需要在模型调用前后插入回调。from langchain_core.callbacks import BaseCallbackHandler class DebugCallback(BaseCallbackHandler): def on_llm_start(self, serialized, prompts, **kwargs): print([LLM 输入], prompts) def on_tool_start(self, serialized, input_str, **kwargs): print([工具调用], serialized.get(name), input_str) def on_tool_end(self, output, **kwargs): print([工具返回], output)把这个 handler 传给ainvoke的config参数即可。executor.ainvoke( {input: 计算 1 2}, config{callbacks: [DebugCallback()]}, )通过回调日志你能非常清楚地看到模型输出文本还是工具调用请求、调用了哪个工具、工具传入了什么参数、MCP Server 返回了什么结果。大多数调不到工具的问题都能在这一层定位。7.3 MCP Server 进程日志单独输出stdio 模式下MCP Server 运行在子进程中。如果 Server 内部崩溃或抛异常LangChain 侧通常只看到一个 generic error很难定位。我的习惯是开发期不让 MCP Server 作为子进程由 LangChain 启动而是先在一个终端手动跑起 SSE 模式的 Server把 stdout 日志打到控制台LangChain 侧连接这个 SSE 地址。这样两边日志都能保留问题快速定位。# 终端 1启动 MCP Server python mcp_server_demo_http.py # 终端 2设置环境变量让 LangChain 连接 SSR 端点 export PYTHONPATH. python your_agent_script.py7.4 排查工具参数 Schema 错误模型调用工具时最常见的报错是参数类型不正确、缺少必填字段、多传了不存在的参数。这类问题大部分来自 MCP Tool 描述和参数 schema 不够清晰。建议工具描述写成命令式模型能直接理解的那种参数加上默认值和 type hint。mcp.tool() def query_weather(city: str 北京) - str: 查询指定城市天气城市用中文全称。 ...8. 生产级部署建议开发阶段能跑通 Agent离生产可用还差得很远。下面这部分针对 LangChain MCP 服务上线时容易踩的问题。8.1 部署形态生产环境一般有两种部署形态。第一种是单体部署Agent 服务、MCP Server 都在同一个容器或进程组里通过 stdio 通信。优点是部署简单适用于内部工具缺点是 MCP 工具与 Agent 强耦合工具更新必须重新发布 Agent。第二种是分离部署MCP Server 以独立服务形式运行在 HTTP SSE 或 Streamable HTTP 上Agent 通过远程 HTTP 调用。工具团队可以独立发版。建议正规团队采用这种。使用 SSE 时系统要注意这个传输虽然兼容常见代理但连接生命周期较长。如果 Nginx 默认没有开 SSE 缓冲需要关闭代理缓冲location /sse { proxy_pass http://mcp_service:8000; proxy_buffering off; proxy_cache off; proxy_read_timeout 1h; }8.2 Agent 无状态化与独立会话存储Agent 服务建议保持无状态会话记录存到外部存储。LangChain 可以用 Redis、PostgreSQL 等做 checkpointer。LangGraph 里只需给compile传一个 checkpointer 就能实现对话中断恢复与状态持久化。from langgraph.checkpoint.memory import MemorySaver app workflow.compile(checkpointerMemorySaver())生产环境不要用 MemorySaver因为重启即丢。建议使用langgraph-checkpoint配合 Redis 或 PostgreSQL 存储状态。8.3 批量任务的编排如果你需要批量处理需求注意不要让 Agent 最高频次的循环在一次执行里占据一个核心加长时间。建议做法消息先进队列用 Celery、Redis Queue 或异步任务框架消费在 Agent 层做并发限制避免同时开启几十个 Process 导致 MCP Server 子进程爆炸。伪代码如下import asyncio from concurrent.futures import Semaphore sem Semaphore(5) async def process_one(input_text: str): async with sem: return await executor.ainvoke({input: input_text})8.4 API 服务层调用的封装给前端或业务系统提供 API 时不要把 agent 内部依赖暴露给外部调用。一般建议在 Agent 外层再做 FastAPI 接口层输入业务参数输出最终回复Agent 内部各步骤不可见。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): question: str session_id: str app.post(/agent/chat) async def chat(req: ChatRequest): result await executor.ainvoke( {input: req.question}, config{configurable: {thread_id: req.session_id}}, ) return {answer: result[output]}生产环境必须对 API 做鉴权、限流、请求体和返回体序列化并且只把需要的字段暴露出去。9. 常见问题与排查方法这里整理一份 Agent 开发落地过程中出现频率较高的排查清单。问题现象可能原因排查方式解决方案LangChain 加载不到 MCP 工具MCP Server 启动失败或 Python 路径不对单独运行 Server检查 stdout 日志和 exit code改用绝对路径确认 python 环境一致Agent 不调用工具模型无 function calling 能力或提示词不明确查看回调日志确认模型是否理解工具用途换支持工具调用的模型优化工具描述工具调用参数缺失工具描述和参数 schema 不准确打印工具传入的 args增加参数说明、默认值和 type hintMCP Server 返回乱码或 JSON 解析失败编码不一致或 Server 额外 print 到 stdout检查 MCP Server 内是否残留 printstdio 模式下不要直接 print用 logging显存/OOM 或请求超时本地模型推理速度过慢加长模型调用超时和重试次数使用更小的模型或云端 APIAgent 陷入死循环条件边判断有问题模型无限 tool_calls添加最大迭代次数限制在 LangGraph 节点内加 recursion_limit端口冲突多个 MCP SSE Server 使用同一端口netstat或lsof查找进程更换端口或统一端口管理API 频繁出现跨域或连接中断Nginx 代理没有处理长连接查看服务端错误日志关闭代理缓冲适当调大超时老式 AgentExecutor 运行报错工具列表为空打印 len(tools)确认 MCP 客户端初始化完成10. 面试真题把 Agent 相关的常见考点整理成一组问答方便你自查。问题 1LangChain 和 LangGraph 有什么区别LangChain 是一个框架提供 prompt、llm、tool、memory 等基础构件LangGraph 是基于状态图的 Agent 编排引擎允许自定义循环、分支、中断和恢复。简单任务用 LangChain 就够涉及复杂流程编排时 LangGraph 更可控。问题 2MCP 解决了什么问题TCP/HTTP/stdio 传输有什么不同MCP 将工具接入标准化为 client-server 协议避免每个 AI 应用重复开发一套工具对接。stdio 适合本地子进程HTTP 和 SSE 适合远程服务。问题 3Agent 的 ReAct 流程是什么ReAct 即 Reason Act 的循环模型分析当前任务根据可用工具生成行动观察工具返回再进入下一步推理直到得到最终答案。LangChain 的 tool calling agent 和 LangGraph 的条件边循环都复用这个思路。问题 4如何保证 Agent 的工具调用不出越权或安全问题定义工具时尽量减小权限不让 Agent 拿到全局命令或写权限执行阶段做审计对敏感动作设置人工确认步骤允许随时中断。问题 5多 Agent 场景下怎么共享工具优先把所有工具统一注册为 MCP Server各 Agent 通过 MCP client 按需加载使用前缀管理工具命名空间。LangGraph 中可以让不同节点配置不同工具子集避免单个模型上下文被塞满无用工具。面试真题想要面试考察不出错关键是真实理解 MCP 协议、工具注册流程、模型循环决策机制和 LangGraph 条件边的关系。只背概念是不行的建议你至少动手跑一遍本文第 4 章和第 5 章的代码。11. 最佳实践与总结这块内容一次性做完整把能踩的坑先帮你扫平。11.1 工具先设计后编码MCP 工具是给模型“看”的不是给人用的。工具描述、参数 schema、返回值结构直接影响模型能不能正确调用。写工具前先问自己模型能从这个工具名判断出用途吗参数名是模型熟悉的自然语义返回值适合直接拼接到后续推理吗信息不足的话模型就算调对工具也接不住结果。11.2 建议准备策略稳定的生产配置建议第一次运行小样本验证工具连接和模型输出不要直接上完整流程。保留一份最小可运行的配置包含一个简单的 MCP Server、一个 client、一个 agent。模型类型、base_url、MCP Server 的连接参数全部放到配置文件或环境变量不能写死在代码里。任务输入和输出做结构化日志字段至少要包含 session_id、时间戳、调用的工具、延迟。11.3 可控性优先于花哨任何让 Agent 变得更自主的功能都用旁路验证的方式添加。比如新增一个“读取数据库全部表”工具先加读权限、白名单和超时不要把它做成 execute-any-sql 的工具。11.4 常见最优实践清单建议做成团队文档持续维护所有 MCP 工具使用显式前缀如 search_query、db_execute_read、db_execute_write。写操作工具在参数层默认以事务和 SQL 白名单控制。用日志追踪每个 tool call 的请求参数和执行耗时。模型温度设为 0必要时关闭 reasoning 模式的开关。登录一次会话后保留带状态的 session而不是每次 restart 新图。Agent 最外层 API 至少需要限流、鉴权、身份审计三层防护。从代码量来看LangChain MCP Agent 的门槛不在于框架本身而在于工程化的习惯。我已经看到很多 demo 能将工具跑起来但工具对接的稳定性、日志可观测性和权限控制才是生产落地的真正分水岭。建议你先按照本文第 4 章和第 5 章跑通最小示例再尝试给 MCP Server 增加一个真实工具比如调用天气 API 或查询数据的只读接口。跑通之后再看 LangGraph 的条件边把循环机制换成你自己定义的状态机。这套链路的调试和部署方案一旦熟悉后续接任何新工具都会很快。