这次我们来看一个正在改变 Agent 开发方式的技术组合LangChain Agent 集成 MCP。如果你最近在关注 AI Agent 开发会发现两个词出现频率越来越高——Agent 框架和 MCP 协议。LangChain 是目前生态最完整的 Agent 编排框架而 MCPModel Context Protocol则是让 Agent 能够稳定调用外部工具和数据源的标准协议。两者结合之后Agent 的开发方式从“为每个工具写胶水代码”变成了“启动一个 MCP ServerAgent 自动发现工具并调用”这个变化是企业级 Agent 落地过程中很值得关注的方向。这篇文章不会只停留在概念讲解。我会从 LangChain Agent 的基础实现开始然后进入 MCP 集成全流程包括 MCP Server 的编写、MCP Client 的封装、如何在 LangChain 的 Tool Calling Agent 里直接使用 MCP 工具。最后重点讲企业级 Agent 记忆系统会话记忆、长期记忆、向量检索、容量控制并给出一套可以直接改造成生产代码的存储方案。看完之后你可以顺手搭出一个具备外部工具能力和多轮记忆能力的 Agent 原型。1. LangChain Agent 集成 MCP 核心能力速览能力项说明项目定位LangChain 是 AI 应用编排框架MCP 是模型上下文协议组合后用于构建可调用外部工具的 Agent技术路线LangChain Agent MCP Client MCP Server工具接入方式MCP Server 通过 JSON Schema 声明工具Agent 自动发现并调用无需手工实现每个工具的解析逻辑协议支持MCP 支持 stdio本地子进程和 SSE/HTTP远程服务两种传输方式Agent 模式Tool Calling Agent、ReAct Agent也可以切换到 LangGraph 做精细编排记忆能力短期会话记忆、长期向量记忆、实体记忆、摘要记忆可接入 Redis、PostgreSQL、向量数据库部署形式本地 Python 进程、Docker 容器、API 服务均可批量任务可封装为 API 服务后对大量输入做并发或队列化处理上手难度中高需要理解 Agent 循环、工具调用和协议封装三层逻辑适合人群正在做 AI Agent 应用、知识库问答、自动化办公、企业内部工具集成的开发者2. 适用场景与使用边界LangChain Agent 集成 MCP 最适合的场景有三类。第一类是内部工具聚合。企业中往往有订单系统、工单系统、IM 机器人、数据库、搜索服务这些系统原本各有各的 API。用 MCP 把这些工具包成统一协议后Agent 不需要针对每个系统写独立的 Function Call 逻辑只要 MCP Server 启动工具列表就自动挂进来了。第二类是知识库问答和业务 Agent。用户问“上个月华东区的销售额是多少”“这个订单为什么被拦截”Agent 需要先判断该调用哪个工具再根据工具返回结果继续回答。这类流程用 LangChain Agent 编排非常合适。第三类是自动化办公流程。比如 Agent 读取邮件附件、解析 PDF、生成摘要、写入表格这些操作都可以通过 MCP 工具暴露给 Agent。同时也要明确使用边界MCP 不是银弹。它解决的是“工具接入标准化”问题不解决“Agent 规划是否可靠”问题。规划能力仍然依赖底座模型本身。不要把所有敏感操作直接暴露给 Agent。写库、删库、发送消息、转账这类能力必须加权限控制、操作确认和审计日志。涉及用户隐私、客户数据、内部业务数据时必须先确认数据来源合法性做脱敏和访问控制。API Key、数据库连接串、内部服务地址等敏感配置不能写死在代码里更不能提交到公开仓库。3. 环境准备与前置条件开始动手之前先确认本机环境。操作系统建议 Windows 10/11、Ubuntu 20.04 或 macOS 12。需要安装 Python 3.10 或更高版本并准备好一个可用的虚拟环境。LLM 调用建议准备 OpenAI 兼容接口的 API Key或者本地部署的模型服务地址如果使用 Anthropic 模型需要对应的 API Key。依赖安装示例# 创建虚拟环境 python -m venv venv source venv/bin/activate # Windows 下执行 venv\Scripts\activate # 安装核心依赖 pip install langchain langchain-openai langchain-anthropic pip install mcp langchain-mcp-adapters # 记忆系统按需安装 pip install redis chromadb sqlalchemy其中mcp是 MCP 官方 Python SDKlangchain-mcp-adapters负责把 MCP 返回的工具列表转换成 LangChain Agent 可识别的工具格式。记忆系统部分如果只是本地原型sqlite加chromadb就够了生产环境建议接 Redis 和 PostgreSQL。还要准备一个目录结构agent-mcp-demo/ ├── .env # 环境变量不提交到仓库 ├── mcp_servers/ │ └── order_server.py # 订单相关的 MCP Server ├── agents/ │ └── agent_runner.py # Agent 启动入口 ├── memory/ │ └── chat_memory.py # 记忆系统存储实现 └── tests/ └── test_agent.py # 功能验证脚本4. LangChain Agent 基础实现4.1 接入 LLM 并定义第一个工具LangChain Agent 的核心是让模型在对话过程中决定“是否需要调用工具”“调用哪个工具”“工具参数是什么”。所以第一步是接入 LLM然后定义一个或多个工具。import os from langchain_openai import ChatOpenAI os.environ[OPENAI_API_KEY] your-api-key llm ChatOpenAI( modelgpt-4o-mini, temperature0, )接着定义一个工具。LangChain 的tool装饰器会读取函数名、docstring 和类型注解自动生成工具声明from langchain_core.tools import tool tool def get_order_status(order_id: str) - str: 根据订单号查询订单当前状态 # 这里替换为真实的订单系统接口 return f订单 {order_id} 当前状态已发货预计 3 天内到达4.2 创建 Tool Calling Agent工具定义好之后用create_tool_calling_agent创建 Agent再用AgentExecutor执行循环。执行循环内部负责将用户输入和工具列表交给模型模型返回工具调用请求执行工具把结果回填给模型直到模型输出最终答案。from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一个可以调用外部工具来回答问题的助手。), (human, {input}), (placeholder, {agent_scratchpad}), ]) tools [get_order_status] agent create_tool_calling_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools) result agent_executor.invoke({input: 查询订单 A10086 的状态}) print(result[output])这段代码已经是一个可运行的 Agent 原型。它的执行链路是用户输入进入 AgentExecutor模型判断需要查询订单输出 tool call执行工具获取结果再次交给模型模型生成最终回答。4.3 AgentExecutor 与 LangGraph 的选型LangChain 早期的 AgentExecutor 是“while 循环式”的执行器简单直接适合大多数常规工具调用场景。LangGraph 则是把 Agent 执行流程构建成一张图节点之间可以加条件判断、并发执行、人工审核适合流程复杂、需要精细控制的企业级场景。选择建议是普通 Demo、内部工具调用、快速验证用 AgentExecutor 足够如果要做多角色协同、人工审核节点、复杂条件路由、状态持久化建议上 LangGraph。MCP 工具接入本身不依赖你选哪个执行器langchain-mcp-adapters的产物是标准的 LangChain ToolAgentExecutor 和 LangGraph 都能用。5. MCP 集成全流程5.1 MCP 架构理解MCP 协议中有三个角色MCP Host承载 Agent 的宿主应用比如 LangChain 程序。MCP Client与 MCP Server 建立连接的客户端。MCP Server暴露工具、资源、提示词的服务端。消息传递有三种核心能力Tools工具调用、Resources资源读取、Prompts提示词模版。对 Agent 开发来说Tools 是最常用的。MCP Server 可以用官方 Python SDK、TypeScript SDK或者直接通过 CLI 包装任意现有脚本。传输层有两种stdioMCP Server 作为子进程启动Agent 和 Server 在同一台机器上通过标准输入输出通信。优点是启动简单不需要开放端口本地开发首选。SSE/HTTPMCP Server 作为独立 HTTP 服务启动Agent 可以远程连接。适合部署在服务器或容器环境。5.2 编写一个 MCP Server下面是一个订单查询 MCP Server 的最小实现使用官方mcpPython SDKfrom mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent app Server(order-server) app.list_tools() async def list_tools(): return [ Tool( namesearch_order, description根据订单号查询订单状态, inputSchema{ type: object, properties: { order_id: {type: string} }, required: [order_id] } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name search_order: order_id arguments[order_id] # 此处替换为真实查询逻辑 return [TextContent(typetext, textf订单 {order_id} 状态已发货)] raise ValueError(f未知工具: {name}) async def main(): async with stdio_server() as (read_stream, write_stream): await app.run(read_stream, write_stream, app.create_initialization_options()) if __name__ __main__: import asyncio asyncio.run(main())这个 Server 引入了list_tools和call_tool两个核心方法。前者声明工具列表后者根据工具名和参数执行具体动作。MCP 工具声明的核心是 JSON SchemaLangChain Agent 拿到这份 Schema 后会自动生成对应的工具描述。5.3 在 LangChain 中加载 MCP 工具启动 MCP Server 后LangChain 程序通过langchain_mcp_adapters加载工具import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools server_params StdioServerParameters( commandpython, args[mcp_servers/order_server.py], ) async def get_mcp_tools(): 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) return toolsload_mcp_tools会遍历 MCP Server 声明的所有工具转换成 LangChain 的BaseTool对象。这样一来LangChain Agent 无需关心工具内部实现直接按照标准 Tool 列表使用即可。注意stdio_client是非同步上下文管理器通常需要把 Agent 执行逻辑也放进同一个async with块里。如果要常驻运行建议把 MCP Server 作为独立服务启动通过 HTTP/SSE 传输避免子进程生命周期管理问题。5.4 本地 MCP 与远程 MCP 的选择本地 stdio 模式适合开发调试和单机部署优点是零网络开销进程内通信。缺点是 MCP Server 不能跨机器共享且每次 Agent 启动时都要拉起子进程进程管理要小心。远程 SSE/HTTP 模式适合生产环境。MCP Server 独立部署成一个服务多个 Agent 或多台机器可以共用一套工具服务。缺点是增加了网络延迟和鉴权复杂度。企业级建议工具服务统一封装成远程 MCP Server部署在 Docker 中前面加网关鉴权开发环境本地用 stdio 加速迭代。两种模式在 LangChain 侧切换成本很低因为load_mcp_tools接收的都是ClientSession只要把stdio_client换成sse_client即可。远程 MCP 示例from mcp import ClientSession from mcp.client.sse import sse_client async def get_remote_mcp_tools(): sse_url http://127.0.0.1:8080/mcp async with sse_client(sse_url) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await load_mcp_tools(session) return tools6. 企业级 Agent 记忆系统设计6.1 记忆分类与存储选型企业级 Agent 记忆系统要解决三件事记住用户说过什么、记住 Agent 自己的结论、在需要时快速找回。常见分类如下记忆类型存储内容典型存储会话记忆当前会话内的多轮消息Redis、SQLite、MySQL摘要记忆长对话的压缩摘要Redis、向量库实体记忆用户偏好、业务实体信息PostgreSQL、向量库长期语义记忆历史会话的语义检索结果Chroma、pgvector、Milvus选型原则会话记忆要求延迟低、顺序读用 Redis 合适长期记忆要求语义检索用向量库实体记忆如果业务结构固定建议落数据库表不要只存在向量库里。生产环境常见组合是 Redis 存短期会话 PostgreSQLpgvector存长期记忆。6.2 会话记忆实现多轮对话是 Agent 的基本要求。最简单的方式是把历史消息传回模型但上下文窗口有限。更合理的方式是只回填最近 N 轮或者对旧消息做摘要。使用 LangChain 的BaseChatMessageHistory做自定义 Redis 存储import json import redis from langchain_core.chat_history import BaseChatMessageHistory from langchain_core.messages import ( BaseMessage, HumanMessage, AIMessage, ) class RedisChatMessageHistory(BaseChatMessageHistory): def __init__(self, session_id: str, redis_url: str redis://localhost:6379/0): self.session_id session_id self.redis redis.from_url(redis_url) self.key fchat_history:{session_id} self._loaded False def _load_if_needed(self): if not self._loaded: self.messages [] for raw in self.redis.lrange(self.key, 0, -1): msg_data json.loads(raw) if msg_data[type] human: self.messages.append(HumanMessage(contentmsg_data[content])) elif msg_data[type] ai: self.messages.append(AIMessage(contentmsg_data[content])) self._loaded True def add_message(self, message: BaseMessage) - None: self._load_if_needed() self.messages.append(message) self.redis.rpush( self.key, json.dumps({ type: human if isinstance(message, HumanMessage) else ai, content: message.content, }) ) def clear(self) - None: self.redis.delete(self.key) self.messages [] self._loaded True使用时只需要在每次请求时实例化这个类把历史消息注入 Prompt 模板。6.3 长期记忆与向量检索长期记忆的核心思路历史会话数据写入向量库当新问题进来时先用语义检索召回相关旧对话再拼接到 Prompt 中。下面是基于 Chroma 的检索记忆模块from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma class LongTermMemory: def __init__(self, collection_name: str agent_memory): self.embeddings OpenAIEmbeddings(modeltext-embedding-3-small) self.vector_store Chroma( collection_namecollection_name, embedding_functionself.embeddings, persist_directory./memory_db, ) def save_memory(self, user_id: str, content: str): self.vector_store.add_texts( texts[content], metadatas[{user_id: user_id}], ) def search_memory(self, user_id: str, query: str, top_k: int 3): docs self.vector_store.similarity_search( query, filter{user_id: user_id}, ktop_k, ) return [doc.page_content for doc in docs]这段代码有一个很重要的细节filter{user_id: user_id}做了用户隔离。企业级记忆系统必须支持多用户隔离否则用户 A 的行为会被用户 B 的会话检索出来这是安全事故。6.4 记忆容量与隐私控制长期运行后记忆量会持续增长必须设计容量策略每条记忆打上时间戳写入时只保留最近 30 天或最近 500 条记录。检索时用 Top-K 限制召回数量防止 Prompt 过长。关键业务场景要做人工复核Agent 写入记忆前先过滤敏感信息。用户提出删除数据时必须提供清空接口这是合规底线。7. 接口 API 与批量任务7.1 将 Agent 封装成 API 服务Agent 原型跑通后要对外提供服务最简单的方式是封装成一个 FastAPI 接口。接口接收用户输入和 session_id内部完成记忆读取、Agent 执行、记忆回写三个步骤。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): session_id: str message: str class ChatResponse(BaseModel): output: str app.post(/chat, response_modelChatResponse) async def chat(req: ChatRequest): history RedisChatMessageHistory(req.session_id) result agent_executor.invoke({ input: req.message, chat_history: history.messages[-10:], }) history.add_message(HumanMessage(contentreq.message)) history.add_message(AIMessage(contentresult[output])) return ChatResponse(outputresult[output])启动服务uvicorn api_server:app --host 0.0.0.0 --port 80007.2 批量任务与重试如果是批量处理场景不建议直接用同步接口压并发。更稳妥的做法是维护一个任务队列用asyncio或 Celery 处理每条任务记录状态和重试次数。import asyncio async def process_batch(task_list: list[dict]): results [] semaphore asyncio.Semaphore(5) async def run_with_limit(task): async with semaphore: for attempt in range(3): try: result agent_executor.invoke({ input: task[message], chat_history: [], }) return {task_id: task[id], status: ok, output: result[output]} except Exception as e: if attempt 2: return {task_id: task[id], status: failed, error: str(e)} await asyncio.sleep(2) tasks [run_with_limit(task) for task in task_list] results await asyncio.gather(*tasks) return results批量任务要特别关注一点MCP Server 的并发能力。如果底层工具服务有并发上限Agent 侧并发再高也没有意义反而会导致大量超时。建议批量场景先压测 MCP Server 的 QPS再决定 Agent 侧的并发数。8. 资源占用与性能观察Agent 服务和传统 Web 服务不同资源消耗主要在模型推理和上下文 Token 上指标集中在以下几处。Context Token 消耗是最大的变量。LangChain Agent 每轮循环都会把系统 Prompt、对话历史、工具描述拼进上下文。工具越多、工具描述越长每次请求的 Token 就越多。MCP 场景下尤其明显因为每个 MCP 工具都带 JSON Schema。可以统计prompt_tokens和completion_tokens可以在调用 OpenAI 接口时从返回的usage字段读取result llm.invoke(prompt) print(result.usage_metadata) # 查看 token 消耗延迟观察。一次 Agent 执行可能包含多轮模型调用和工具调用。工具调用越多链路越长。MCP 工具如果是本地 stdio 进程还要加上子进程启动和序列化的开销。建议在代码里记录每次调用的耗时import time start time.time() result agent_executor.invoke({input: 查询订单 A10086 状态}) print(fAgent 执行耗时: {time.time() - start:.2f}s)进程资源方面stdio 模式的 MCP Server 是一个子进程多个 Agent 并发时要注意子进程数量远程 MCP 模式则要监控服务端 CPU 和内存。9. 常见问题与排查方法问题现象可能原因排查方式解决方案Agent 报错 Tool not foundMCP Server 工具列表未正确加载打印load_mcp_tools返回的工具列表检查 MCP Server 的list_tools声明是否正确模型输出了空工具参数LLM 不支持 Tool Calling或提示词中缺少工具说明换用支持 Function Calling 的模型检查工具描述是否清晰使用gpt-4o-mini、claude-3-5-sonnet等支持 Tool Calling 的模型调用 MCP 工具一直超时stdio 子进程卡住或远程 MCP 服务不可达手动运行 MCP Server 脚本看能否正常返回检查网络端口增加超时设置或切换到远程 SSE 模式多轮对话后回答质量下降历史消息全部拼进上下文Token 过多检查 prompt_tokens查看历史消息长度限制回填历史轮数使用摘要压缩Redis 会话记忆不生效session_id 不一致检查每次请求传入的 session_id前端显式传递 session_id服务端做校验批量任务偶发失败并发触发 MCP Server 资源争抢查看服务端日志、进程数、连接数降低并发数增加重试机制必要时给 MCP Server 做连接池记忆检索到其他用户数据向量检索 filter 缺失或失效检查查询代码中的 filter 参数所有检索必须强制带 user_id 过滤最好在代码层面做封装10. 最佳实践与使用建议先小场景验证再上企业级架构。第一次跑通不需要接 Redis也不需要接向量库。先让 Agent 调用一个本地 MCP 工具确认链路通再逐步加入记忆系统和批量任务。工具权限要收敛。MCP Server 不要一股脑把所有内部系统接口全部暴露。按最小权限原则每个工具只暴露必要参数服务端做参数校验和操作白名单。涉及写操作的工具Agent 调用前最好有人工确认节点这个用 LangGraph 很容易实现。记忆系统先明确数据边界。session_id、user_id 是记忆系统的两条核心主键所有读写都必须带上。写入前过滤手机号、身份证、密钥等敏感字段。任何用户数据删除请求都要能通过接口完成。日志和追踪必须从一开始就加上。Agent 的执行链路通常有多个 tool call每步都应当记录输入、调用的工具、工具返回、模型输出、耗时、Token 消耗。否则生产环境出现问题很难复盘。MCP 工具声明要尽量具体。工具名称和 description 会直接影响模型的选择判断。描述模糊会导致模型选了工具但传错参数。例如search_order比query_data更清晰根据订单号查询订单当前状态比查询信息更好。11. 总结与下一步LangChain Agent 集成 MCP 的技术链路已经很成熟MCP Server 负责工具暴露LangChain Agent 负责规划与调用记忆系统负责多轮与长期信息保持。对团队来说最大的收益是工具接入成本大幅下降——新系统只要写一个 MCP Server 就能被 Agent 复用不需要为每个 Agent 单独适配。建议先从最小闭环开始写一个 MCP Server用load_mcp_tools加载工具再跑通一个 AgentExecutor 示例。然后做两件立刻能见到效果的事一是接入 Redis 会话记忆让 Agent 能记住上下文二是把记忆和工具调用过程加上日志方便后续排查。最容易踩的坑有三个MCP 工具描述不清晰导致模型调用失败、会话记忆 session_id 不一致导致上下文丢失、批量并发时 MCP Server 被打满。这三类问题在架构设计阶段就要考虑进去。如果你已经在使用 LangChain 但还没接 MCP建议尽快迁移。MCP 工具可以被 LangGraph、各种 Agent 框架复用标准化程度比手写 Function Call 高很多。接下来值得继续深入的方向是 LangGraph 的图编排、MCP 网关的权限设计以及长会话场景下的摘要记忆压缩策略。建议收藏备用动手搭建时可以直接对照这份流程走。