大家好我是专注于AI应用开发的技术博主。在构建智能体Agent应用时你是否遇到过这样的困境想让Agent调用一个外部工具却需要编写大量胶水代码来处理API调用、数据转换和错误处理或者当你想复用一套工具给不同的Agent使用时发现配置繁琐且难以维护。这正是传统Agent开发中工具集成的痛点。本文将深入探讨如何通过LangChain Agent接入MCPModel Context Protocol与Skills来解决这些问题。我们将从核心概念入手逐步拆解其技术原理并通过一个完整的实战案例展示如何利用这套组合拳为你的AI Agent赋予强大的外部能力从而全方位提升开发效率与应用智能化水平。无论你是刚接触LangChain的新手还是希望优化现有Agent架构的开发者都能从中获得可直接复用的代码与配置方案。1. 背景与核心概念为什么需要MCP和Skills在深入技术细节之前我们首先要理解当前AI Agent开发面临的挑战以及MCP和Skills旨在解决什么问题。传统Agent工具集成的痛点一个典型的LangChain Agent工作流程是接收用户输入 - LLM大语言模型思考 - 决定调用哪个工具 - 执行工具 - 解析工具结果 - 继续思考或返回最终答案。这里的“工具”可以是搜索引擎、数据库查询、代码执行器等。然而传统的工具集成方式存在几个明显问题紧耦合工具的逻辑、API调用、参数解析直接写在Agent的代码中难以复用和独立更新。协议不统一不同工具如数据库、文件系统、Web API有各自的通信协议和数据格式Agent需要为每一种工具编写特定的适配器。上下文管理复杂如何安全、高效地将大型文档、数据库schema等“上下文”信息提供给LLM是一个复杂的工程问题。开发效率低每增加一个新工具或能力都需要重新修改Agent核心代码并进行大量测试。MCP (Model Context Protocol)协议层解耦MCP是一个开放协议它定义了一套标准化的方式让服务器Server可以向客户端Client提供“资源”Resources和“工具”Tools。这里的客户端通常就是AI应用或Agent框架如LangChain。它的核心价值在于解耦和标准化。对于工具提供方Server只需按照MCP协议实现一个服务对外暴露统一的接口如SSE或stdio声明自己有哪些工具和资源即可无需关心客户端是LangChain、Claude Desktop还是其他任何兼容MCP的应用。对于Agent开发者Client无需关心工具后端的实现细节是Python、Node.js还是Go也无需处理具体的HTTP API。只需要按照MCP协议去“发现”和“调用”服务器提供的工具大大降低了集成复杂度。Skills能力的模块化封装Skills技能是一个更上层的概念它代表Agent所能执行的一个个具体能力单元。在LangChain的语境下一个Skill通常由一个或多个Tools工具和相关的Prompts提示词封装而成形成一个可复用的功能模块。例如“文件阅读技能”可能包含“列出目录”、“读取文件内容”等多个工具并配有针对文件操作的优化提示词。 MCP可以看作是Skills底层依赖的“协议总线”或“能力供给层”。一个Skill的实现其背后调用的工具完全可以由MCP Server来提供。三者关系总结AI大模型 (LLM)提供核心的推理与决策能力。LangChain Agent作为智能体的“大脑”和“调度中心”基于LLM的决策来协调执行。MCP (Protocol)作为“神经系统”或“标准接口”定义了能力供给的规范。Skills / Tools (via MCP Server)作为“四肢”和“感官”提供具体的外部世界交互能力。通过MCPAgent可以动态、安全地接入各种外部能力Skills而无需与具体实现绑定这正是提升开发效率和系统可维护性的关键。2. 环境准备与版本说明在开始实战之前我们需要搭建好开发环境。本文将使用Python作为主要开发语言。核心环境要求操作系统macOS / Linux / Windows (WSL2推荐)Python版本 3.10 强烈推荐3.10或3.11以确保依赖兼容性包管理工具pip 或 poetry主要依赖库及版本说明我们将使用langchain核心库以及langchain-community来构建Agent同时需要mcp客户端库来连接MCP服务器。以下版本在撰写本文时经过测试具有较好的稳定性。# 创建并进入项目目录 mkdir langchain-mcp-agent cd langchain-mcp-agent # 创建虚拟环境可选但推荐 python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 安装核心依赖 pip install langchain0.1.0 pip install langchain-community0.0.10 pip install langchain-openai0.0.5 # 如果你使用OpenAI模型 pip install mcp0.1.0 # MCP客户端库 # 可选安装用于示例的MCP服务器库 # 例如一个简单的计算器MCP服务器 pip install mcp-server-demo # 这是一个假设的示例包实际需寻找可用实现版本兼容性提示LangChain和MCP生态发展迅速版本迭代可能带来API变化。如果遇到问题请优先查阅官方文档。本文的重点是阐述原理和通用模式代码示例会尽量使用稳定接口。项目结构预览langchain-mcp-agent/ ├── venv/ # Python虚拟环境 ├── mcp_servers/ # 存放MCP服务器实现或配置 │ └── calculator_server.py # 示例计算器MCP服务器 ├── agents/ # 存放Agent核心逻辑 │ └── mcp_agent.py # 集成MCP的LangChain Agent ├── .env # 环境变量如API密钥 ├── requirements.txt # 项目依赖 └── main.py # 应用入口3. 核心原理拆解MCP协议与LangChain集成机制要熟练应用必须理解其内部工作原理。本节将深入MCP协议的工作流程和LangChain的集成点。3.1 MCP协议通信模型MCP协议主要采用两种传输方式Stdio标准输入输出和SSE服务器发送事件。Stdio模式更适合本地CLI工具或守护进程而SSE模式适用于网络服务。我们以Stdio模式为例其工作流程如下启动与初始化客户端我们的LangChain程序启动一个MCP服务器进程例如python calculator_server.py。客户端通过标准输入(stdin)向服务器发送初始化请求。能力发现服务器通过标准输出(stdout)回复告知客户端自己提供了哪些“工具”(Tools)和“资源”(Resources)。工具代表可执行的操作如“相加两个数”资源代表可读取的数据如“数据库schema”。工具调用当Agent需要执行某个操作时客户端通过stdin向服务器发送一个tools/call请求包含工具名和参数。结果返回服务器执行工具并通过stdout返回执行结果或错误信息。会话保持整个会话期间进程保持连接允许多次工具调用。关键协议消息示例概念性JSON// 客户端 - 服务器初始化 {jsonrpc: 2.0, id: 1, method: initialize, params: {...}} // 服务器 - 客户端列出可用工具 {jsonrpc: 2.0, id: 1, result: {tools: [{name: add, description: Add two numbers, inputSchema: {...}}]}} // 客户端 - 服务器调用工具 {jsonrpc: 2.0, id: 2, method: tools/call, params: {name: add, arguments: {a: 5, b: 3}}} // 服务器 - 客户端返回结果 {jsonrpc: 2.0, id: 2, result: {content: [{type: text, text: 8}]}}3.2 LangChain如何集成MCP ToolsLangChain通过langchain.agents.Agent框架来使用工具。要让Agent使用MCP提供的工具我们需要一个“适配器”将MCP协议下的工具转换为LangChain能识别的Tool对象。这个适配过程的核心是连接MCP服务器使用MCP客户端库建立与MCP Server的连接。获取工具列表调用MCP的list_tools方法获取服务器暴露的所有工具定义。动态创建LangChain Tool遍历工具列表为每个MCP工具创建一个StructuredTool或BaseTool实例。这个实例的_run方法内部会通过MCP客户端发起tools/call请求。注入Agent将创建好的Tool列表提供给LangChain Agent如create_react_agent。这样当Agent决定调用add工具时实际上调用的是我们创建的适配器Tool该适配器再将调用转发给MCP服务器并返回结果。3.3 Skills的构建逻辑Skill是比单一Tool更高级的抽象。一个Skill可能包含多个相关Tools例如“文件管理Skill”包含读、写、删、列表等工具。包含特定的系统提示词(Prompt)指导LLM更好地使用这些工具例如“你是一个文件助手可以帮用户操作文件...”。包含预定义的Few-shot示例提供工具使用的范例。在实践中我们可以将一个MCP Server提供的所有工具打包成一个Skill也可以组合多个MCP Server的工具来形成一个功能更复杂的Skill。Skill的封装使得能力模块可以像插件一样被Agent加载和卸载。4. 完整实战案例构建一个接入MCP计算器的智能体现在我们动手实现一个完整的例子。我们将创建一个简单的MCP服务器提供计算器功能然后构建一个LangChain Agent来动态接入并使用这个服务器提供的工具。4.1 创建MCP服务器能力提供方首先我们实现一个提供基本数学运算的MCP服务器。我们将使用mcp库的服务器SDK。文件mcp_servers/calculator_server.py#!/usr/bin/env python3 import sys from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio from mcp.types import Tool, TextContent # 创建MCP服务器实例 server Server(calculator-mcp-server) # 定义工具加法 server.list_tools() async def handle_list_tools(): return [ Tool( nameadd, descriptionAdd two numbers together., inputSchema{ type: object, properties: { a: {type: number, description: The first number}, b: {type: number, description: The second number}, }, required: [a, b], }, ), Tool( namemultiply, descriptionMultiply two numbers together., inputSchema{ type: object, properties: { x: {type: number, description: The first factor}, y: {type: number, description: The second factor}, }, required: [x, y], }, ), ] # 处理工具调用加法 server.call_tool() async def handle_call_tool(name: str, arguments: dict) - list[TextContent]: if name add: result arguments[a] arguments[b] return [TextContent(typetext, textstr(result))] elif name multiply: result arguments[x] * arguments[y] return [TextContent(typetext, textstr(result))] else: raise ValueError(fUnknown tool: {name}) # 主函数启动stdio服务器 async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_namecalculator-server, server_version0.1.0, capabilitiesserver.get_capabilities( notification_optionsNotificationOptions(), experimental_capabilities{}, ), ), ) if __name__ __main__: import asyncio asyncio.run(main())这个服务器通过Stdio运行向外提供了add和multiply两个工具。每个工具都有清晰的描述和输入参数定义这至关重要因为LangChain Agent的LLM需要根据这些描述来决定是否以及如何调用它们。4.2 构建LangChain Agent能力使用方接下来我们创建LangChain Agent它将连接并利用上面的MCP服务器。文件agents/mcp_agent.pyimport asyncio from typing import List, Any from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import BaseTool, StructuredTool, Tool from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage from mcp import ClientSession, StdioServerParameters from mcp.client import stdio class MCPToolWrapper(BaseTool): 将MCP工具包装成LangChain Tool的适配器 name: str description: str session: ClientSession # MCP会话 def _run(self, **kwargs: Any) - str: 同步运行方法实际调用异步方法 # 注意在异步环境外运行需要asyncio.run生产环境建议用异步Agent return asyncio.run(self._arun(**kwargs)) async def _arun(self, **kwargs: Any) - str: 异步运行方法调用MCP工具 try: result await self.session.call_tool(self.name, argumentskwargs) # 假设返回的是TextContent列表拼接所有文本 texts [c.text for c in result.content if hasattr(c, text)] return \n.join(texts) if texts else Tool executed successfully (no text output). except Exception as e: return fError calling tool {self.name}: {str(e)} async def create_mcp_tools(server_params: StdioServerParameters) - List[BaseTool]: 连接MCP服务器并创建对应的LangChain Tools tools [] # 建立与MCP服务器的连接 async with stdio.stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # 初始化会话 await session.initialize() # 获取服务器提供的所有工具 mcp_tools await session.list_tools() for tool_info in mcp_tools: # 为每个MCP工具创建一个包装器 # 注意这里简化了实际需要将会话或客户端传递给包装器。 # 更稳健的做法是创建一个工具工厂管理会话生命周期。 # 此处为演示我们创建一个需要后续注入session的Tool。 # 我们先定义工具稍后在主流程中注入session。 def make_tool_func(t_name, t_desc, t_session): async def func(**kwargs): result await t_session.call_tool(t_name, argumentskwargs) texts [c.text for c in result.content if hasattr(c, text)] return \n.join(texts) if texts else Done. return func # 由于需要异步我们使用StructuredTool.from_function # 但需要处理session绑定问题。这里展示一个更清晰的模式 pass # 具体实现见主函数 return tools # 由于MCP工具的动态性更常见的模式是在主异步函数中集中创建 async def create_agent_with_mcp_tools(): 主函数创建集成MCP工具的Agent # 1. 定义MCP服务器我们刚写的计算器服务器 server_params StdioServerParameters( commandpython, args[mcp_servers/calculator_server.py] ) # 2. 连接MCP服务器并获取工具 async with stdio.stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() mcp_tool_defs await session.list_tools() # 3. 动态构建LangChain Tools langchain_tools [] for tool_def in mcp_tool_defs: # 为每个工具创建一个异步函数闭包捕获当前的session和tool_def async def tool_func(**kwargs): try: result await session.call_tool(tool_def.name, argumentskwargs) texts [c.text for c in result.content if hasattr(c, text)] return \n.join(texts) except Exception as e: return fTool error: {e} # 使用StructuredTool.from_function创建Tool # 注意需要将异步函数包装成同步函数或使用支持异步的Agent。 # LangChain的Tool默认是同步的。我们可以使用asyncio.run包装但更好的方式是使用异步Agent执行器。 # 这里我们创建一个同步包装器以简化演示。 def sync_wrapper(**kwargs): return asyncio.run(tool_func(**kwargs)) langchain_tool StructuredTool.from_function( funcsync_wrapper, nametool_def.name, descriptiontool_def.description, args_schemaNone, # 可以根据tool_def.inputSchema动态生成这里简化 ) langchain_tools.append(langchain_tool) # 4. 初始化LLM以OpenAI为例需要设置环境变量OPENAI_API_KEY llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 5. 创建ReAct Agent的提示词 prompt PromptTemplate.from_template( 你是一个有帮助的助手可以使用以下工具 {tools} 使用以下格式 问题用户的问题 思考你需要思考做什么 行动要使用的工具名必须是[{tool_names}]中的一个 行动输入工具的输入必须是有效的JSON格式 观察工具返回的结果 ... (这个思考/行动/行动输入/观察可以重复多次) 最终答案根据观察得出的最终答案 开始 问题{input} 思考{agent_scratchpad} ) # 6. 创建Agent agent create_react_agent(llm, langchain_tools, prompt) # 7. 创建Agent执行器 agent_executor AgentExecutor(agentagent, toolslangchain_tools, verboseTrue, handle_parsing_errorsTrue) # 8. 测试Agent print(Agent已启动集成了来自MCP服务器的工具, [t.name for t in langchain_tools]) result await agent_executor.ainvoke({input: 请计算 42 加上 15 是多少然后再乘以 2。}) print(\n--- 最终答案 ---) print(result[output]) # 可以继续会话 # result2 await agent_executor.ainvoke({input: 刚才的结果减去10是多少}) if __name__ __main__: asyncio.run(create_agent_with_mcp_tools())4.3 运行与验证确保环境变量如果你使用OpenAI模型请设置OPENAI_API_KEY。export OPENAI_API_KEYyour-api-key-here # 或在项目根目录创建 .env 文件运行Agent程序python agents/mcp_agent.py预期输出 程序会启动连接到本地的计算器MCP服务器获取add和multiply工具然后启动Agent。当你提问“请计算 42 加上 15 是多少然后再乘以 2。”时Agent的思考过程如果verboseTrue会显示如下 进入新的AgentExecutor链... 思考用户想先计算4215再用结果乘以2。我需要使用加法工具和乘法工具。 行动add 行动输入{a: 42, b: 15} 观察57 思考现在我有57需要用它乘以2。 行动multiply 行动输入{x: 57, y: 2} 观察114 思考我得到了最终结果114。 最终答案42加15等于5757乘以2等于114。 链结束。 --- 最终答案 --- 42加15等于5757乘以2等于114。4.4 结果说明通过这个案例我们成功实现了一个独立的MCP服务器提供计算能力完全独立于Agent应用。一个动态集成MCP工具的LangChain AgentAgent在运行时发现并加载了MCP服务器提供的工具无需在代码中硬编码工具逻辑。完整的思考-行动循环Agent能够根据问题自主规划步骤依次调用正确的工具并整合结果。这验证了MCP协议在解耦工具与Agent方面的强大能力。你可以轻松地将calculator_server.py替换为任何其他MCP服务器如文件系统服务器、数据库服务器而mcp_agent.py的主体代码几乎无需改动Agent就能获得新的能力。5. 常见问题与排查思路在实际集成过程中你可能会遇到以下问题。这里提供一份排查清单。问题现象可能原因排查步骤与解决方案Agent无法启动提示MCP连接错误1. MCP服务器脚本路径错误。2. Python环境不一致服务器与客户端环境不同。3. MCP服务器脚本有语法错误或依赖缺失。1. 检查StdioServerParameters中的command和args确保能正确启动服务器。使用绝对路径更可靠。2. 确保Agent和MCP服务器在同一个Python虚拟环境中运行或使用全路径指定python解释器。3. 单独运行MCP服务器脚本(python mcp_servers/calculator_server.py)看是否有错误输出。Agent运行后找不到工具或工具列表为空1. MCP服务器未正确实现list_tools方法。2. 客户端与服务器初始化失败。3. 协议版本不兼容。1. 在MCP服务器的handle_list_tools函数中添加打印日志确认被调用。2. 检查客户端session.initialize()是否成功。可以在create_agent_with_mcp_tools函数中打印mcp_tool_defs。3. 确保使用的mcp库版本在客户端和服务器端尽量一致。工具调用失败返回参数错误1. Agent生成的参数格式与MCP工具定义的inputSchema不匹配。2. LLM未能正确理解工具描述。1. 在Agent的verbose输出中检查“行动输入”的JSON格式。确保参数名如a,b和类型数字正确。2. 优化MCP工具的描述(description)使其对LLM更清晰。例如“Add two numbers”比“Calculate sum”更好。3. 在Agent提示词中强化工具使用格式的要求。异步/同步上下文错误在同步代码中直接调用异步的MCP客户端方法。本文示例使用了asyncio.run在同步函数中调用异步函数这在简单脚本中可行。对于生产级应用如FastAPI服务建议1. 使用完全异步的LangChain Agent执行器如AgentExecutor.ainvoke。2. 确保整个调用链Web框架 - Agent - MCP工具都在异步上下文中。Agent陷入循环或调用错误工具1. 工具描述模糊导致LLM误解。2. ReAct提示词对复杂任务规划能力不足。1. 细化工具描述明确输入输出。例如“Multiply two integers”而非“Do multiplication”。2. 考虑使用更强大的Agent类型如create_openai_functions_agent如果LLM支持Function Calling它对工具调用的支持更精准。3. 在系统提示词中加入工具使用约束如“一次只使用一个工具”。性能问题工具调用慢1. 每次调用都重新启动MCP服务器进程。2. MCP服务器本身处理慢。1.复用连接确保MCP服务器进程在Agent生命周期内保持运行而不是每次调用都新建。本文的async with上下文管理器实现了这一点。2.池化与超时对于高并发场景考虑实现MCP客户端连接池。为工具调用设置合理的超时时间。6. 最佳实践与工程建议将MCP和Skills投入生产环境需要遵循一些工程最佳实践。6.1 MCP Server设计规范单一职责一个MCP Server应专注于提供一类能力。例如calculator-server、filesystem-server、sql-server。避免创建“上帝服务器”。清晰的工具定义name使用动词开头如search_web,read_file。description详细描述工具功能、输入输出。这是LLM理解工具的关键。例如“根据关键词在互联网上搜索最新信息。输入query(字符串搜索关键词)。输出搜索结果的摘要列表。”inputSchema严格定义参数类型和是否必需。使用JSON Schema标准。健壮的错误处理在MCP Server的call_tool函数中捕获所有内部异常并返回结构化的错误信息而不是让进程崩溃。资源管理如果工具涉及资源如数据库连接、文件句柄确保在服务器生命周期内妥善管理。考虑使用连接池。6.2 LangChain Agent集成优化会话管理不要为每个用户请求都创建新的MCP连接。应该在应用启动时初始化并复用MCP客户端会话或使用连接池。工具动态发现与更新实现一个后台任务定期或通过通知检查MCP Server的工具列表是否有更新并动态更新Agent的Tool列表。这可以实现技能的“热插拔”。使用Function Calling如果底层LLM支持Function Calling如GPT-4, Claude优先使用create_openai_functions_agent。它能产生更结构化、更可靠的工具调用参数。技能(Skill)封装不要直接将一堆MCP Tools扔给Agent。将相关的Tools和特定的系统提示词封装成一个Skill类。例如class DataAnalysisSkill: def __init__(self, mcp_session): self.tools self._load_tools(mcp_session) # 加载数据库查询、图表生成等工具 self.system_prompt 你是一个数据分析专家擅长使用SQL查询数据和绘制图表... def get_tools(self): return self.tools def get_prompt(self): return self.system_prompt这样你可以根据用户意图动态地为Agent加载不同的Skill。6.3 安全与权限最小权限原则MCP Server运行时应具有完成其功能所需的最小系统权限。例如文件服务器不应有权限删除根目录。输入验证与净化在MCP Server端对输入参数进行严格验证防止注入攻击如SQL注入、命令注入。访问控制在Agent与MCP Server之间或在一个中心化的MCP路由层实现基于用户/角色的工具访问控制。不是所有用户都能调用所有工具。审计日志记录所有的工具调用包括调用者、参数、结果和时间戳便于追踪和调试。6.4 生产环境部署容器化将每个MCP Server容器化Docker便于独立部署、扩展和管理。服务发现与健康检查当有多个MCP Server时需要一套机制让Agent发现可用的Server。可以结合服务发现工具如Consul或简单的注册中心。监控与告警监控MCP Server的进程状态、资源使用率和错误率。设置告警确保能力可用性。版本管理对MCP Server的接口工具列表和参数schema进行版本管理。Agent客户端应能处理向后兼容或版本协商。通过遵循这些最佳实践你可以构建出一个灵活、健壮、可扩展的基于LangChain Agent和MCP的技能化AI应用系统。这不仅仅是技术集成更是一种面向未来的AI应用架构范式。 大家好我是专注于AI应用开发的技术博主。在构建智能体Agent应用时你是否遇到过这样的困境想让Agent调用一个外部工具却需要编写大量胶水代码来处理API调用、数据转换和错误处理或者当你想复用一套工具给不同的Agent使用时发现配置繁琐且难以维护。这正是传统Agent开发中工具集成的痛点。本文将深入探讨如何通过LangChain Agent接入MCPModel Context Protocol与Skills来解决这些问题。我们将从核心概念入手逐步拆解其技术原理并通过一个完整的实战案例展示如何利用这套组合拳为你的AI Agent赋予强大的外部能力从而全方位提升开发效率与应用智能化水平。无论你是刚接触LangChain的新手还是希望优化现有Agent架构的开发者都能从中获得可直接复用的代码与配置方案。1. 背景与核心概念为什么需要MCP和Skills在深入技术细节之前我们首先要理解当前AI Agent开发面临的挑战以及MCP和Skills旨在解决什么问题。传统Agent工具集成的痛点一个典型的LangChain Agent工作流程是接收用户输入 - LLM大语言模型思考 - 决定调用哪个工具 - 执行工具 - 解析工具结果 - 继续思考或返回最终答案。这里的“工具”可以是搜索引擎、数据库查询、代码执行器等。然而传统的工具集成方式存在几个明显问题紧耦合工具的逻辑、API调用、参数解析直接写在Agent的代码中难以复用和独立更新。协议不统一不同工具如数据库、文件系统、Web API有各自的通信协议和数据格式Agent需要为每一种工具编写特定的适配器。上下文管理复杂如何安全、高效地将大型文档、数据库schema等“上下文”信息提供给LLM是一个复杂的工程问题。开发效率低每增加一个新工具或能力都需要重新修改Agent核心代码并进行大量测试。MCP (Model Context Protocol)协议层解耦MCP是一个开放协议它定义了一套标准化的方式让服务器Server可以向客户端Client提供“资源”Resources和“工具”Tools。这里的客户端通常就是AI应用或Agent框架如LangChain。它的核心价值在于解耦和标准化。对于工具提供方Server只需按照MCP协议实现一个服务对外暴露统一的接口如SSE或stdio声明自己有哪些工具和资源即可无需关心客户端是LangChain、Claude Desktop还是其他任何兼容MCP的应用。对于Agent开发者Client无需关心工具后端的实现细节是Python、Node.js还是Go也无需处理具体的HTTP API。只需要按照MCP协议去“发现”和“调用”服务器提供的工具大大降低了集成复杂度。Skills能力的模块化封装Skills技能是一个更上层的概念它代表Agent所能执行的一个个具体能力单元。在LangChain的语境下一个Skill通常由一个或多个Tools工具和相关的Prompts提示词封装而成形成一个可复用的功能模块。例如“文件阅读技能”可能包含“列出目录”、“读取文件内容”等多个工具并配有针对文件操作的优化提示词。 MCP可以看作是Skills底层依赖的“协议总线”或“能力供给层”。一个Skill的实现其背后调用的工具完全可以由MCP Server来提供。三者关系总结AI大模型 (LLM)提供核心的推理与决策能力。LangChain Agent作为智能体的“大脑”和“调度中心”基于LLM的决策来协调执行。MCP (Protocol)作为“神经系统”或“标准接口”定义了能力供给的规范。Skills / Tools (via MCP Server)作为“四肢”和“感官”提供具体的外部世界交互能力。通过MCPAgent可以动态、安全地接入各种外部能力Skills而无需与具体实现绑定这正是提升开发效率和系统可维护性的关键。2. 环境准备与版本说明在开始实战之前我们需要搭建好开发环境。本文将使用Python作为主要开发语言。核心环境要求操作系统macOS / Linux / Windows (WSL2推荐)Python版本 3.10 强烈推荐3.10或3.11以确保依赖兼容性包管理工具pip 或 poetry主要依赖库及版本说明我们将使用langchain核心库以及langchain-community来构建Agent同时需要mcp客户端库来连接MCP服务器。以下版本在撰写本文时经过测试具有较好的稳定性。# 创建并进入项目目录 mkdir langchain-mcp-agent cd langchain-mcp-agent # 创建虚拟环境可选但推荐 python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 安装核心依赖 pip install langchain0.1.0 pip install langchain-community0.0.10 pip install langchain-openai0.0.5 # 如果你使用OpenAI模型 pip install mcp0.1.0 # MCP客户端库 # 可选安装用于示例的MCP服务器库 # 例如一个简单的计算器MCP服务器 pip install mcp-server-demo # 这是一个假设的示例包实际需寻找可用实现版本兼容性提示LangChain和MCP生态发展迅速版本迭代可能带来API变化。如果遇到问题请优先查阅官方文档。本文的重点是阐述原理和通用模式代码示例会尽量使用稳定接口。项目结构预览langchain-mcp-agent/ ├── venv/ # Python虚拟环境 ├── mcp_servers/ # 存放MCP服务器实现或配置 │ └── calculator_server.py # 示例计算器MCP服务器 ├── agents/ # 存放Agent核心逻辑 │ └── mcp_agent.py # 集成MCP的LangChain Agent ├── .env # 环境变量如API密钥 ├── requirements.txt # 项目依赖 └── main.py # 应用入口3. 核心原理拆解MCP协议与LangChain集成机制要熟练应用必须理解其内部工作原理。本节将深入MCP协议的工作流程和LangChain的集成点。3.1 MCP协议通信模型MCP协议主要采用两种传输方式Stdio标准输入输出和SSE服务器发送事件。Stdio模式更适合本地CLI工具或守护进程而SSE模式适用于网络服务。我们以Stdio模式为例其工作流程如下启动与初始化客户端我们的LangChain程序启动一个MCP服务器进程例如python calculator_server.py。客户端通过标准输入(stdin)向服务器发送初始化请求。能力发现服务器通过标准输出(stdout)回复告知客户端自己提供了哪些“工具”(Tools)和“资源”(Resources)。工具代表可执行的操作如“相加两个数”资源代表可读取的数据如“数据库schema”。工具调用当Agent需要执行某个操作时客户端通过stdin向服务器发送一个tools/call请求包含工具名和参数。结果返回服务器执行工具并通过stdout返回执行结果或错误信息。会话保持整个会话期间进程保持连接允许多次工具调用。关键协议消息示例概念性JSON// 客户端 - 服务器初始化 {jsonrpc: 2.0, id: 1, method: initialize, params: {...}} // 服务器 - 客户端列出可用工具 {jsonrpc: 2.0, id: 1, result: {tools: [{name: add, description: Add two numbers, inputSchema: {...}}]}} // 客户端 - 服务器调用工具 {jsonrpc: 2.0, id: 2, method: tools/call, params: {name: add, arguments: {a: 5, b: 3}}} // 服务器 - 客户端返回结果 {jsonrpc: 2.0, id: 2, result: {content: [{type: text, text: 8}]}}3.2 LangChain如何集成MCP ToolsLangChain通过langchain.agents.Agent框架来使用工具。要让Agent使用MCP提供的工具我们需要一个“适配器”将MCP协议下的工具转换为LangChain能识别的Tool对象。这个适配过程的核心是连接MCP服务器使用MCP客户端库建立与MCP Server的连接。获取工具列表调用MCP的list_tools方法获取服务器暴露的所有工具定义。动态创建LangChain Tool遍历工具列表为每个MCP工具创建一个StructuredTool或BaseTool实例。这个实例的_run方法内部会通过MCP客户端发起tools/call请求。注入Agent将创建好的Tool列表提供给LangChain Agent如create_react_agent。这样当Agent决定调用add工具时实际上调用的是我们创建的适配器Tool该适配器再将调用转发给MCP服务器并返回结果。3.3 Skills的构建逻辑Skill是比单一Tool更高级的抽象。一个Skill可能包含多个相关Tools例如“文件管理Skill”包含读、写、删、列表等工具。包含特定的系统提示词(Prompt)指导LLM更好地使用这些工具例如“你是一个文件助手可以帮用户操作文件...”。包含预定义的Few-shot示例提供工具使用的范例。在实践中我们可以将一个MCP Server提供的所有工具打包成一个Skill也可以组合多个MCP Server的工具来形成一个功能更复杂的Skill。Skill的封装使得能力模块可以像插件一样被Agent加载和卸载。4. 完整实战案例构建一个接入MCP计算器的智能体现在我们动手实现一个完整的例子。我们将创建一个简单的MCP服务器提供计算器功能然后构建一个LangChain Agent来动态接入并使用这个服务器提供的工具。4.1 创建MCP服务器能力提供方首先我们实现一个提供基本数学运算的MCP服务器。我们将使用mcp库的服务器SDK。文件mcp_servers/calculator_server.py#!/usr/bin/env python3 import sys from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio from mcp.types import Tool, TextContent # 创建MCP服务器实例 server Server(calculator-mcp-server) # 定义工具加法 server.list_tools() async def handle_list_tools(): return [ Tool( nameadd, descriptionAdd two numbers together., inputSchema{ type: object, properties: { a: {type: number, description: The first number}, b: {type: number, description: The second number}, }, required: [a, b], }, ), Tool( namemultiply, descriptionMultiply two numbers together., inputSchema{ type: object, properties: { x: {type: number, description: The first factor}, y: {type: number, description: The second factor}, }, required: [x, y], }, ), ] # 处理工具调用加法 server.call_tool() async def handle_call_tool(name: str, arguments: dict) - list[TextContent]: if name add: result arguments[a] arguments[b] return [TextContent(typetext, textstr(result))] elif name multiply: result arguments[x] * arguments[y] return [TextContent(typetext, textstr(result))] else: raise ValueError(fUnknown tool: {name}) # 主函数启动stdio服务器 async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_namecalculator-server, server_version0.1.0, capabilitiesserver.get_capabilities( notification_optionsNotificationOptions(), experimental_capabilities{}, ), ), ) if __name__ __main__: import asyncio asyncio.run(main())这个服务器通过Stdio运行向外提供了add和multiply两个工具。每个工具都有清晰的描述和输入参数定义这至关重要因为LangChain Agent的LLM需要根据这些描述来决定是否以及如何调用它们。4.2 构建LangChain Agent能力使用方接下来我们创建LangChain Agent它将连接并利用上面的MCP服务器。文件agents/mcp_agent.pyimport asyncio from typing import List, Any from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import BaseTool, StructuredTool, Tool from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage from mcp import ClientSession, StdioServerParameters from mcp.client import stdio class MCPToolWrapper(BaseTool): 将MCP工具包装成LangChain Tool的适配器 name: str description: str session: ClientSession # MCP会话 def _run(self, **kwargs: Any) - str: 同步运行方法实际调用异步方法 # 注意在异步环境外运行需要asyncio.run生产环境建议用异步Agent return asyncio.run(self._arun(**kwargs)) async def _arun(self, **kwargs: Any) - str: 异步运行方法调用MCP工具 try: result await self.session.call_tool(self.name, argumentskwargs) # 假设返回的是TextContent列表拼接所有文本 texts [c.text for c in result.content if hasattr(c, text)] return \n.join(texts) if texts else Tool executed successfully (no text output). except Exception as e: return fError calling tool {self.name}: {str(e)} async def create_mcp_tools(server_params: StdioServerParameters) - List[BaseTool]: 连接MCP服务器并创建对应的LangChain Tools tools [] # 建立与MCP服务器的连接 async with stdio.stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # 初始化会话 await session.initialize() # 获取服务器提供的所有工具 mcp_tools await session.list_tools() for tool_info in mcp_tools: # 为每个MCP工具创建一个包装器 # 注意这里简化了实际需要将会话或客户端传递给包装器。 # 更稳健的做法是创建一个工具工厂管理会话生命周期。 # 此处为演示我们创建一个需要后续注入session的Tool。 # 我们先定义工具稍后在主流程中注入session。 def make_tool_func(t_name, t_desc, t_session): async def func(**kwargs): result await t_session.call_tool(t_name, argumentskwargs) texts [c.text for c in result.content if hasattr(c, text)] return \n.join(texts) if texts else Done. return func # 由于需要异步我们使用StructuredTool.from_function # 但需要处理session绑定问题。这里展示一个更清晰的模式 pass # 具体实现见主函数 return tools # 由于MCP工具的动态性更常见的模式是在主异步函数中集中创建 async def create_agent_with_mcp_tools(): 主函数创建集成MCP工具的Agent # 1. 定义MCP服务器我们刚写的计算器服务器 server_params StdioServerParameters( commandpython, args[mcp_servers/calculator_server.py] ) # 2. 连接MCP服务器并获取工具 async with stdio.stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() mcp_tool_defs await session.list_tools() # 3. 动态构建LangChain Tools langchain_tools [] for tool_def in mcp_tool_defs: # 为每个工具创建一个异步函数闭包捕获当前的session和tool_def async def tool_func(**kwargs): try: result await session.call_tool(tool_def.name, argumentskwargs) texts [c.text for c in result.content if hasattr(c, text)] return \n.join(texts) except Exception as e: return fTool error: {e} # 使用StructuredTool.from_function创建Tool # 注意需要将异步函数包装成同步函数或使用支持异步的Agent。 # LangChain的Tool默认是同步的。我们可以使用asyncio.run包装但更好的方式是使用异步Agent执行器。 # 这里我们创建一个同步包装器以简化演示。 def sync_wrapper(**kwargs): return asyncio.run(tool_func(**kwargs)) langchain_tool StructuredTool.from_function( funcsync_wrapper, nametool_def.name, descriptiontool_def.description, args_schemaNone, # 可以根据tool_def.inputSchema动态生成这里简化 ) langchain_tools.append(langchain_tool) # 4. 初始化LLM以OpenAI为例需要设置环境变量OPENAI_API_KEY llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 5. 创建ReAct Agent的提示词 prompt PromptTemplate.from_template( 你是一个有帮助的助手可以使用以下工具 {tools} 使用以下格式 问题用户的问题 思考你需要思考做什么 行动要使用的工具名必须是[{tool_names}]中的一个 行动输入工具的输入必须是有效的JSON格式 观察工具返回的结果 ... (这个思考/行动/行动输入/观察可以重复多次) 最终答案根据观察得出的最终答案 开始 问题{input} 思考{agent_scratchpad} ) # 6. 创建Agent agent create_react_agent(llm, langchain_tools, prompt) # 7. 创建Agent执行器 agent_executor AgentExecutor(agentagent, toolslangchain_tools, verboseTrue, handle_parsing_errorsTrue) # 8. 测试Agent print(Agent已启动集成了来自MCP服务器的工具, [t.name for t in langchain_tools]) result await agent_executor.ainvoke({input: 请计算 42 加上 15 是多少然后再乘以 2。}) print(\n--- 最终答案 ---) print(result[output]) # 可以继续会话 # result2 await agent_executor.ainvoke({input: 刚才的结果减去10是多少}) if __name__ __main__: asyncio.run(create_agent_with_mcp_tools())4.3 运行与验证确保环境变量如果你使用OpenAI模型请设置OPENAI_API_KEY。export OPENAI_API_KEYyour-api-key-here # 或在项目根目录创建 .env 文件运行Agent程序python agents/mcp_agent.py预期输出 程序会启动连接到本地的计算器MCP服务器获取add和multiply工具然后启动Agent。当你提问“请计算 42 加上 15 是多少然后再乘以 2。”时Agent的思考过程如果verboseTrue会显示如下 进入新的AgentExecutor链... 思考用户想先计算4215再用结果乘以2。我需要使用加法工具和乘法工具。 行动add 行动输入{a: 42, b: 15} 观察57 思考现在我有57需要用它乘以2。 行动multiply 行动输入{x: 57, y: 2} 观察114 思考我得到了最终结果114。 最终答案42加15等于5757乘以2等于114。 链结束。 --- 最终答案 --- 42加15等于5757乘以2等于114。4.4 结果说明通过这个案例我们成功实现了一个独立的MCP服务器提供计算能力完全独立于Agent应用。一个动态集成MCP工具的LangChain AgentAgent在运行时发现并加载了MCP服务器提供的工具无需在代码中硬编码工具逻辑。完整的思考-行动循环Agent能够根据问题自主规划步骤依次调用正确的工具并整合结果。这验证了MCP协议在解耦工具与Agent方面的强大能力。你可以轻松地将calculator_server.py替换为任何其他MCP服务器如文件系统服务器、数据库服务器而mcp_agent.py的主体代码几乎无需改动Agent就能获得新的能力。5. 常见问题与排查思路在实际集成过程中你可能会遇到以下问题。这里提供一份排查清单。问题现象可能原因排查步骤与解决方案Agent无法启动提示MCP连接错误1. MCP服务器脚本路径错误。2. Python环境不一致服务器与客户端环境不同。3. MCP服务器脚本有语法错误或依赖缺失。1. 检查StdioServerParameters中的command和args确保能正确启动服务器。使用绝对路径更可靠。2. 确保Agent和MCP服务器在同一个Python虚拟环境中运行或使用全路径指定python解释器。3. 单独运行MCP服务器脚本(python mcp_servers/calculator_server.py)看是否有错误输出。Agent运行后找不到工具或工具列表为空1. MCP服务器未正确实现list_tools方法。2. 客户端与服务器初始化失败。3. 协议版本不兼容。1. 在MCP服务器的handle_list_tools函数中添加打印日志确认被调用。2. 检查客户端session.initialize()是否成功。可以在create_agent_with_mcp_tools函数中打印mcp_tool_defs。3. 确保使用的mcp库版本在客户端和服务器端尽量一致。工具调用失败返回参数错误1. Agent生成的参数格式与MCP工具定义的inputSchema不匹配。2. LLM未能正确理解工具描述。1. 在Agent的verbose输出中检查“行动输入”的JSON格式。确保参数名如a,b和类型数字正确。2. 优化MCP工具的描述(description)使其对LLM更清晰。例如“Add two numbers”比“Calculate sum”更好。3. 在Agent提示词中强化工具使用格式的要求。异步/同步上下文错误在同步代码中直接调用异步的MCP客户端方法。本文示例使用了asyncio.run在同步函数中调用异步函数这在简单脚本中可行。对于生产级应用如FastAPI服务建议1. 使用完全异步的LangChain Agent执行器如AgentExecutor.ainvoke。2. 确保整个调用链Web框架 - Agent - MCP工具都在异步上下文中。Agent陷入循环或调用错误工具1. 工具描述模糊导致LLM误解。2. ReAct提示词对复杂任务规划能力不足。1. 细化工具描述明确输入输出。例如“Multiply two integers”而非“Do multiplication”。2. 考虑使用更强大的Agent类型如create_openai_functions_agent如果LLM支持Function Calling它对工具调用的支持更精准。3. 在系统提示词中加入工具使用约束如“一次只使用一个工具”。性能问题工具调用慢1. 每次调用都重新启动MCP服务器进程。2. MCP服务器本身处理慢。1.复用连接确保MCP服务器进程在Agent生命周期内保持运行而不是每次调用都新建。本文的async with上下文管理器实现了这一点。2.池化与超时对于高并发场景考虑实现MCP客户端连接池。为工具调用设置合理的超时时间。6. 最佳实践与工程建议将MCP和Skills投入生产环境需要遵循一些工程最佳实践。6.1 MCP Server设计规范单一职责一个MCP Server应专注于提供一类能力。例如calculator-server、filesystem-server、sql-server。避免创建“上帝服务器”。清晰的工具定义name使用动词开头如search_web,read_file。description详细描述工具功能、输入输出。这是LLM理解工具的关键。例如“根据关键词在互联网上搜索最新信息。输入query(字符串搜索关键词)。输出搜索结果的摘要列表。”inputSchema严格定义参数类型和是否必需。使用JSON Schema标准。健壮的错误处理在MCP Server的call_tool函数中捕获所有内部异常并返回结构化的错误信息而不是让进程崩溃。资源管理如果工具涉及资源如数据库连接、文件句柄确保在服务器生命周期内妥善管理。考虑使用连接池。6.2 LangChain Agent集成优化会话管理不要为每个用户请求都创建新的MCP连接。应该在应用启动时初始化并复用MCP客户端会话或使用连接池。工具动态发现与更新实现一个后台任务定期或通过通知检查MCP Server的工具列表是否有更新并动态更新Agent的Tool列表。这可以实现技能的“热插拔”。使用Function Calling如果底层LLM支持Function Calling如GPT-4, Claude优先使用create_openai_functions_agent。它能产生更结构化、更可靠的工具调用参数。技能(Skill)封装不要直接将一堆MCP Tools扔给Agent。将相关的Tools和特定的系统提示词封装成一个Skill类。例如class DataAnalysisSkill: def __init__(self, mcp_session): self.tools self._load_tools(mcp_session) # 加载数据库查询、图表生成等工具 self.system_prompt 你是一个数据分析专家擅长使用SQL查询数据和绘制图表... def get_tools(self): return self.tools def get_prompt(self): return self.system_prompt这样你可以根据用户意图动态地为Agent加载不同的Skill。6.3 安全与权限最小权限原则MCP Server运行时应具有完成其功能所需的最小系统权限。例如文件服务器不应有权限删除根目录。输入验证与净化在MCP Server端对输入参数进行严格验证防止注入攻击如SQL注入、命令注入。访问控制在Agent与MCP Server之间或在一个中心化的MCP路由层实现基于用户/角色的工具访问控制。不是所有用户都能调用所有工具。审计日志记录所有的工具调用包括调用者、参数、结果和时间戳便于追踪和调试。6.4 生产环境部署容器化将每个MCP Server容器化Docker便于独立部署、扩展和管理。服务发现与健康检查当有多个MCP Server时需要一套机制让Agent发现可用的Server。可以结合服务发现工具如Consul或简单的注册中心。监控与告警监控MCP Server的进程状态、资源使用率和错误率。设置告警确保能力可用性。版本管理对MCP Server的接口工具列表和参数schema进行版本管理。Agent客户端应能处理向后兼容或版本协商。通过遵循这些最佳实践你可以构建出一个灵活、健壮、可扩展的基于LangChain Agent和MCP的技能化AI应用系统。这不仅仅是技术集成更是一种面向未来的AI应用架构范式。 
