最近在开发基于大模型的智能助手时我遇到了一个核心瓶颈Agent 的能力被其内置的工具集严重限制。无论是文件操作、数据库查询还是调用第三方 API都需要开发者预先编写大量适配代码这不仅开发效率低下也使得 Agent 难以灵活应对复杂多变的真实业务场景。直到深入研究了 LangChain Agent 与 MCPModel Context Protocol及 Skills 的集成方案才真正找到了让 Agent 能力实现“跃升”的钥匙。本文将系统拆解 LangChain Agent 如何通过 MCP 协议接入海量 Skills从而获得动态扩展的工具能力。我们将从核心概念入手逐步深入到环境搭建、原理剖析、实战编码并最终给出生产级的最佳实践。无论你是刚接触 LangChain 的新手还是希望优化现有 AI 应用架构的资深开发者都能从本文中获得一套完整、可落地的技术方案。1. 背景与核心概念为什么需要 MCP 和 Skills在传统的 LangChain Agent 开发模式中我们通常通过Tool类来定义 Agent 可以使用的工具。例如一个搜索工具、一个计算器工具。这些工具需要被硬编码到 Agent 的初始化参数中。这种方式存在几个明显问题工具固化Agent 一旦创建其工具集就固定了。要新增工具必须修改代码并重新部署。开发繁琐每个工具都需要手动实现其_run方法处理输入输出并与大模型进行适配。能力孤岛不同项目、不同团队开发的工具难以共享和复用形成“能力孤岛”。MCPModel Context Protocol正是为了解决这些问题而诞生的一个开放协议。你可以把它想象成 AI 模型的“USB 接口”标准。它定义了一套标准化的方式让任何服务器MCP Server都可以向客户端如 LangChain、Claude Desktop 等声明自己提供哪些“能力”即 Tools 或 Skills以及如何调用这些能力。Skills在 MCP 的语境下可以理解为通过 MCP Server 暴露出来的、可供 AI 模型调用的具体功能单元。一个 Skill 就是一个工具例如“读取文件”、“执行 SQL 查询”、“调用天气 API”。核心价值通过 MCPLangChain Agent 不再需要预先知道所有工具的具体实现。它可以在运行时动态发现并连接到一个或多个 MCP Server从而即时获取并利用这些 Server 提供的所有 Skills。这实现了工具能力的“即插即用”和无限扩展。概念区分LangChain ToolLangChain 框架内定义的、用于构建 Agent 的基本能力单元。MCP Server一个独立的进程或服务遵循 MCP 协议对外提供一组 Skills。SkillMCP Server 提供的一个具体功能在 LangChain 中会被适配成一个Tool实例。MCP Client遵循 MCP 协议、能够连接和调用 MCP Server 的客户端。LangChain 通过MCPClient来扮演这个角色。2. 环境准备与版本说明在开始实战之前我们需要搭建好开发环境。本文将使用 Python 作为主要开发语言。基础环境要求操作系统macOS / Linux / Windows (WSL2 推荐)Python 版本 3.10包管理工具pip 或 poetry核心依赖库及版本说明我们将创建一个新的项目目录并通过requirements.txt管理依赖。关键库的作用如下# requirements.txt langchain0.1.0 # LangChain 核心库版本建议 0.1 langchain-community0.0.10 # 包含社区贡献的集成如 MCP 支持 langchain-core0.1.0 # LangChain 核心抽象 mcp0.1.0 # MCP 协议的 Python SDK用于创建和运行 MCP Server anthropic0.18.0 # 用于调用 Claude 模型也可替换为 openai 等 python-dotenv1.0.0 # 用于管理环境变量如 API Key版本兼容性提示MCP 和 LangChain 对该集成的支持在快速迭代中。本文示例基于langchain-community0.0.10 和mcp0.1.0 编写。如果遇到导入错误请查阅官方文档确认最新接口。安装命令# 创建并进入项目目录 mkdir langchain-mcp-agent cd langchain-mcp-agent # 创建虚拟环境可选但推荐 python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 安装依赖 pip install -r requirements.txt项目结构预览langchain-mcp-agent/ ├── requirements.txt ├── .env # 存储 API KEY 等敏感信息 ├── simple_mcp_server.py # 自定义的简单 MCP Server ├── file_operations_server.py # 提供文件操作的 MCP Server ├── agent_with_mcp.py # 集成 MCP 的 LangChain Agent 主程序 └── README.md3. 核心原理拆解LangChain 如何与 MCP 协同工作理解其工作原理是灵活应用和排查问题的基础。整个流程可以分解为以下几个关键步骤3.1 MCP 通信协议基础MCP 本质上是一个基于 JSON-RPC 的通信协议。它运行在stdio标准输入输出或sse服务器发送事件等传输层之上。MCP Server 启动后会等待 Client 连接。连接建立后双方会进行初始化握手。关键交互Client → Server发送initialize请求。Server → Client回复initialized通知并在initialize响应中列出自己提供的Tools(即 Skills) 列表。每个 Tool 的定义包括名称、描述、输入参数 schema。Client → Server当需要执行某个 Skill 时发送tools/call请求。Server → Client执行对应的业务逻辑然后通过tools/call响应返回结果。3.2 LangChain 的集成层MCPClient与MCPToolLangChain 通过langchain_community.tools.mcp模块提供了 MCP 的集成。MCPClient封装了与 MCP Server 的通信细节。它负责连接 Server、获取工具列表、并将工具调用请求发送给 Server。MCPTool一个适配器类继承自 LangChain 的BaseTool。每个从 MCP Server 获取到的 Skill 都会被包装成一个MCPTool实例。这样LangChain Agent 就可以像使用普通 Tool 一样使用这些远程 Skill。3.3 动态工具发现与加载这是最核心的“能力跃升”点。流程如下启动 Server一个或多个 MCP Server 进程被启动。创建 Client在 LangChain 应用中创建MCPClient实例并配置其连接到目标 Server。获取工具调用client.get_tools()方法。该方法内部会与 Server 通信获取其声明的所有 Skills 的元数据。工具适配MCPClient根据元数据自动创建对应的MCPTool列表。注入 Agent将这些MCPTool添加到 LangChain Agent 的tools参数中。动态调用Agent 在推理过程中如果决定使用某个工具会调用对应的MCPTool._run()该方法会将请求转发给MCPClient再由Client通过 MCP 协议调用远程 Server 上的实际实现。优势你可以在不重启 Agent 应用的情况下通过启动新的 MCP Server 来为 Agent 增加新的能力。Agent 只需重新连接或刷新工具列表即可。4. 完整实战案例构建一个具备文件操作和网络搜索能力的 Agent接下来我们将通过一个完整的例子演示如何创建自定义 MCP Server并将其提供的 Skills 集成到 LangChain Agent 中。我们的目标是让 Agent 能够回答“请总结我桌面上的notes.txt文件内容并搜索最新的 LangChain 新闻”。4.1 创建自定义 MCP Server文件操作首先我们创建一个提供基础文件读取和写入技能的 MCP Server。# file: file_operations_server.py import json from typing import Any, List from mcp import Server, NotificationOptions import mcp.types as types from pathlib import Path class FileOperationsServer(Server): 一个提供基础文件操作的 MCP Server def __init__(self): super().__init__(file-operations-server) # 注册此 Server 提供的工具Skills self.register_tool( nameread_file, description读取指定路径的文本文件内容, input_schema{ type: object, properties: { file_path: { type: string, description: 要读取的文件的绝对路径 } }, required: [file_path] }, callbackself._read_file ) self.register_tool( namewrite_file, description向指定路径的文本文件写入内容, input_schema{ type: object, properties: { file_path: { type: string, description: 要写入的文件的绝对路径 }, content: { type: string, description: 要写入的文本内容 } }, required: [file_path, content] }, callbackself._write_file ) async def _read_file(self, arguments: dict) - str: 读取文件的工具实现 file_path Path(arguments[file_path]) if not file_path.exists(): return f错误文件 {file_path} 不存在。 if not file_path.is_file(): return f错误{file_path} 不是一个文件。 try: content file_path.read_text(encodingutf-8) return f文件 {file_path} 的内容如下\n\n{content}\n except Exception as e: return f读取文件时发生错误{e} async def _write_file(self, arguments: dict) - str: 写入文件的工具实现 file_path Path(arguments[file_path]) content arguments[content] try: # 确保目录存在 file_path.parent.mkdir(parentsTrue, exist_okTrue) file_path.write_text(content, encodingutf-8) return f成功将内容写入文件 {file_path}。 except Exception as e: return f写入文件时发生错误{e} async def list_tools(self) - List[types.Tool]: 列出所有可用工具MCP协议要求 return list(self.tools.values()) if __name__ __main__: import asyncio # 使用 stdio 传输层运行 Server这是与 Client 通信的标准方式 server FileOperationsServer() asyncio.run(server.run(transportstdio))这个 Server 定义了两个 Skillread_file和write_file。它通过stdio运行意味着它将通过标准输入/输出与客户端通信这是与 LangChain 集成最常见的方式。4.2 配置 LangChain Agent 并连接 MCP Server现在我们创建主程序启动上述 Server并让 LangChain Agent 使用其 Skills。# file: agent_with_mcp.py import asyncio import subprocess import sys from pathlib import Path from dotenv import load_dotenv from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_community.tools.mcp import MCPClient, create_mcp_tool from langchain_community.chat_models import ChatAnthropic # 使用 Claude # 也可以使用 OpenAI: from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate # 1. 加载环境变量用于存放 API Key load_dotenv() async def main(): # 2. 启动 MCP Server 进程 # 这里我们启动刚才编写的文件操作 Server server_script_path Path(__file__).parent / file_operations_server.py server_process subprocess.Popen( [sys.executable, str(server_script_path)], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, bufsize1 ) # 3. 创建 MCP Client 并连接到 Server # 注意我们使用 server_process 的 stdin/stdout 来通信 mcp_client MCPClient( server_process.stdin, server_process.stdout, # 可以配置多个 Server 的传输 transportstdio ) # 4. 初始化 Client 连接并获取远程工具列表 await mcp_client.initialize() # 获取 Server 提供的所有 Skills并自动转换为 LangChain Tool mcp_tools await mcp_client.get_tools() print(f从 MCP Server 获取到 {len(mcp_tools)} 个工具: {[t.name for t in mcp_tools]}) # 5. 创建 LangChain 模型这里以 Claude 为例 # 请确保你的 .env 文件中有 ANTHROPIC_API_KEY llm ChatAnthropic(modelclaude-3-sonnet-20240229, temperature0) # 如果使用 OpenAI则替换为 # from langchain_openai import ChatOpenAI # llm ChatOpenAI(modelgpt-4-turbo-preview) # 6. 定义 Agent 的提示词模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个强大的助手可以调用工具来完成用户的请求。请根据用户的问题决定是否需要使用工具以及使用哪个工具。在回复时对工具返回的结果进行总结和提炼。), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}), ]) # 7. 创建 Agent # 将 MCP Tools 和任何本地工具组合在一起 all_tools mcp_tools # 这里我们只使用 MCP Tools agent create_tool_calling_agent(llmllm, toolsall_tools, promptprompt) # 8. 创建 Agent 执行器 agent_executor AgentExecutor(agentagent, toolsall_tools, verboseTrue, handle_parsing_errorsTrue) # 9. 测试 Agent # 示例 1读取文件 print(\n--- 测试读取文件 ---) test_file_path Path.home() / Desktop / notes.txt # 为了测试我们先创建一个示例文件如果不存在 if not test_file_path.exists(): test_file_path.write_text(这是桌面上的笔记文件。\n记录了一些关于 MCP 和 LangChain 的学习心得。\n明天需要完成项目原型。) result1 await agent_executor.ainvoke({ input: f请帮我读取并总结这个文件的内容{test_file_path} }) print(结果:, result1[output]) # 示例 2写入文件 print(\n--- 测试写入文件 ---) result2 await agent_executor.ainvoke({ input: f请帮我在桌面创建一个名为 todo.txt 的文件内容写上购买 groceries和完成 LangChain 文章。 }) print(结果:, result2[output]) # 10. 清理停止 Server 进程 server_process.terminate() await server_process.wait() if __name__ __main__: asyncio.run(main())4.3 运行与验证设置 API Key在项目根目录创建.env文件填入你的 Anthropic 或 OpenAI API Key。# .env ANTHROPIC_API_KEYyour_anthropic_api_key_here # 或 # OPENAI_API_KEYyour_openai_api_key_here运行程序python agent_with_mcp.py预期输出程序会启动 MCP Server连接并获取工具列表然后 Agent 会开始执行。在verboseTrue模式下你将看到详细的 Agent 思考过程ReAct 模式包括它决定调用哪个工具、传递什么参数、以及工具返回的结果。最终Agent 会输出对文件内容的总结或操作成功的确认信息。4.4 集成更多 Skills添加网络搜索能力仅仅文件操作还不够。我们可以轻松集成另一个提供网络搜索技能的 MCP Server。实际上社区已经有很多现成的 MCP Server。这里我们以连接一个“模拟”的搜索 Server 为例展示如何连接多个 Server。假设我们有一个运行在本地 8000 端口的搜索 MCP Server例如使用mcp serve命令启动的brave-searchserver。我们需要修改主程序创建第二个MCPClient来连接这个搜索 Server。# 在 agent_with_mcp.py 的 main 函数中修改工具集合部分 async def main(): # ... [启动 file_operations_server 的代码不变] ... # 创建第一个 Client文件操作 mcp_client_files MCPClient(server_process.stdin, server_process.stdout, transportstdio) await mcp_client_files.initialize() file_tools await mcp_client_files.get_tools() # 创建第二个 Client网络搜索- 假设搜索 Server 运行在 SSE 模式 import aiohttp from langchain_community.tools.mcp import MCPClient # 注意这里需要根据实际搜索 Server 的传输方式调整 # 例如如果搜索 Server 通过 HTTP SSE 暴露可能需要不同的连接方式 # 以下为概念性代码具体取决于 Server 实现 # async with aiohttp.ClientSession() as session: # mcp_client_search MCPClient(session, http://localhost:8000/sse, transportsse) # await mcp_client_search.initialize() # search_tools await mcp_client_search.get_tools() # 为了简化演示我们这里添加一个 LangChain 内置的 DuckDuckGo 搜索工具作为替代 from langchain_community.tools import DuckDuckGoSearchRun search_tool DuckDuckGoSearchRun() # 组合所有工具 all_tools file_tools [search_tool] print(f可用工具总数: {len(all_tools)}) # ... [后续创建 Agent 和执行的代码不变] ... # 测试组合能力 print(\n--- 测试组合能力读文件搜索---) result3 await agent_executor.ainvoke({ input: f先帮我读取 {test_file_path} 文件看看里面提到了什么技术。然后去网上搜索一下关于这项技术的最新动态。 }) print(结果:, result3[output])通过这种方式你的 Agent 就同时具备了文件操作和网络搜索的能力。你可以继续集成数据库查询、代码执行、图像处理等任何通过 MCP Server 暴露的 Skill。5. 常见问题与排查思路在集成 MCP 与 LangChain 的过程中你可能会遇到以下典型问题问题现象可能原因排查步骤与解决方案导入错误No module named langchain_community.tools.mcp1.langchain-community版本过低。2. 包未正确安装。1. 升级langchain-community:pip install -U langchain-community。2. 确认安装的版本是否包含 mcp 模块0.0.10 通常包含。运行时报错MCPClient初始化失败或通信错误1. MCP Server 进程未启动或已崩溃。2.stdio管道通信异常。3. MCP 协议版本不兼容。1. 检查server_process是否成功启动查看其stderr输出。2. 确保MCPClient使用的是正确的stdin和stdout对象。3. 尝试使用transportstdio并确保 Server 也以stdio模式运行。Agent 无法识别或错误调用 MCP Tool1. 工具描述不清晰导致 LLM 无法理解其用途。2. 输入参数 schema 定义有误与 LLM 预期不符。3. 工具返回结果格式不符合 Agent 预期。1. 在 MCP Server 中为每个 Tool 提供清晰、具体的description。2. 仔细检查input_schema确保properties和required字段定义正确。3. 确保 Tool 回调函数返回的是字符串且内容易于 LLM 理解。性能问题工具调用缓慢1. MCP Server 处理耗时。2. 网络延迟远程 Server。3. Agent 频繁调用复杂工具。1. 优化 MCP Server 内部逻辑。2. 对于远程 Server考虑使用更快的网络或将其部署在靠近 Agent 的位置。3. 在 Agent 提示词中引导其“思考”减少不必要的工具调用。无法连接到社区 MCP Server1. Server 地址或端口错误。2. Server 未运行或需要认证。3. 传输协议 (stdio/sse/http) 不匹配。1. 使用mcp ls命令如果安装了 mcp cli查看可用 Server。2. 参考该 Server 的文档确认启动命令和连接方式。3. 对于sse传输确保使用正确的 URL 端点如http://host:port/sse。通用排查流程隔离测试首先单独运行你的 MCP Server使用mcp inspect或编写一个简单的测试客户端来验证其是否能正常响应list_tools和call_tool。检查日志启用 LangChain Agent 的verboseTrue模式观察 Agent 的思考链和工具调用请求。简化问题从一个最简单的 “echo” MCP Server 和一个最简单的 Agent 开始逐步增加复杂度。6. 最佳实践与工程建议将 MCP 用于生产环境时需要考虑以下几点6.1 MCP Server 设计原则单一职责一个 MCP Server 应专注于一类功能如文件操作、数据库访问、特定 API 封装。这便于维护和复用。清晰的接口Tool 的name和description必须精准。好的描述能让 LLM 更准确地判断何时调用它。输入参数的description也同样重要。健壮的错误处理在 Tool 的回调函数中必须进行充分的错误捕获和验证如文件是否存在、参数是否合法并返回对人和LLM 都有意义的错误信息。资源管理妥善管理数据库连接、网络会话等资源考虑使用连接池或上下文管理器。6.2 LangChain Agent 集成优化工具筛选与分类不是所有 MCP Tool 都需要一次性加载给 Agent。可以根据会话上下文或用户权限动态加载不同的工具集。MCPClient.get_tools()返回的是列表你可以进行过滤。提示词工程在 Agent 的 System Prompt 中简要介绍可用的工具类别和能力引导 LLM 更有效地利用它们。例如“你可以使用文件操作工具来读写用户指定的文件使用搜索工具来获取最新信息。”超时与重试对 MCP 工具调用配置合理的超时和重试机制避免因单个 Server 无响应导致整个 Agent 卡住。异步并发利用asyncio并发启动多个 MCP Server 或并发调用多个工具提升整体响应速度。6.3 安全与权限这是重中之重MCP 极大地扩展了 Agent 的能力也意味着更大的风险。沙箱化对执行代码、访问文件系统、执行系统命令的 MCP Server必须运行在严格的沙箱环境中如 Docker 容器限制其权限和资源访问。输入验证与净化所有从 LLM 传递给 MCP Tool 的参数都必须视为不可信的需要在 Server 端进行严格的验证和净化防止路径遍历、命令注入等攻击。访问控制实现基于用户或会话的工具访问控制。不是每个用户都能使用所有 Skills。审计日志记录所有 MCP 工具的调用详情包括调用者、参数、结果、时间戳用于安全审计和问题追踪。6.4 生产环境部署进程管理使用systemd,supervisord或容器编排工具如 Kubernetes来管理 MCP Server 进程的生命周期确保其高可用。健康检查为 MCP Server 实现健康检查端点方便监控系统探测其状态。配置化将 MCP Server 的连接信息地址、传输方式、认证信息抽取到配置文件或环境变量中便于不同环境开发、测试、生产的切换。7. 总结与扩展方向通过本文的讲解和实战你应该已经掌握了 LangChain Agent 与 MCP 协议集成的核心技术。这套架构的核心优势在于“解耦”和“动态扩展”。AI Agent 的核心逻辑规划、推理、记忆与具体的能力实现工具被分离开后者可以通过标准的 MCP 协议以“插件”形式动态接入。下一步可以探索的方向探索丰富的社区 MCP Server社区已经涌现了大量优秀的 MCP Server例如brave-search: 提供网络搜索。github: 访问 GitHub API。sql: 执行 SQL 查询。filesystem: 更强大的文件系统操作。 使用mcp ls或查阅https://github.com/modelcontextprotocol/servers来发现更多。开发自定义业务 Skill将企业内部系统的 API如 CRM、ERP、数据库封装成 MCP Server让 Agent 能够安全、可控地操作业务数据。与 Claude Desktop/Code 集成MCP 协议最初由 Anthropic 推动Claude Desktop 和 Claude Code 编辑器原生支持 MCP。这意味着你开发的 MCP Server 可以直接被这些客户端使用极大地扩展了 Claude 的能力边界。结合 LangGraph 实现复杂工作流LangChain 的 LangGraph 库支持构建有状态的、多环节的 Agent 工作流。你可以将 MCP Tools 嵌入到 LangGraph 的节点中构建出能够处理复杂、多步骤任务的超级助手。技术的最终目标是服务于业务。当你需要为 AI Agent 添加一个新能力时不再需要修改核心 Agent 代码只需启动或连接一个提供该能力的 MCP Server。这种架构为构建真正强大、灵活且易于维护的 AI 应用打开了新的大门。现在就从将一个现有的脚本或 API 改造成 MCP Server 开始你的实践吧。
LangChain Agent通过MCP协议动态扩展工具能力实战指南
最近在开发基于大模型的智能助手时我遇到了一个核心瓶颈Agent 的能力被其内置的工具集严重限制。无论是文件操作、数据库查询还是调用第三方 API都需要开发者预先编写大量适配代码这不仅开发效率低下也使得 Agent 难以灵活应对复杂多变的真实业务场景。直到深入研究了 LangChain Agent 与 MCPModel Context Protocol及 Skills 的集成方案才真正找到了让 Agent 能力实现“跃升”的钥匙。本文将系统拆解 LangChain Agent 如何通过 MCP 协议接入海量 Skills从而获得动态扩展的工具能力。我们将从核心概念入手逐步深入到环境搭建、原理剖析、实战编码并最终给出生产级的最佳实践。无论你是刚接触 LangChain 的新手还是希望优化现有 AI 应用架构的资深开发者都能从本文中获得一套完整、可落地的技术方案。1. 背景与核心概念为什么需要 MCP 和 Skills在传统的 LangChain Agent 开发模式中我们通常通过Tool类来定义 Agent 可以使用的工具。例如一个搜索工具、一个计算器工具。这些工具需要被硬编码到 Agent 的初始化参数中。这种方式存在几个明显问题工具固化Agent 一旦创建其工具集就固定了。要新增工具必须修改代码并重新部署。开发繁琐每个工具都需要手动实现其_run方法处理输入输出并与大模型进行适配。能力孤岛不同项目、不同团队开发的工具难以共享和复用形成“能力孤岛”。MCPModel Context Protocol正是为了解决这些问题而诞生的一个开放协议。你可以把它想象成 AI 模型的“USB 接口”标准。它定义了一套标准化的方式让任何服务器MCP Server都可以向客户端如 LangChain、Claude Desktop 等声明自己提供哪些“能力”即 Tools 或 Skills以及如何调用这些能力。Skills在 MCP 的语境下可以理解为通过 MCP Server 暴露出来的、可供 AI 模型调用的具体功能单元。一个 Skill 就是一个工具例如“读取文件”、“执行 SQL 查询”、“调用天气 API”。核心价值通过 MCPLangChain Agent 不再需要预先知道所有工具的具体实现。它可以在运行时动态发现并连接到一个或多个 MCP Server从而即时获取并利用这些 Server 提供的所有 Skills。这实现了工具能力的“即插即用”和无限扩展。概念区分LangChain ToolLangChain 框架内定义的、用于构建 Agent 的基本能力单元。MCP Server一个独立的进程或服务遵循 MCP 协议对外提供一组 Skills。SkillMCP Server 提供的一个具体功能在 LangChain 中会被适配成一个Tool实例。MCP Client遵循 MCP 协议、能够连接和调用 MCP Server 的客户端。LangChain 通过MCPClient来扮演这个角色。2. 环境准备与版本说明在开始实战之前我们需要搭建好开发环境。本文将使用 Python 作为主要开发语言。基础环境要求操作系统macOS / Linux / Windows (WSL2 推荐)Python 版本 3.10包管理工具pip 或 poetry核心依赖库及版本说明我们将创建一个新的项目目录并通过requirements.txt管理依赖。关键库的作用如下# requirements.txt langchain0.1.0 # LangChain 核心库版本建议 0.1 langchain-community0.0.10 # 包含社区贡献的集成如 MCP 支持 langchain-core0.1.0 # LangChain 核心抽象 mcp0.1.0 # MCP 协议的 Python SDK用于创建和运行 MCP Server anthropic0.18.0 # 用于调用 Claude 模型也可替换为 openai 等 python-dotenv1.0.0 # 用于管理环境变量如 API Key版本兼容性提示MCP 和 LangChain 对该集成的支持在快速迭代中。本文示例基于langchain-community0.0.10 和mcp0.1.0 编写。如果遇到导入错误请查阅官方文档确认最新接口。安装命令# 创建并进入项目目录 mkdir langchain-mcp-agent cd langchain-mcp-agent # 创建虚拟环境可选但推荐 python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 安装依赖 pip install -r requirements.txt项目结构预览langchain-mcp-agent/ ├── requirements.txt ├── .env # 存储 API KEY 等敏感信息 ├── simple_mcp_server.py # 自定义的简单 MCP Server ├── file_operations_server.py # 提供文件操作的 MCP Server ├── agent_with_mcp.py # 集成 MCP 的 LangChain Agent 主程序 └── README.md3. 核心原理拆解LangChain 如何与 MCP 协同工作理解其工作原理是灵活应用和排查问题的基础。整个流程可以分解为以下几个关键步骤3.1 MCP 通信协议基础MCP 本质上是一个基于 JSON-RPC 的通信协议。它运行在stdio标准输入输出或sse服务器发送事件等传输层之上。MCP Server 启动后会等待 Client 连接。连接建立后双方会进行初始化握手。关键交互Client → Server发送initialize请求。Server → Client回复initialized通知并在initialize响应中列出自己提供的Tools(即 Skills) 列表。每个 Tool 的定义包括名称、描述、输入参数 schema。Client → Server当需要执行某个 Skill 时发送tools/call请求。Server → Client执行对应的业务逻辑然后通过tools/call响应返回结果。3.2 LangChain 的集成层MCPClient与MCPToolLangChain 通过langchain_community.tools.mcp模块提供了 MCP 的集成。MCPClient封装了与 MCP Server 的通信细节。它负责连接 Server、获取工具列表、并将工具调用请求发送给 Server。MCPTool一个适配器类继承自 LangChain 的BaseTool。每个从 MCP Server 获取到的 Skill 都会被包装成一个MCPTool实例。这样LangChain Agent 就可以像使用普通 Tool 一样使用这些远程 Skill。3.3 动态工具发现与加载这是最核心的“能力跃升”点。流程如下启动 Server一个或多个 MCP Server 进程被启动。创建 Client在 LangChain 应用中创建MCPClient实例并配置其连接到目标 Server。获取工具调用client.get_tools()方法。该方法内部会与 Server 通信获取其声明的所有 Skills 的元数据。工具适配MCPClient根据元数据自动创建对应的MCPTool列表。注入 Agent将这些MCPTool添加到 LangChain Agent 的tools参数中。动态调用Agent 在推理过程中如果决定使用某个工具会调用对应的MCPTool._run()该方法会将请求转发给MCPClient再由Client通过 MCP 协议调用远程 Server 上的实际实现。优势你可以在不重启 Agent 应用的情况下通过启动新的 MCP Server 来为 Agent 增加新的能力。Agent 只需重新连接或刷新工具列表即可。4. 完整实战案例构建一个具备文件操作和网络搜索能力的 Agent接下来我们将通过一个完整的例子演示如何创建自定义 MCP Server并将其提供的 Skills 集成到 LangChain Agent 中。我们的目标是让 Agent 能够回答“请总结我桌面上的notes.txt文件内容并搜索最新的 LangChain 新闻”。4.1 创建自定义 MCP Server文件操作首先我们创建一个提供基础文件读取和写入技能的 MCP Server。# file: file_operations_server.py import json from typing import Any, List from mcp import Server, NotificationOptions import mcp.types as types from pathlib import Path class FileOperationsServer(Server): 一个提供基础文件操作的 MCP Server def __init__(self): super().__init__(file-operations-server) # 注册此 Server 提供的工具Skills self.register_tool( nameread_file, description读取指定路径的文本文件内容, input_schema{ type: object, properties: { file_path: { type: string, description: 要读取的文件的绝对路径 } }, required: [file_path] }, callbackself._read_file ) self.register_tool( namewrite_file, description向指定路径的文本文件写入内容, input_schema{ type: object, properties: { file_path: { type: string, description: 要写入的文件的绝对路径 }, content: { type: string, description: 要写入的文本内容 } }, required: [file_path, content] }, callbackself._write_file ) async def _read_file(self, arguments: dict) - str: 读取文件的工具实现 file_path Path(arguments[file_path]) if not file_path.exists(): return f错误文件 {file_path} 不存在。 if not file_path.is_file(): return f错误{file_path} 不是一个文件。 try: content file_path.read_text(encodingutf-8) return f文件 {file_path} 的内容如下\n\n{content}\n except Exception as e: return f读取文件时发生错误{e} async def _write_file(self, arguments: dict) - str: 写入文件的工具实现 file_path Path(arguments[file_path]) content arguments[content] try: # 确保目录存在 file_path.parent.mkdir(parentsTrue, exist_okTrue) file_path.write_text(content, encodingutf-8) return f成功将内容写入文件 {file_path}。 except Exception as e: return f写入文件时发生错误{e} async def list_tools(self) - List[types.Tool]: 列出所有可用工具MCP协议要求 return list(self.tools.values()) if __name__ __main__: import asyncio # 使用 stdio 传输层运行 Server这是与 Client 通信的标准方式 server FileOperationsServer() asyncio.run(server.run(transportstdio))这个 Server 定义了两个 Skillread_file和write_file。它通过stdio运行意味着它将通过标准输入/输出与客户端通信这是与 LangChain 集成最常见的方式。4.2 配置 LangChain Agent 并连接 MCP Server现在我们创建主程序启动上述 Server并让 LangChain Agent 使用其 Skills。# file: agent_with_mcp.py import asyncio import subprocess import sys from pathlib import Path from dotenv import load_dotenv from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_community.tools.mcp import MCPClient, create_mcp_tool from langchain_community.chat_models import ChatAnthropic # 使用 Claude # 也可以使用 OpenAI: from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate # 1. 加载环境变量用于存放 API Key load_dotenv() async def main(): # 2. 启动 MCP Server 进程 # 这里我们启动刚才编写的文件操作 Server server_script_path Path(__file__).parent / file_operations_server.py server_process subprocess.Popen( [sys.executable, str(server_script_path)], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, bufsize1 ) # 3. 创建 MCP Client 并连接到 Server # 注意我们使用 server_process 的 stdin/stdout 来通信 mcp_client MCPClient( server_process.stdin, server_process.stdout, # 可以配置多个 Server 的传输 transportstdio ) # 4. 初始化 Client 连接并获取远程工具列表 await mcp_client.initialize() # 获取 Server 提供的所有 Skills并自动转换为 LangChain Tool mcp_tools await mcp_client.get_tools() print(f从 MCP Server 获取到 {len(mcp_tools)} 个工具: {[t.name for t in mcp_tools]}) # 5. 创建 LangChain 模型这里以 Claude 为例 # 请确保你的 .env 文件中有 ANTHROPIC_API_KEY llm ChatAnthropic(modelclaude-3-sonnet-20240229, temperature0) # 如果使用 OpenAI则替换为 # from langchain_openai import ChatOpenAI # llm ChatOpenAI(modelgpt-4-turbo-preview) # 6. 定义 Agent 的提示词模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个强大的助手可以调用工具来完成用户的请求。请根据用户的问题决定是否需要使用工具以及使用哪个工具。在回复时对工具返回的结果进行总结和提炼。), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}), ]) # 7. 创建 Agent # 将 MCP Tools 和任何本地工具组合在一起 all_tools mcp_tools # 这里我们只使用 MCP Tools agent create_tool_calling_agent(llmllm, toolsall_tools, promptprompt) # 8. 创建 Agent 执行器 agent_executor AgentExecutor(agentagent, toolsall_tools, verboseTrue, handle_parsing_errorsTrue) # 9. 测试 Agent # 示例 1读取文件 print(\n--- 测试读取文件 ---) test_file_path Path.home() / Desktop / notes.txt # 为了测试我们先创建一个示例文件如果不存在 if not test_file_path.exists(): test_file_path.write_text(这是桌面上的笔记文件。\n记录了一些关于 MCP 和 LangChain 的学习心得。\n明天需要完成项目原型。) result1 await agent_executor.ainvoke({ input: f请帮我读取并总结这个文件的内容{test_file_path} }) print(结果:, result1[output]) # 示例 2写入文件 print(\n--- 测试写入文件 ---) result2 await agent_executor.ainvoke({ input: f请帮我在桌面创建一个名为 todo.txt 的文件内容写上购买 groceries和完成 LangChain 文章。 }) print(结果:, result2[output]) # 10. 清理停止 Server 进程 server_process.terminate() await server_process.wait() if __name__ __main__: asyncio.run(main())4.3 运行与验证设置 API Key在项目根目录创建.env文件填入你的 Anthropic 或 OpenAI API Key。# .env ANTHROPIC_API_KEYyour_anthropic_api_key_here # 或 # OPENAI_API_KEYyour_openai_api_key_here运行程序python agent_with_mcp.py预期输出程序会启动 MCP Server连接并获取工具列表然后 Agent 会开始执行。在verboseTrue模式下你将看到详细的 Agent 思考过程ReAct 模式包括它决定调用哪个工具、传递什么参数、以及工具返回的结果。最终Agent 会输出对文件内容的总结或操作成功的确认信息。4.4 集成更多 Skills添加网络搜索能力仅仅文件操作还不够。我们可以轻松集成另一个提供网络搜索技能的 MCP Server。实际上社区已经有很多现成的 MCP Server。这里我们以连接一个“模拟”的搜索 Server 为例展示如何连接多个 Server。假设我们有一个运行在本地 8000 端口的搜索 MCP Server例如使用mcp serve命令启动的brave-searchserver。我们需要修改主程序创建第二个MCPClient来连接这个搜索 Server。# 在 agent_with_mcp.py 的 main 函数中修改工具集合部分 async def main(): # ... [启动 file_operations_server 的代码不变] ... # 创建第一个 Client文件操作 mcp_client_files MCPClient(server_process.stdin, server_process.stdout, transportstdio) await mcp_client_files.initialize() file_tools await mcp_client_files.get_tools() # 创建第二个 Client网络搜索- 假设搜索 Server 运行在 SSE 模式 import aiohttp from langchain_community.tools.mcp import MCPClient # 注意这里需要根据实际搜索 Server 的传输方式调整 # 例如如果搜索 Server 通过 HTTP SSE 暴露可能需要不同的连接方式 # 以下为概念性代码具体取决于 Server 实现 # async with aiohttp.ClientSession() as session: # mcp_client_search MCPClient(session, http://localhost:8000/sse, transportsse) # await mcp_client_search.initialize() # search_tools await mcp_client_search.get_tools() # 为了简化演示我们这里添加一个 LangChain 内置的 DuckDuckGo 搜索工具作为替代 from langchain_community.tools import DuckDuckGoSearchRun search_tool DuckDuckGoSearchRun() # 组合所有工具 all_tools file_tools [search_tool] print(f可用工具总数: {len(all_tools)}) # ... [后续创建 Agent 和执行的代码不变] ... # 测试组合能力 print(\n--- 测试组合能力读文件搜索---) result3 await agent_executor.ainvoke({ input: f先帮我读取 {test_file_path} 文件看看里面提到了什么技术。然后去网上搜索一下关于这项技术的最新动态。 }) print(结果:, result3[output])通过这种方式你的 Agent 就同时具备了文件操作和网络搜索的能力。你可以继续集成数据库查询、代码执行、图像处理等任何通过 MCP Server 暴露的 Skill。5. 常见问题与排查思路在集成 MCP 与 LangChain 的过程中你可能会遇到以下典型问题问题现象可能原因排查步骤与解决方案导入错误No module named langchain_community.tools.mcp1.langchain-community版本过低。2. 包未正确安装。1. 升级langchain-community:pip install -U langchain-community。2. 确认安装的版本是否包含 mcp 模块0.0.10 通常包含。运行时报错MCPClient初始化失败或通信错误1. MCP Server 进程未启动或已崩溃。2.stdio管道通信异常。3. MCP 协议版本不兼容。1. 检查server_process是否成功启动查看其stderr输出。2. 确保MCPClient使用的是正确的stdin和stdout对象。3. 尝试使用transportstdio并确保 Server 也以stdio模式运行。Agent 无法识别或错误调用 MCP Tool1. 工具描述不清晰导致 LLM 无法理解其用途。2. 输入参数 schema 定义有误与 LLM 预期不符。3. 工具返回结果格式不符合 Agent 预期。1. 在 MCP Server 中为每个 Tool 提供清晰、具体的description。2. 仔细检查input_schema确保properties和required字段定义正确。3. 确保 Tool 回调函数返回的是字符串且内容易于 LLM 理解。性能问题工具调用缓慢1. MCP Server 处理耗时。2. 网络延迟远程 Server。3. Agent 频繁调用复杂工具。1. 优化 MCP Server 内部逻辑。2. 对于远程 Server考虑使用更快的网络或将其部署在靠近 Agent 的位置。3. 在 Agent 提示词中引导其“思考”减少不必要的工具调用。无法连接到社区 MCP Server1. Server 地址或端口错误。2. Server 未运行或需要认证。3. 传输协议 (stdio/sse/http) 不匹配。1. 使用mcp ls命令如果安装了 mcp cli查看可用 Server。2. 参考该 Server 的文档确认启动命令和连接方式。3. 对于sse传输确保使用正确的 URL 端点如http://host:port/sse。通用排查流程隔离测试首先单独运行你的 MCP Server使用mcp inspect或编写一个简单的测试客户端来验证其是否能正常响应list_tools和call_tool。检查日志启用 LangChain Agent 的verboseTrue模式观察 Agent 的思考链和工具调用请求。简化问题从一个最简单的 “echo” MCP Server 和一个最简单的 Agent 开始逐步增加复杂度。6. 最佳实践与工程建议将 MCP 用于生产环境时需要考虑以下几点6.1 MCP Server 设计原则单一职责一个 MCP Server 应专注于一类功能如文件操作、数据库访问、特定 API 封装。这便于维护和复用。清晰的接口Tool 的name和description必须精准。好的描述能让 LLM 更准确地判断何时调用它。输入参数的description也同样重要。健壮的错误处理在 Tool 的回调函数中必须进行充分的错误捕获和验证如文件是否存在、参数是否合法并返回对人和LLM 都有意义的错误信息。资源管理妥善管理数据库连接、网络会话等资源考虑使用连接池或上下文管理器。6.2 LangChain Agent 集成优化工具筛选与分类不是所有 MCP Tool 都需要一次性加载给 Agent。可以根据会话上下文或用户权限动态加载不同的工具集。MCPClient.get_tools()返回的是列表你可以进行过滤。提示词工程在 Agent 的 System Prompt 中简要介绍可用的工具类别和能力引导 LLM 更有效地利用它们。例如“你可以使用文件操作工具来读写用户指定的文件使用搜索工具来获取最新信息。”超时与重试对 MCP 工具调用配置合理的超时和重试机制避免因单个 Server 无响应导致整个 Agent 卡住。异步并发利用asyncio并发启动多个 MCP Server 或并发调用多个工具提升整体响应速度。6.3 安全与权限这是重中之重MCP 极大地扩展了 Agent 的能力也意味着更大的风险。沙箱化对执行代码、访问文件系统、执行系统命令的 MCP Server必须运行在严格的沙箱环境中如 Docker 容器限制其权限和资源访问。输入验证与净化所有从 LLM 传递给 MCP Tool 的参数都必须视为不可信的需要在 Server 端进行严格的验证和净化防止路径遍历、命令注入等攻击。访问控制实现基于用户或会话的工具访问控制。不是每个用户都能使用所有 Skills。审计日志记录所有 MCP 工具的调用详情包括调用者、参数、结果、时间戳用于安全审计和问题追踪。6.4 生产环境部署进程管理使用systemd,supervisord或容器编排工具如 Kubernetes来管理 MCP Server 进程的生命周期确保其高可用。健康检查为 MCP Server 实现健康检查端点方便监控系统探测其状态。配置化将 MCP Server 的连接信息地址、传输方式、认证信息抽取到配置文件或环境变量中便于不同环境开发、测试、生产的切换。7. 总结与扩展方向通过本文的讲解和实战你应该已经掌握了 LangChain Agent 与 MCP 协议集成的核心技术。这套架构的核心优势在于“解耦”和“动态扩展”。AI Agent 的核心逻辑规划、推理、记忆与具体的能力实现工具被分离开后者可以通过标准的 MCP 协议以“插件”形式动态接入。下一步可以探索的方向探索丰富的社区 MCP Server社区已经涌现了大量优秀的 MCP Server例如brave-search: 提供网络搜索。github: 访问 GitHub API。sql: 执行 SQL 查询。filesystem: 更强大的文件系统操作。 使用mcp ls或查阅https://github.com/modelcontextprotocol/servers来发现更多。开发自定义业务 Skill将企业内部系统的 API如 CRM、ERP、数据库封装成 MCP Server让 Agent 能够安全、可控地操作业务数据。与 Claude Desktop/Code 集成MCP 协议最初由 Anthropic 推动Claude Desktop 和 Claude Code 编辑器原生支持 MCP。这意味着你开发的 MCP Server 可以直接被这些客户端使用极大地扩展了 Claude 的能力边界。结合 LangGraph 实现复杂工作流LangChain 的 LangGraph 库支持构建有状态的、多环节的 Agent 工作流。你可以将 MCP Tools 嵌入到 LangGraph 的节点中构建出能够处理复杂、多步骤任务的超级助手。技术的最终目标是服务于业务。当你需要为 AI Agent 添加一个新能力时不再需要修改核心 Agent 代码只需启动或连接一个提供该能力的 MCP Server。这种架构为构建真正强大、灵活且易于维护的 AI 应用打开了新的大门。现在就从将一个现有的脚本或 API 改造成 MCP Server 开始你的实践吧。