LangChain 和 MCP 这两个词过去一年被反复提起但真正能把它们落到代码里跑通的人并不多。很多教程要么只讲 LangChain 的 Chain 和 Prompt要么只介绍 MCP 的协议概念没人说清楚“这两个东西为什么需要组合在一起以及组合之后 Agent 的开发方式到底变成了什么样”。这次我们直接围绕“LangChain MCP 从入门到代码实战”这条线把核心概念、环境准备、MCP Server 编写、LangChain 工具接入、API 封装和批量任务全部过一遍最终给出一套能直接改造成自己项目的工程骨架。先说这篇文章覆盖的重点第一LangChain 在 Agent 场景里到底负责什么Chain、Memory、Tool、Agent 这些组件如何协作第二MCP 的协议设计解决了工具接入的什么问题它和 Function Calling、Skill、Plugin 有什么区别第三如何用 Python 快速写一个 MCP Server并让 LangChain Agent 通过 MCP 协议调用它的工具第四如何把 Agent 封装成 HTTP API并设计一个可靠的批量任务处理流程。适合的读者是已经写过简单 LangChain 调用但没跑通工具调用或者想在项目中引入 MCP 标准化工具接入方式的开发者。文中的代码以示例为主具体版本和路径需要按你本机环境微调。涉及大模型调用时请使用你自己有权限的 API Key并遵守服务商的使用条款。1. 核心能力速览在展开代码之前先用一张表把 LangChain MCP 这套组合的能力边界列清楚能力项说明项目定位LLM 应用编排框架 标准化工具接入协议核心组件LangChainChain / Agent / Memory / Tool、MCPClient / Server / Tool主流编程语言Python本文以 Python 3.10 为例模型接入方式任意 OpenAI 兼容接口或通过 LangChain 官方集成接入各类大模型是否支持本地模型支持取决于 LangChain 对应模型封装是否支持你的推理服务是否支持远程工具支持MCP Server 可运行在本地进程也可以部署为远程服务是否支持批量任务支持可以通过队列任务循环或异步任务框架实现是否支持 API 封装支持LangChain Agent 可封装为 FastAPI / Flask 服务启动方式命令行启动 MCP Server Python 脚本运行 Agent无强制 WebUI适合场景企业知识库工具、数据库查询助手、API 聚合助手、自动化工作流需要说明的是LangChain 本身不提供大模型能力它负责的是“流程编排”MCP 本身也不提供具体业务功能它负责的是“工具描述和调用标准化”。两者组合后真正干活的是大模型和你暴露给模型的那些工具。2. LangChain 基础Chain、Memory、Agent 各管什么很多新手第一次接触 LangChain 时会被 Prompt Template、Chat Model、OutputParser、Memory、Retriever、Agent 这些概念淹没。其实 LangChain 的核心只有一条主线它把“大模型对话”拆成了可编排的单元让开发者可以用代码控制模型调用前后的逻辑。2.1 Chain串起提示词、模型和输出解析最基础的用法是 PromptTemplate LLM。比如我们要让模型把用户问题改写成标准查询语句from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI prompt ChatPromptTemplate.from_messages([ (system, 你是一个数据处理助手把用户输入改写成结构化查询语句。), (human, {input}), ]) model ChatOpenAI(modelgpt-4o-mini, temperature0) chain prompt | model result chain.invoke({input: 查一下上个月销量最高的商品}) print(result.content)这里prompt | model就是一条最简单的 Chain。LangChain 真正有价值的地方在于你可以在 Chain 中间插入自定义函数from langchain_core.runnables import RunnableLambda def post_process(text: str) - str: return text.strip().replace(查询语句, ) chain prompt | model | RunnableLambda(lambda x: post_process(x.content))这种管道式写法让“提示词拼接 - 模型调用 - 结果处理”变成一条可视化链路便于调试和复用。2.2 Memory让多轮对话有上下文普通 Chain 是无状态的每次invoke都是独立调用。如果要做多轮对话就需要把历史消息传给模型。LangChain 的 Memory 组件负责管理这部分历史from langchain.memory import ConversationBufferMemory from langchain_core.chat_history import InMemoryChatMessageHistory history InMemoryChatMessageHistory() history.add_user_message(我的账号是 10086) history.add_ai_message(已记录需要我帮你做什么) memory ConversationBufferMemory(chat_memoryhistory, return_messagesTrue)在实际开发中多轮记忆需要注意 token 长度。对话历史如果无限增长最终会撑爆上下文窗口。LangChain 提供了多个策略比如滑动窗口、摘要记忆等但最稳妥的方式还是在工程层控制历史长度比如只保留最近 10 轮。2.3 Agent让模型决定下一步调用什么工具Agent 是 LangChain 里最接近“智能体”概念的部分。它的工作方式是模型根据用户的问题决定调用哪个工具、传入什么参数然后观察工具返回值再决定是继续调用还是给出最终答案。整个过程由 AgentExecutor 或 LangGraph 控制循环。一个完整 Agent 至少包含三部分工具列表每个工具包含名称、描述、参数 Schema。模型支持工具调用Function Calling或 ReAct 推理的模型。执行循环不断“思考 - 调用工具 - 获取结果 - 再思考”直到结束。这里的关键点是“工具描述”。如果工具描述写得含糊模型可能不知道该用哪个工具如果参数 Schema 不准确模型可能传错参数。MCP 的作用正是在这一层它把工具定义和调用方式统一成标准协议让 Agent 不用为每个系统单独写适配器。3. MCP 是什么模型上下文协议MCP全称 Model Context Protocol是一个开放的、用于让大模型应用与外部工具、数据源通信的协议。它的目标很简单当你想让智能体连接数据库、文件系统、浏览器、设计稿或内部 API 时不再需要为每个系统手写一套工具封装而是通过统一的 MCP Server 暴露能力通过统一的 MCP Client 接入 LangChain、Claude 或其他支持 MCP 的应用。3.1 MCP 的三大抽象Tools、Resources、PromptsMCP 协议定义了三种能力类型Tools可被模型调用的函数比如“查询天气”“读取文件”“执行 SQL”。Resources可被读取的数据源比如文件内容、数据库表结构。Prompts可复用的提示词模板用于引导模型完成特定任务。在 LangChain 集成场景里最常用的是 Tools。模型看到的是“有哪些工具可用、每个工具的参数是什么、如何调用结果”实际执行逻辑都封装在 MCP Server 中。一个典型的 MCP Server 配置可以放在进程内启动也可以写在mcp.json中{ mcpServers: { query-server: { command: python, args: [path/to/mcp_server.py], env: {} } } }这种配置文件在 Claude Desktop、Dify、Trae 等支持 MCP 的工具中通用。也就是说你写好的 MCP Server 可以同时被 LangChain、Dify 和很多其他工具复用不需要为每个平台单独改造。3.2 MCP 与 Function Calling、Skill、Plugin 的区别这部分是热词搜索里关注度很高的问题也是很多开发者容易混淆的地方。Function Calling 是模型服务商提供的一种接口能力让模型可以输出结构化工具调用请求。它主要解决“模型怎么表达调用意图”但不解决“工具怎么被注册、怎么被发现”。Plugin 是特定平台比如浏览器插件的扩展机制耦合了宿主平台的 API。Skill 通常是某个智能体平台对“能力包”的封装可能是提示词、工具脚本和配置的组合平台相关。MCP 是独立于模型和平台的协议。它用统一方式描述工具名、参数、返回格式并且支持本地进程和远程服务两种传输方式。更直接地说MCP 可以与 Function Calling 共存模型通过 Function Calling 输出工具调用意图而工具的真实能力由 MCP Server 提供。LangChain Agent 负责组织模型和 MCP 工具的交互。3.3 LangChain 和 LangGraph 怎么选在热词里也能看到很多人在搜 LangGraph 和 LangChain 的区别。简单理解LangChain 是一整套开发框架LangGraph 是面向复杂 Agent 状态流、循环和并发的图编排框架。如果你的业务流程是固定的分支判断用 LangChain Chain 就够如果 Agent 需要动态规划多步任务、递归、人工确认LangGraph 更合适。MCP 工具接入在两者中都适用本文以 LangChain 传统 Agent 为主方便入门理解。4. 环境准备与前置条件下面这套环境是运行后续示例的最小组合。我这里以 Python 3.10 或更高版本为例操作系统不限Windows、Linux、macOS 都可以。4.1 创建虚拟环境强烈建议用虚拟环境隔离依赖python -m venv .venv source .venv/bin/activate # Linux/macOS .venv\Scripts\activate # Windows PowerShell4.2 安装依赖pip install langchain langchain-openai mcp fastmcp langchain-mcp-adapters httpx说明langchain-mcp-adapters是一个将 MCP 工具转换成 LangChain 工具的适配层具体包名和版本请以官方仓库为准。如果安装失败可以分步安装先装mcp和langchain再装适配器。4.3 准备模型访问LangChain 的ChatOpenAI可以连接绝大多数 OpenAI 兼容接口既可以是官方服务也可以是本地部署的 vLLM、Ollama 等。使用时设置环境变量export OPENAI_API_KEYsk-xxxx export OPENAI_API_BASEhttps://your-endpoint/v1如果使用国内服务商请从服务商控制台获取正确的 Base URL 和 Key。如果使用本地模型更稳妥的判断是等待模型完全加载后再发起请求避免连接超时。4.4 磁盘和端口MCP Server 作为本地子进程启动时需要保证端口不被占用。如果你使用stdio传输方式实际上不依赖端口如果使用sse或streamable-http传输方式则需要提前检查端口。建议本地调试先用 stdio减少网络因素干扰。5. MCP Server 实战写一个本地查询工具为了演示完整链路我们写一个非常简单的 MCP Server它暴露两个工具一个是“获取用户信息”一个是“查询最近订单”。实际项目里你可以把这两个函数改成内部数据库或 API 调用。下面使用 FastMCP 库来简化 MCP Server 的开发。FastMCP 是 MCP Python SDK 之上的高层封装可以让我们用装饰器快速定义工具。# mcp_query_server.py from fastmcp import FastMCP mcp FastMCP(QueryServer) USERS {1001: {name: 张三, level: VIP}, 1002: {name: 李四, level: 普通}} ORDERS { 1001: [{order_id: A001, amount: 299, time: 2026-01-01}], 1002: [{order_id: B002, amount: 129, time: 2026-01-02}], } mcp.tool() def get_user_info(user_id: str) - dict: 根据用户 ID 返回用户基本信息 return USERS.get(user_id, {error: user not found}) mcp.tool() def get_recent_orders(user_id: str, limit: int 5) - list: 根据用户 ID 返回最近订单列表 orders ORDERS.get(user_id, []) return orders[:limit] if __name__ __main__: mcp.run(transportstdio)在这个文件里mcp.tool()会自动生成工具的 JSON Schema工具的 docstring 和参数类型注解会作为模型的参考描述。也就是说模型不是直接执行 Python 函数而是看到“工具名、参数、描述、返回结构”然后决定是否调用。启动这个 MCP Server直接在终端运行python mcp_query_server.py如果一切正常程序会以 stdio 模式等待父进程传 JSON-RPC 消息。这个进程本身不会打印任何内容但可以被 MCP Client 发现和调用。你也可以在mcp.json里配置这个 Server方便其他工具加载{ mcpServers: { query-server: { command: python, args: [mcp_query_server.py], env: {} } } }这样就完成了“从工具代码到协议暴露”的第一步。6. LangChain 接入 MCP 工具并运行 Agent真正让 LangChain 和 MCP 协同工作的关键是把 MCP Server 提供的工具暴露给 LangChain Agent。这里使用langchain-mcp-adapters中的load_mcp_tools来加载工具然后把这些工具交给 Agent。6.1 加载 MCP Server 中的工具# agent_mcp_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 from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import PromptTemplate server_params StdioServerParameters( commandpython, args[mcp_query_server.py], ) async def run_agent(): 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(modelgpt-4o-mini, temperature0) prompt PromptTemplate.from_template( 你是一个查询助手可以调用工具来查询用户信息和订单。 用户问题{input} 工具列表{tools} 工具名称{tool_names} 请按步骤处理最终用中文回答。 ) agent create_react_agent(model, tools, prompt) executor AgentExecutor(agentagent, toolstools, verboseTrue) result await executor.ainvoke( {input: 用户 1001 是谁最近有什么订单} ) print(result[output]) if __name__ __main__: asyncio.run(run_agent())这段代码的流程是通过stdio_client启动 MCP Server 子进程。建立ClientSession并执行initialize()握手。调用load_mcp_tools(session)把 MCP Server 里定义的get_user_info和get_recent_orders转成 LangChain 工具对象。创建 ReAct Agent并赋予工具列表。通过AgentExecutor执行多轮推理模型自行决定是否调用工具。如果运行成功控制台会输出类似“用户 1001 是张三等级 VIP最近订单有 A001金额 299 元”的回答。6.2 让 Agent 支持多轮对话实际业务很少是单轮问答更多是连续追问。比如用户先问“1001 是谁”再问“他最近买了什么”。这种情况下必须保留对话历史。LangChain 官方推荐的复杂 Agent 方案是 LangGraph但我们可以在入门阶段用 ChatPromptTemplate 把历史拼进入口消息from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder prompt ChatPromptTemplate.from_messages([ (system, 你是查询助手。), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ])然后把chat_history传给ainvoke。这里要注意历史消息必须是消息对象列表不能只传字符串。6.3 使用 LangGraph 作为 Agent 执行框架如果你的业务需要更精细的控制比如某个工具调用失败后重新规划或者需要人工确认LangGraph 比 AgentExecutor 更适合。LangGraph 可以接受 MCP 返回的 LangChain 工具并编排成一个带状态的图from langgraph.prebuilt import create_react_agent graph create_react_agent(model, tools)创建后可像普通 Chain 一样调用但内部状态管理更清晰支持 checkpoint 和断点续跑。从 LangChain 切换到 LangGraph 并不冲突工具和模型封装都是可复用的。7. 接口 API 与批量任务设计当 Agent 在本地跑通后下一步通常是把它封装成 HTTP API供前端、定时任务或其他服务调用。7.1 用 FastAPI 封装 Agent用一个 Python 文件启动 FastAPI 服务内部复用之前的 Agent 逻辑。注意MCP Server 子进程的创建和销毁比较耗时建议在服务启动时创建一次会话而不是每个请求都重新拉起子进程。# api_server.py from fastapi import FastAPI from pydantic import BaseModel import asyncio app FastAPI() class QueryRequest(BaseModel): question: str user_id: str app.post(/agent/query) async def query_agent(req: QueryRequest): result await run_agent_once(req.question, req.user_id) return {answer: result}这里run_agent_once是封装了上一节 Agent 调用逻辑的函数你可以把它放在一个独立模块里。实际部署时建议把 MCP 会话做成异步单例避免并发请求导致端口连接冲突。7.2 curl 调用示例启动服务后用 curl 测试curl -X POST http://127.0.0.1:8000/agent/query \ -H Content-Type: application/json \ -d {question: 用户 1001 是谁, user_id: 1001}预期返回{ answer: 用户 1001 是张三VIP 用户。 }如果你使用的是streamable-http传输方式MCP Server 本身也可以暴露为远程服务那么 LangChain Client 就不需要在同一个机器上启动子进程而是通过 HTTP 连接。这在多服务部署时更灵活。7.3 批量任务队列设计批量任务的关键不是用for循环一个接一个调用而是要区分“需要上下文关联的任务序列”和“相互独立的批量任务”。如果是独立批量任务最简单的方式是用异步并发或消息队列import asyncio from tqdm import tqdm async def process_batch(questions: list[str], max_concurrency: int 5): semaphore asyncio.Semaphore(max_concurrency) async def worker(q: str): async with semaphore: return await run_agent_once(q) results [] for coro in tqdm(asyncio.as_completed([worker(q) for q in questions]), totallen(questions)): result await coro results.append(result) return results这里的run_agent_once每次调用会走完整的 Agent 推理。如果批量请求需要调用外部 API必须加上超时、重试和失败隔离超时每个请求设置timeout120。重试遇到临时错误尝试 2 次即可不要无限重试。失败隔离单条失败不影响整批结果把失败原因记到日志里。如果任务之间需要共享上下文比如先查用户再查订单建议把它们设计成一个 Agent 多步规划或者通过 LangGraph 维护同一个状态对象而不是拆成多个独立请求。8. 资源占用与性能观察LangChain MCP 这套组合的资源占用主要来自三个地方MCP Server 进程、LangChain Agent 执行过程、大模型推理或 API 调用。MCP Server 如果是本地 Python 子进程内存占用通常在几十 MB 到几百 MB 之间具体取决于你 import 的库和工具复杂度。如果你的 MCP Server 内部加载了重模型资源占用就会明显上升。这一点只能依赖实际运行来观察不建议照搬别人的数字。在代理执行过程中LangChain Agent 会做多次模型调用。以 ReAct Agent 为例模型可能要“思考 - 调用工具 - 观察结果 - 再思考”多轮每一轮都是一次 API 请求。这意味着相比单次普通对话Agent 场景的 token 消耗会翻倍甚至更多。性能优化的几个方向精简工具数量工具太多会让模型选择变慢甚至选错。精简工具描述长描述会增加 prompt 长度导致 token 上升。使用更快的小模型如果业务不复杂大模型负责规划小模型负责总结。缓存对于稳定数据查询可以在 MCP Server 层做结果缓存。并发限制API 服务商通常有 QPS 限制批量任务要控制并发数。另一个观察点是 Python 异步并发。标准 MCP Client 的stdio_client在多次调用时会有一些通信开销如果 Agent 要频繁调用工具建议把耗时工具放在 MCP Server 端做批量聚合而不是让模型多次小粒度调用。9. 常见问题与排查方法实战中LangChain MCP 最容易出问题的不是概念而是环境、协议和服务生命周期。下面这张表覆盖了高频故障问题现象可能原因排查方式解决方案MCP Server 连接失败子进程启动失败或 Python 命令找不到单独运行python mcp_server.py看是否有报错在配置中使用绝对路径 Python 路径工具未注册到 Agentload_mcp_tools返回空列表打印 tools 列表检查 MCP Server 是否有mcp.tool()确认 MCP Server 初始化成功且工具函数已定义模型选择错误工具工具描述不清晰或参数 Schema 不准确查看 Agent verbose 日志观察模型输出意图优化 docstring 和参数类型增加示例值Agent 无限循环模型反复调用工具但结果无法满足结束条件设置max_iterations参数用 AgentExecutor 时配置max_iterations5请求超时模型 API 响应慢或 MCP 工具耗时过长在工具内打印耗时日志增加 timeout或优化工具执行逻辑端口冲突使用 SSE 或 Streamable HTTP 传输时端口被占用lsof -i :8000或netstat -ano查看端口更换端口或关闭残留进程API Key 无效环境变量未生效打印os.getenv(OPENAI_API_KEY)检查配置文件和加载顺序批量任务卡住并发过高或 API 限流查看服务端错误日志降低并发数增加重试退避策略MCP 配置不生效mcp.json中 command 不是绝对路径检查配置格式路径不能含特殊字符使用绝对路径并确认可执行权限9.1 Windows 下创建 MCP Server 的注意点很多 Windows 用户会遇到command: python无法启动的问题。这通常是 Python 命令不在系统 PATH 中或者系统里同时存在多个 Python 环境。更稳妥的做法是在终端执行where python拿到绝对路径。在mcp.json的command中填写绝对路径例如C:\\Python310\\python.exe。确保虚拟环境中的包能被该 Python 解释器找到。如果你使用 Dify 或 Trae 等客户端添加本地 MCP 服务配置方式类似重点关注启动目录和 Python 解释器路径。9.2 MCP 工具注册后参数校验失败当模型调用 MCP 工具时参数必须符合 JSON Schema。比如get_recent_orders要求user_id是字符串如果模型传了数字 1001严格模式下会校验失败。解决方案是在工具函数内部做宽松转换或在 Schema 中标记参数类型为 string并在描述中写明“user_id 为字符串类型”。10. 最佳实践与合规建议工程化使用 LangChain MCP不能只停留在“能跑通”阶段还要考虑稳定性、可维护性和合规边界。10.1 保持 MCP Server 的独立性MCP Server 应该是纯粹的工具服务不依赖 LangChain。它只暴露协议不管上层是 LangChain 还是 Dify。这样你升级 LangChain 或换 Agent 框架时MCP Server 不需要改动。建议把 MCP Server 单独放在一个目录或独立模块里和 Agent 代码解耦。10.2 工具描述要面向模型写工具描述不是给人看的是给模型看的。写工具说明时要注意描述用途而不是实现细节。说明输入参数的格式和范围。给出典型示例减少模型猜测。提示边界条件比如“用户不存在时返回 error”。10.3 日志、追踪与异常隔离Agent 调用链路比较长建议每个阶段都输出结构化日志进入 Agent 的用户问题。模型选择了哪个工具。工具实际传入参数。工具返回结果摘要。最终回答。这样排查问题时能快速定位是模型选择错误还是工具返回异常还是最终生成出现问题。10.4 合规与授权如果你的 MCP 工具会访问数据库、客户信息、文件系统或第三方 API必须遵守最小权限原则。不要给 Agent 提供无限制的 SQL 执行工具尤其是生产环境。涉及用户隐私数据的查询要在 MCP Server 层做鉴权不能只靠模型自觉。对外部版权内容、人物肖像、商业秘密等敏感信息务必获得合法授权后再使用。10.5 版本锁定LangChain 生态迭代速度很快今天能跑通的代码几个月后可能因为 API 变更失效。建议在项目的requirements.txt或pyproject.toml中锁定关键依赖版本并保留一份最小可运行配置方便快速回滚。11. 总结与下一步LangChain MCP 组合的意义在于把“模型能力”和“系统工具”之间的连接从定制化变成标准化。LangChain 负责编排 Agent 的思考和行动MCP 负责让工具以统一协议被任意支持方调用。对于开发者来说先跑通一个最简单的 MCP Server再让 LangChain Agent 调用它是理解整套体系最快的路径。如果你正准备上手建议按这个顺序做先写一个不依赖任何外部服务的 MCP Server比如查内存字典然后让 LangChain Agent 调用它观察 verbose 日志理解模型的工具调用过程接着把 Memory 和历史消息加进去最后再封装 API 和批量任务。第一篇代码可以小但链路要完整。最容易踩的坑是跳过 MCP Server 单独调试直接在 LangChain 里加载工具等到出现问题时搞不清是协议层问题还是模型选择问题。所以第一步一定要先把 MCP Server 单独跑通再接入 Agent。下一步可以继续玩的方向包括用 LangGraph 替换 AgentExecutor让 Agent 支持更复杂的多步规划把 MCP Server 部署到远程通过 SSE 或 HTTP 连接把批量任务改造成 Celery 或任务队列在 MCP Server 中接入数据库、Figma、Playwright 等真实业务工具。这套流程跑熟之后后面接任何新工具都会很顺。 
长河君 · 2026/8/30 9:44:21 · 阅读 1.2k LangChain + MCP 实战：从入门到代码构建智能体工具调用框架
LangChain 和 MCP 这两个词过去一年被反复提起但真正能把它们落到代码里跑通的人并不多。很多教程要么只讲 LangChain 的 Chain 和 Prompt要么只介绍 MCP 的协议概念没人说清楚“这两个东西为什么需要组合在一起以及组合之后 Agent 的开发方式到底变成了什么样”。这次我们直接围绕“LangChain MCP 从入门到代码实战”这条线把核心概念、环境准备、MCP Server 编写、LangChain 工具接入、API 封装和批量任务全部过一遍最终给出一套能直接改造成自己项目的工程骨架。先说这篇文章覆盖的重点第一LangChain 在 Agent 场景里到底负责什么Chain、Memory、Tool、Agent 这些组件如何协作第二MCP 的协议设计解决了工具接入的什么问题它和 Function Calling、Skill、Plugin 有什么区别第三如何用 Python 快速写一个 MCP Server并让 LangChain Agent 通过 MCP 协议调用它的工具第四如何把 Agent 封装成 HTTP API并设计一个可靠的批量任务处理流程。适合的读者是已经写过简单 LangChain 调用但没跑通工具调用或者想在项目中引入 MCP 标准化工具接入方式的开发者。文中的代码以示例为主具体版本和路径需要按你本机环境微调。涉及大模型调用时请使用你自己有权限的 API Key并遵守服务商的使用条款。1. 核心能力速览在展开代码之前先用一张表把 LangChain MCP 这套组合的能力边界列清楚能力项说明项目定位LLM 应用编排框架 标准化工具接入协议核心组件LangChainChain / Agent / Memory / Tool、MCPClient / Server / Tool主流编程语言Python本文以 Python 3.10 为例模型接入方式任意 OpenAI 兼容接口或通过 LangChain 官方集成接入各类大模型是否支持本地模型支持取决于 LangChain 对应模型封装是否支持你的推理服务是否支持远程工具支持MCP Server 可运行在本地进程也可以部署为远程服务是否支持批量任务支持可以通过队列任务循环或异步任务框架实现是否支持 API 封装支持LangChain Agent 可封装为 FastAPI / Flask 服务启动方式命令行启动 MCP Server Python 脚本运行 Agent无强制 WebUI适合场景企业知识库工具、数据库查询助手、API 聚合助手、自动化工作流需要说明的是LangChain 本身不提供大模型能力它负责的是“流程编排”MCP 本身也不提供具体业务功能它负责的是“工具描述和调用标准化”。两者组合后真正干活的是大模型和你暴露给模型的那些工具。2. LangChain 基础Chain、Memory、Agent 各管什么很多新手第一次接触 LangChain 时会被 Prompt Template、Chat Model、OutputParser、Memory、Retriever、Agent 这些概念淹没。其实 LangChain 的核心只有一条主线它把“大模型对话”拆成了可编排的单元让开发者可以用代码控制模型调用前后的逻辑。2.1 Chain串起提示词、模型和输出解析最基础的用法是 PromptTemplate LLM。比如我们要让模型把用户问题改写成标准查询语句from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI prompt ChatPromptTemplate.from_messages([ (system, 你是一个数据处理助手把用户输入改写成结构化查询语句。), (human, {input}), ]) model ChatOpenAI(modelgpt-4o-mini, temperature0) chain prompt | model result chain.invoke({input: 查一下上个月销量最高的商品}) print(result.content)这里prompt | model就是一条最简单的 Chain。LangChain 真正有价值的地方在于你可以在 Chain 中间插入自定义函数from langchain_core.runnables import RunnableLambda def post_process(text: str) - str: return text.strip().replace(查询语句, ) chain prompt | model | RunnableLambda(lambda x: post_process(x.content))这种管道式写法让“提示词拼接 - 模型调用 - 结果处理”变成一条可视化链路便于调试和复用。2.2 Memory让多轮对话有上下文普通 Chain 是无状态的每次invoke都是独立调用。如果要做多轮对话就需要把历史消息传给模型。LangChain 的 Memory 组件负责管理这部分历史from langchain.memory import ConversationBufferMemory from langchain_core.chat_history import InMemoryChatMessageHistory history InMemoryChatMessageHistory() history.add_user_message(我的账号是 10086) history.add_ai_message(已记录需要我帮你做什么) memory ConversationBufferMemory(chat_memoryhistory, return_messagesTrue)在实际开发中多轮记忆需要注意 token 长度。对话历史如果无限增长最终会撑爆上下文窗口。LangChain 提供了多个策略比如滑动窗口、摘要记忆等但最稳妥的方式还是在工程层控制历史长度比如只保留最近 10 轮。2.3 Agent让模型决定下一步调用什么工具Agent 是 LangChain 里最接近“智能体”概念的部分。它的工作方式是模型根据用户的问题决定调用哪个工具、传入什么参数然后观察工具返回值再决定是继续调用还是给出最终答案。整个过程由 AgentExecutor 或 LangGraph 控制循环。一个完整 Agent 至少包含三部分工具列表每个工具包含名称、描述、参数 Schema。模型支持工具调用Function Calling或 ReAct 推理的模型。执行循环不断“思考 - 调用工具 - 获取结果 - 再思考”直到结束。这里的关键点是“工具描述”。如果工具描述写得含糊模型可能不知道该用哪个工具如果参数 Schema 不准确模型可能传错参数。MCP 的作用正是在这一层它把工具定义和调用方式统一成标准协议让 Agent 不用为每个系统单独写适配器。3. MCP 是什么模型上下文协议MCP全称 Model Context Protocol是一个开放的、用于让大模型应用与外部工具、数据源通信的协议。它的目标很简单当你想让智能体连接数据库、文件系统、浏览器、设计稿或内部 API 时不再需要为每个系统手写一套工具封装而是通过统一的 MCP Server 暴露能力通过统一的 MCP Client 接入 LangChain、Claude 或其他支持 MCP 的应用。3.1 MCP 的三大抽象Tools、Resources、PromptsMCP 协议定义了三种能力类型Tools可被模型调用的函数比如“查询天气”“读取文件”“执行 SQL”。Resources可被读取的数据源比如文件内容、数据库表结构。Prompts可复用的提示词模板用于引导模型完成特定任务。在 LangChain 集成场景里最常用的是 Tools。模型看到的是“有哪些工具可用、每个工具的参数是什么、如何调用结果”实际执行逻辑都封装在 MCP Server 中。一个典型的 MCP Server 配置可以放在进程内启动也可以写在mcp.json中{ mcpServers: { query-server: { command: python, args: [path/to/mcp_server.py], env: {} } } }这种配置文件在 Claude Desktop、Dify、Trae 等支持 MCP 的工具中通用。也就是说你写好的 MCP Server 可以同时被 LangChain、Dify 和很多其他工具复用不需要为每个平台单独改造。3.2 MCP 与 Function Calling、Skill、Plugin 的区别这部分是热词搜索里关注度很高的问题也是很多开发者容易混淆的地方。Function Calling 是模型服务商提供的一种接口能力让模型可以输出结构化工具调用请求。它主要解决“模型怎么表达调用意图”但不解决“工具怎么被注册、怎么被发现”。Plugin 是特定平台比如浏览器插件的扩展机制耦合了宿主平台的 API。Skill 通常是某个智能体平台对“能力包”的封装可能是提示词、工具脚本和配置的组合平台相关。MCP 是独立于模型和平台的协议。它用统一方式描述工具名、参数、返回格式并且支持本地进程和远程服务两种传输方式。更直接地说MCP 可以与 Function Calling 共存模型通过 Function Calling 输出工具调用意图而工具的真实能力由 MCP Server 提供。LangChain Agent 负责组织模型和 MCP 工具的交互。3.3 LangChain 和 LangGraph 怎么选在热词里也能看到很多人在搜 LangGraph 和 LangChain 的区别。简单理解LangChain 是一整套开发框架LangGraph 是面向复杂 Agent 状态流、循环和并发的图编排框架。如果你的业务流程是固定的分支判断用 LangChain Chain 就够如果 Agent 需要动态规划多步任务、递归、人工确认LangGraph 更合适。MCP 工具接入在两者中都适用本文以 LangChain 传统 Agent 为主方便入门理解。4. 环境准备与前置条件下面这套环境是运行后续示例的最小组合。我这里以 Python 3.10 或更高版本为例操作系统不限Windows、Linux、macOS 都可以。4.1 创建虚拟环境强烈建议用虚拟环境隔离依赖python -m venv .venv source .venv/bin/activate # Linux/macOS .venv\Scripts\activate # Windows PowerShell4.2 安装依赖pip install langchain langchain-openai mcp fastmcp langchain-mcp-adapters httpx说明langchain-mcp-adapters是一个将 MCP 工具转换成 LangChain 工具的适配层具体包名和版本请以官方仓库为准。如果安装失败可以分步安装先装mcp和langchain再装适配器。4.3 准备模型访问LangChain 的ChatOpenAI可以连接绝大多数 OpenAI 兼容接口既可以是官方服务也可以是本地部署的 vLLM、Ollama 等。使用时设置环境变量export OPENAI_API_KEYsk-xxxx export OPENAI_API_BASEhttps://your-endpoint/v1如果使用国内服务商请从服务商控制台获取正确的 Base URL 和 Key。如果使用本地模型更稳妥的判断是等待模型完全加载后再发起请求避免连接超时。4.4 磁盘和端口MCP Server 作为本地子进程启动时需要保证端口不被占用。如果你使用stdio传输方式实际上不依赖端口如果使用sse或streamable-http传输方式则需要提前检查端口。建议本地调试先用 stdio减少网络因素干扰。5. MCP Server 实战写一个本地查询工具为了演示完整链路我们写一个非常简单的 MCP Server它暴露两个工具一个是“获取用户信息”一个是“查询最近订单”。实际项目里你可以把这两个函数改成内部数据库或 API 调用。下面使用 FastMCP 库来简化 MCP Server 的开发。FastMCP 是 MCP Python SDK 之上的高层封装可以让我们用装饰器快速定义工具。# mcp_query_server.py from fastmcp import FastMCP mcp FastMCP(QueryServer) USERS {1001: {name: 张三, level: VIP}, 1002: {name: 李四, level: 普通}} ORDERS { 1001: [{order_id: A001, amount: 299, time: 2026-01-01}], 1002: [{order_id: B002, amount: 129, time: 2026-01-02}], } mcp.tool() def get_user_info(user_id: str) - dict: 根据用户 ID 返回用户基本信息 return USERS.get(user_id, {error: user not found}) mcp.tool() def get_recent_orders(user_id: str, limit: int 5) - list: 根据用户 ID 返回最近订单列表 orders ORDERS.get(user_id, []) return orders[:limit] if __name__ __main__: mcp.run(transportstdio)在这个文件里mcp.tool()会自动生成工具的 JSON Schema工具的 docstring 和参数类型注解会作为模型的参考描述。也就是说模型不是直接执行 Python 函数而是看到“工具名、参数、描述、返回结构”然后决定是否调用。启动这个 MCP Server直接在终端运行python mcp_query_server.py如果一切正常程序会以 stdio 模式等待父进程传 JSON-RPC 消息。这个进程本身不会打印任何内容但可以被 MCP Client 发现和调用。你也可以在mcp.json里配置这个 Server方便其他工具加载{ mcpServers: { query-server: { command: python, args: [mcp_query_server.py], env: {} } } }这样就完成了“从工具代码到协议暴露”的第一步。6. LangChain 接入 MCP 工具并运行 Agent真正让 LangChain 和 MCP 协同工作的关键是把 MCP Server 提供的工具暴露给 LangChain Agent。这里使用langchain-mcp-adapters中的load_mcp_tools来加载工具然后把这些工具交给 Agent。6.1 加载 MCP Server 中的工具# agent_mcp_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 from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import PromptTemplate server_params StdioServerParameters( commandpython, args[mcp_query_server.py], ) async def run_agent(): 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(modelgpt-4o-mini, temperature0) prompt PromptTemplate.from_template( 你是一个查询助手可以调用工具来查询用户信息和订单。 用户问题{input} 工具列表{tools} 工具名称{tool_names} 请按步骤处理最终用中文回答。 ) agent create_react_agent(model, tools, prompt) executor AgentExecutor(agentagent, toolstools, verboseTrue) result await executor.ainvoke( {input: 用户 1001 是谁最近有什么订单} ) print(result[output]) if __name__ __main__: asyncio.run(run_agent())这段代码的流程是通过stdio_client启动 MCP Server 子进程。建立ClientSession并执行initialize()握手。调用load_mcp_tools(session)把 MCP Server 里定义的get_user_info和get_recent_orders转成 LangChain 工具对象。创建 ReAct Agent并赋予工具列表。通过AgentExecutor执行多轮推理模型自行决定是否调用工具。如果运行成功控制台会输出类似“用户 1001 是张三等级 VIP最近订单有 A001金额 299 元”的回答。6.2 让 Agent 支持多轮对话实际业务很少是单轮问答更多是连续追问。比如用户先问“1001 是谁”再问“他最近买了什么”。这种情况下必须保留对话历史。LangChain 官方推荐的复杂 Agent 方案是 LangGraph但我们可以在入门阶段用 ChatPromptTemplate 把历史拼进入口消息from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder prompt ChatPromptTemplate.from_messages([ (system, 你是查询助手。), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ])然后把chat_history传给ainvoke。这里要注意历史消息必须是消息对象列表不能只传字符串。6.3 使用 LangGraph 作为 Agent 执行框架如果你的业务需要更精细的控制比如某个工具调用失败后重新规划或者需要人工确认LangGraph 比 AgentExecutor 更适合。LangGraph 可以接受 MCP 返回的 LangChain 工具并编排成一个带状态的图from langgraph.prebuilt import create_react_agent graph create_react_agent(model, tools)创建后可像普通 Chain 一样调用但内部状态管理更清晰支持 checkpoint 和断点续跑。从 LangChain 切换到 LangGraph 并不冲突工具和模型封装都是可复用的。7. 接口 API 与批量任务设计当 Agent 在本地跑通后下一步通常是把它封装成 HTTP API供前端、定时任务或其他服务调用。7.1 用 FastAPI 封装 Agent用一个 Python 文件启动 FastAPI 服务内部复用之前的 Agent 逻辑。注意MCP Server 子进程的创建和销毁比较耗时建议在服务启动时创建一次会话而不是每个请求都重新拉起子进程。# api_server.py from fastapi import FastAPI from pydantic import BaseModel import asyncio app FastAPI() class QueryRequest(BaseModel): question: str user_id: str app.post(/agent/query) async def query_agent(req: QueryRequest): result await run_agent_once(req.question, req.user_id) return {answer: result}这里run_agent_once是封装了上一节 Agent 调用逻辑的函数你可以把它放在一个独立模块里。实际部署时建议把 MCP 会话做成异步单例避免并发请求导致端口连接冲突。7.2 curl 调用示例启动服务后用 curl 测试curl -X POST http://127.0.0.1:8000/agent/query \ -H Content-Type: application/json \ -d {question: 用户 1001 是谁, user_id: 1001}预期返回{ answer: 用户 1001 是张三VIP 用户。 }如果你使用的是streamable-http传输方式MCP Server 本身也可以暴露为远程服务那么 LangChain Client 就不需要在同一个机器上启动子进程而是通过 HTTP 连接。这在多服务部署时更灵活。7.3 批量任务队列设计批量任务的关键不是用for循环一个接一个调用而是要区分“需要上下文关联的任务序列”和“相互独立的批量任务”。如果是独立批量任务最简单的方式是用异步并发或消息队列import asyncio from tqdm import tqdm async def process_batch(questions: list[str], max_concurrency: int 5): semaphore asyncio.Semaphore(max_concurrency) async def worker(q: str): async with semaphore: return await run_agent_once(q) results [] for coro in tqdm(asyncio.as_completed([worker(q) for q in questions]), totallen(questions)): result await coro results.append(result) return results这里的run_agent_once每次调用会走完整的 Agent 推理。如果批量请求需要调用外部 API必须加上超时、重试和失败隔离超时每个请求设置timeout120。重试遇到临时错误尝试 2 次即可不要无限重试。失败隔离单条失败不影响整批结果把失败原因记到日志里。如果任务之间需要共享上下文比如先查用户再查订单建议把它们设计成一个 Agent 多步规划或者通过 LangGraph 维护同一个状态对象而不是拆成多个独立请求。8. 资源占用与性能观察LangChain MCP 这套组合的资源占用主要来自三个地方MCP Server 进程、LangChain Agent 执行过程、大模型推理或 API 调用。MCP Server 如果是本地 Python 子进程内存占用通常在几十 MB 到几百 MB 之间具体取决于你 import 的库和工具复杂度。如果你的 MCP Server 内部加载了重模型资源占用就会明显上升。这一点只能依赖实际运行来观察不建议照搬别人的数字。在代理执行过程中LangChain Agent 会做多次模型调用。以 ReAct Agent 为例模型可能要“思考 - 调用工具 - 观察结果 - 再思考”多轮每一轮都是一次 API 请求。这意味着相比单次普通对话Agent 场景的 token 消耗会翻倍甚至更多。性能优化的几个方向精简工具数量工具太多会让模型选择变慢甚至选错。精简工具描述长描述会增加 prompt 长度导致 token 上升。使用更快的小模型如果业务不复杂大模型负责规划小模型负责总结。缓存对于稳定数据查询可以在 MCP Server 层做结果缓存。并发限制API 服务商通常有 QPS 限制批量任务要控制并发数。另一个观察点是 Python 异步并发。标准 MCP Client 的stdio_client在多次调用时会有一些通信开销如果 Agent 要频繁调用工具建议把耗时工具放在 MCP Server 端做批量聚合而不是让模型多次小粒度调用。9. 常见问题与排查方法实战中LangChain MCP 最容易出问题的不是概念而是环境、协议和服务生命周期。下面这张表覆盖了高频故障问题现象可能原因排查方式解决方案MCP Server 连接失败子进程启动失败或 Python 命令找不到单独运行python mcp_server.py看是否有报错在配置中使用绝对路径 Python 路径工具未注册到 Agentload_mcp_tools返回空列表打印 tools 列表检查 MCP Server 是否有mcp.tool()确认 MCP Server 初始化成功且工具函数已定义模型选择错误工具工具描述不清晰或参数 Schema 不准确查看 Agent verbose 日志观察模型输出意图优化 docstring 和参数类型增加示例值Agent 无限循环模型反复调用工具但结果无法满足结束条件设置max_iterations参数用 AgentExecutor 时配置max_iterations5请求超时模型 API 响应慢或 MCP 工具耗时过长在工具内打印耗时日志增加 timeout或优化工具执行逻辑端口冲突使用 SSE 或 Streamable HTTP 传输时端口被占用lsof -i :8000或netstat -ano查看端口更换端口或关闭残留进程API Key 无效环境变量未生效打印os.getenv(OPENAI_API_KEY)检查配置文件和加载顺序批量任务卡住并发过高或 API 限流查看服务端错误日志降低并发数增加重试退避策略MCP 配置不生效mcp.json中 command 不是绝对路径检查配置格式路径不能含特殊字符使用绝对路径并确认可执行权限9.1 Windows 下创建 MCP Server 的注意点很多 Windows 用户会遇到command: python无法启动的问题。这通常是 Python 命令不在系统 PATH 中或者系统里同时存在多个 Python 环境。更稳妥的做法是在终端执行where python拿到绝对路径。在mcp.json的command中填写绝对路径例如C:\\Python310\\python.exe。确保虚拟环境中的包能被该 Python 解释器找到。如果你使用 Dify 或 Trae 等客户端添加本地 MCP 服务配置方式类似重点关注启动目录和 Python 解释器路径。9.2 MCP 工具注册后参数校验失败当模型调用 MCP 工具时参数必须符合 JSON Schema。比如get_recent_orders要求user_id是字符串如果模型传了数字 1001严格模式下会校验失败。解决方案是在工具函数内部做宽松转换或在 Schema 中标记参数类型为 string并在描述中写明“user_id 为字符串类型”。10. 最佳实践与合规建议工程化使用 LangChain MCP不能只停留在“能跑通”阶段还要考虑稳定性、可维护性和合规边界。10.1 保持 MCP Server 的独立性MCP Server 应该是纯粹的工具服务不依赖 LangChain。它只暴露协议不管上层是 LangChain 还是 Dify。这样你升级 LangChain 或换 Agent 框架时MCP Server 不需要改动。建议把 MCP Server 单独放在一个目录或独立模块里和 Agent 代码解耦。10.2 工具描述要面向模型写工具描述不是给人看的是给模型看的。写工具说明时要注意描述用途而不是实现细节。说明输入参数的格式和范围。给出典型示例减少模型猜测。提示边界条件比如“用户不存在时返回 error”。10.3 日志、追踪与异常隔离Agent 调用链路比较长建议每个阶段都输出结构化日志进入 Agent 的用户问题。模型选择了哪个工具。工具实际传入参数。工具返回结果摘要。最终回答。这样排查问题时能快速定位是模型选择错误还是工具返回异常还是最终生成出现问题。10.4 合规与授权如果你的 MCP 工具会访问数据库、客户信息、文件系统或第三方 API必须遵守最小权限原则。不要给 Agent 提供无限制的 SQL 执行工具尤其是生产环境。涉及用户隐私数据的查询要在 MCP Server 层做鉴权不能只靠模型自觉。对外部版权内容、人物肖像、商业秘密等敏感信息务必获得合法授权后再使用。10.5 版本锁定LangChain 生态迭代速度很快今天能跑通的代码几个月后可能因为 API 变更失效。建议在项目的requirements.txt或pyproject.toml中锁定关键依赖版本并保留一份最小可运行配置方便快速回滚。11. 总结与下一步LangChain MCP 组合的意义在于把“模型能力”和“系统工具”之间的连接从定制化变成标准化。LangChain 负责编排 Agent 的思考和行动MCP 负责让工具以统一协议被任意支持方调用。对于开发者来说先跑通一个最简单的 MCP Server再让 LangChain Agent 调用它是理解整套体系最快的路径。如果你正准备上手建议按这个顺序做先写一个不依赖任何外部服务的 MCP Server比如查内存字典然后让 LangChain Agent 调用它观察 verbose 日志理解模型的工具调用过程接着把 Memory 和历史消息加进去最后再封装 API 和批量任务。第一篇代码可以小但链路要完整。最容易踩的坑是跳过 MCP Server 单独调试直接在 LangChain 里加载工具等到出现问题时搞不清是协议层问题还是模型选择问题。所以第一步一定要先把 MCP Server 单独跑通再接入 Agent。下一步可以继续玩的方向包括用 LangGraph 替换 AgentExecutor让 Agent 支持更复杂的多步规划把 MCP Server 部署到远程通过 SSE 或 HTTP 连接把批量任务改造成 Celery 或任务队列在 MCP Server 中接入数据库、Figma、Playwright 等真实业务工具。这套流程跑熟之后后面接任何新工具都会很顺。