企业官网建设流程全解析

这次我们来看一个正在改变 Agent 开发方式的技术组合：LangChain Agent 集成 MCP。如果你最近在关注 AI Agent 开发，会发现两个词出现频率越来越高——Agent 框架和 MCP 协议。LangChain 是目前生态最完整的 Agent 编排框架，而 MCP（Model Context Protocol）则是让 Agent 能够稳定调用外部工具和数据源的标准协议。两者结合之后，Agent 的开发方式从“为每个工具写胶水代码”变成了“启动一个 MCP Server，Agent 自动发现工具并调用”，这个变化是企业级 Agent 落地过程中很值得关注的方向。

这篇文章不会只停留在概念讲解。我会从 LangChain Agent 的基础实现开始，然后进入 MCP 集成全流程，包括 MCP Server 的编写、MCP Client 的封装、如何在 LangChain 的 Tool Calling Agent 里直接使用 MCP 工具。最后重点讲企业级 Agent 记忆系统：会话记忆、长期记忆、向量检索、容量控制，并给出一套可以直接改造成生产代码的存储方案。看完之后，你可以顺手搭出一个具备外部工具能力和多轮记忆能力的 Agent 原型。

1. LangChain Agent 集成 MCP 核心能力速览

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 SDK，langchain-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( model="gpt-4o-mini", temperature=0, )

接着定义一个工具。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(agent=agent, tools=tools) 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 Tool，AgentExecutor 和 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 包装任意现有脚本。

传输层有两种：

• stdio：MCP Server 作为子进程启动，Agent 和 Server 在同一台机器上通过标准输入输出通信。优点是启动简单，不需要开放端口，本地开发首选。
• SSE/HTTP：MCP Server 作为独立 HTTP 服务启动，Agent 可以远程连接。适合部署在服务器或容器环境。

5.2 编写一个 MCP Server

下面是一个订单查询 MCP Server 的最小实现，使用官方mcpPython SDK：

from 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( name="search_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(type="text", text=f"订单 {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 Schema，LangChain 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( command="python", 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 tools

load_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 tools

6. 企业级 Agent 记忆系统设计

6.1 记忆分类与存储选型

企业级 Agent 记忆系统要解决三件事：记住用户说过什么、记住 Agent 自己的结论、在需要时快速找回。常见分类如下：

选型原则：会话记忆要求延迟低、顺序读，用 Redis 合适；长期记忆要求语义检索，用向量库；实体记忆如果业务结构固定，建议落数据库表，不要只存在向量库里。生产环境常见组合是 Redis 存短期会话 + PostgreSQL（pgvector）存长期记忆。

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 = f"chat_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(content=msg_data["content"])) elif msg_data["type"] == "ai": self.messages.append(AIMessage(content=msg_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(model="text-embedding-3-small") self.vector_store = Chroma( collection_name=collection_name, embedding_function=self.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}, k=top_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_model=ChatResponse) 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(content=req.message)) history.add_message(AIMessage(content=result["output"])) return ChatResponse(output=result["output"])

启动服务：

uvicorn api_server:app --host 0.0.0.0 --port 8000

7.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(f"Agent 执行耗时: {time.time() - start:.2f}s")

进程资源方面，stdio 模式的 MCP Server 是一个子进程，多个 Agent 并发时要注意子进程数量；远程 MCP 模式则要监控服务端 CPU 和内存。

9. 常见问题与排查方法

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 网关的权限设计，以及长会话场景下的摘要记忆压缩策略。建议收藏备用，动手搭建时可以直接对照这份流程走。