如果你正在构建基于大模型的智能应用可能会遇到这样的困境模型本身能力强大但让它稳定、可靠地调用外部工具如数据库、API、文件系统却异常困难。传统的Agent开发往往需要为每个工具编写繁琐的适配代码一旦工具更新或新增整个Agent的逻辑就可能需要重构。这不仅是开发效率的问题更关乎系统的可维护性和扩展性。最近一个名为MCPModel Context Protocol的协议开始在AI应用开发社区中受到关注。它并非一个具体的框架而是一个标准化的“通信协议”旨在解决大模型与外部工具Skills之间连接混乱的痛点。当它与成熟的Agent框架如LangChain结合时能带来一种全新的开发范式将工具能力的定义与Agent的调度逻辑彻底解耦。本文要探讨的核心正是如何将LangChain Agent接入MCP协议并利用Skills来构建更强大、更易维护的AI应用。我们将不止步于“是什么”而是深入剖析“为什么”MCPSkills的组合值得关注并通过一个从零到一的完整实践展示它如何“全方位提升工作效率”。你会发现这不仅仅是多了一个工具调用选项更是对AI应用架构的一次重要升级。1. 这篇文章真正要解决的问题在深入代码之前我们必须先厘清一个根本问题为什么在已有LangChain Tools的情况下我们还需要关注MCP和SkillsLangChain的Tools机制已经非常成熟它允许开发者将函数包装成工具供Agent调用。然而随着项目复杂度提升Tools模式暴露出几个核心痛点紧耦合与高维护成本Tools的逻辑如API调用、数据处理通常直接写在Agent的应用代码中。当工具接口变更、新增工具或工具逻辑需要优化时你必须修改并重新部署整个Agent应用。能力发现与管理的缺失Agent在运行时无法动态感知到有哪些可用的Tools除非你硬编码进去。在微服务或分布式环境下管理成百上千个分散的工具变得极其困难。开发协作的壁垒工具开发者例如负责数据库的团队和Agent应用开发者负责业务逻辑的团队需要紧密协作。工具的任何改动都可能成为跨团队沟通和联调的负担。安全与权限控制的复杂性如何控制Agent只能访问特定的工具或数据在紧耦合架构下这通常需要在每个工具函数内部进行重复的权限校验。MCP协议的核心价值正是为了解决上述痛点。它定义了一套标准让工具在MCP语境下称为“Skills”或“Servers”能够以独立进程或服务的形式存在并通过统一的协议向“客户端”如你的LangChain Agent宣告自己的能力。这带来了几个关键转变解耦Skills独立开发、部署、版本化管理。Agent无需关心Skill的内部实现只需知道如何调用它。动态发现Agent可以在启动时或运行时通过MCP协议从多个Skill Server发现可用的工具列表。标准化通信所有Skill都通过相同的JSON-RPC over stdio/HTTP协议与Agent通信降低了集成复杂度。集中化管理可以有一个中心化的Skill仓库或注册中心方便管理和复用。因此本文要解决的不是“如何让Agent多调用一个工具”的简单问题而是“如何构建一个松耦合、易扩展、好维护的AI应用架构”的系统性问题。如果你正在开发或维护一个需要集成多种能力如数据分析、文档处理、系统操作的复杂Agent那么MCPSkills的实践将为你提供一条更清晰的路径。2. 基础概念与核心原理在开始实践前我们需要准确理解几个关键概念及其之间的关系。2.1 LangChain Agent智能调度中枢LangChain Agent是一个基于大语言模型LLM的推理引擎。它接收用户输入或系统指令通过LLM进行思考ReAct模式等决定下一步该执行哪个工具Tool获取工具执行结果后再进行下一轮思考直至完成任务或达到停止条件。它是整个智能工作流的“大脑”和“调度器”。2.2 MCP (Model Context Protocol)工具连接的标准“插座”你可以把MCP想象成电子设备中的“USB协议”或“蓝牙协议”。它本身不提供任何具体功能如读文件、查数据库但它定义了一套所有工具Skill和设备Client都必须遵守的通信规范。核心思想标准化大模型或Agent与外部上下文、工具之间的交互方式。核心组件MCP Server (Skill Provider)提供具体能力的服务端。例如一个“文件系统Skill Server”可以提供read_file,write_file等能力。它启动后会按照MCP协议向客户端宣告自己有哪些“工具Tools”可用。MCP Client (Skill Consumer)消费这些能力的客户端。我们的LangChain Agent就需要扮演这个角色。它会连接一个或多个MCP Server获取工具列表并在需要时按照协议调用它们。通信方式通常采用JSON-RPCoverstdio标准输入输出或HTTP。Stdio模式非常适合将Skill作为本地子进程运行简单高效HTTP模式则适用于远程服务。2.3 Skills即插即用的标准化工具在MCP体系下Skill就是实现了MCP Server协议的具体工具实例。一个Skill Server可以提供多个相关的工具Tools。例如filesystemSkill提供read_file,list_files,search_files等工具。sqlSkill提供execute_query,list_tables等工具。githubSkill提供search_issues,create_pr等工具。Skills与LangChain Tools的关系它们的目标一致——为Agent提供外部能力。但Skills通过MCP协议实现了标准化和独立化。在LangChain中我们可以通过一个适配器将远程MCP Skill“转换”成本地LangChain Tool从而被Agent无缝调用。2.4 核心原理与工作流程整个系统的工作流程可以概括为以下几步启动Skill Servers启动一个或多个独立的MCP Skill服务进程如filesystem server。LangChain Agent连接MCP在LangChain应用中通过MCP Client库连接到这些Server。发现可用工具MCP Client从所有连接的Server获取它们提供的工具列表及其描述包括名称、参数schema等。封装为LangChain Tools将这些远程工具描述动态创建为LangChain Tool对象。Agent调度执行用户提问 - AgentLLM根据问题选择最合适的Tool - 通过MCP协议调用对应的Skill Server - 获取结果 - 继续思考或输出最终答案。这个架构的关键优势在于第1步Skill开发/部署和第2-5步Agent应用开发可以完全独立进行。只要协议一致它们就能协同工作。3. 环境准备与前置条件为了完成本次实践你需要准备以下环境。我们将以一个相对简单的场景为例构建一个能读取本地文件并回答问题的Agent。操作系统本文示例基于 macOS/Linux 环境Windows用户建议使用 WSL2 以获得最佳体验。Python环境请确保已安装 Python 3.10 或更高版本。推荐使用conda或venv创建独立的虚拟环境。# 创建并激活虚拟环境以venv为例 python -m venv langchain-mcp-env source langchain-mcp-env/bin/activate # Linux/macOS # langchain-mcp-env\Scripts\activate # Windows核心依赖安装我们将安装 LangChain 相关库以及官方提供的 MCP 客户端和示例 Servers。# 升级pip pip install --upgrade pip # 安装LangChain及其OpenAI集成我们使用OpenAI的模型作为Agent大脑 pip install langchain langchain-openai # 安装MCP相关的核心库 # mcp 是官方Python SDK用于构建Client和Server # langchain-mcp-adapters 是LangChain与MCP的桥接库 pip install mcp langchain-mcp-adapters # 安装一些可能用到的工具库以及示例MCP Servers # 这里安装 filesystem 和 sqlite 两个示例server pip install mcp[cli] mcp-server-filesystem mcp-server-sqlite模型API密钥本文使用 OpenAI GPT-4 或 GPT-3.5-turbo 作为Agent的LLM你需要准备一个有效的OPENAI_API_KEY。如果你没有可以考虑使用其他LangChain支持的模型如 Anthropic Claude、本地部署的Ollama等但后续代码需要相应调整。# 在终端中设置环境变量临时 export OPENAI_API_KEYyour-api-key-here # 或者写入 ~/.bashrc 或 ~/.zshrc 永久生效验证基础环境python -c import langchain, mcp; print(LangChain版本:, langchain.__version__); print(MCP版本:, mcp.__version__)4. 核心流程拆解从启动Skill到构建Agent让我们将“构建一个接入MCP Skills的LangChain Agent”这个目标拆解为五个清晰的步骤。4.1 第一步启动独立的MCP Skill ServerSkill Server是能力的提供者。我们首先让一个提供文件操作能力的Server在后台运行起来。MCP官方和社区提供了许多现成的Server。启动一个文件系统Skill Server 这个Server会暴露读取文件、列出目录等工具。我们通过Python的subprocess模块或直接在另一个终端窗口启动它。# 在新的终端窗口执行或使用 在后台运行 mcp run filesystem # 默认情况下它会将当前工作目录(.)作为根目录并监听 stdio。运行后这个进程会阻塞等待通过标准输入stdin接收MCP协议指令。我们的LangChain应用将作为客户端连接到这个进程。4.2 第二步在LangChain中连接MCP Server并发现Tools这是关键的一步。我们需要在Python代码中创建一个MCP客户端连接到上一步启动的Server进程然后获取它提供的所有工具。# 文件connect_mcp.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client # 定义要连接的MCP Server这里以filesystem为例 server_params StdioServerParameters( commandpython, # 解释器 args[-m, mcp_server_filesystem, .], # 模块和参数以当前目录为根 # 注意这里直接调用模块与上面mcp run命令等效 ) async def list_available_tools(): 连接MCP Server并列出所有可用工具 async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # 初始化连接 await session.initialize() # 获取Server提供的工具列表 tools await session.list_tools() print(f连接到MCP Server成功) print(f发现 {len(tools.tools)} 个工具) for tool in tools.tools: print(f - 名称: {tool.name}) print(f 描述: {tool.description}) if tool.inputSchema: print(f 输入参数: {tool.inputSchema}) print() return tools.tools # 运行异步函数 if __name__ __main__: tools asyncio.run(list_available_tools())运行这段代码你应该能看到类似以下的输出这证明你的LangChain应用已经成功与MCP Server握手并发现了它提供的工具如read_file,list_files等。连接到MCP Server成功 发现 2 个工具 - 名称: read_file 描述: Read the contents of a file at the given path. 输入参数: {type: object, properties: {path: {type: string, description: The path of the file to read.}}, required: [path]} - 名称: list_files 描述: List files and directories in a given directory path. 输入参数: {type: object, properties: {path: {type: string, description: The directory path to list.}}, required: [path]}4.3 第三步将MCP Tools适配为LangChain Tools发现工具只是第一步我们需要将这些符合MCP协议的工具描述转换成LangChain Agent能够识别和调用的Tool对象。langchain-mcp-adapters库提供了这个桥梁。# 文件create_langchain_tools.py import asyncio from langchain_mcp_adapters.tools import MCPToolkit from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def create_mcp_toolkit(): 创建并返回一个包含所有MCP Tools的LangChain Toolkit server_params StdioServerParameters( commandpython, args[-m, mcp_server_filesystem, .], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 关键步骤使用 MCPToolkit 将 session 转换为 LangChain 工具集 toolkit await MCPToolkit.from_session(session).ainit() # 获取LangChain格式的Tool列表 tools toolkit.get_tools() print(f成功创建了 {len(tools)} 个LangChain Tools。) for tool in tools: print(fTool: {tool.name} - {tool.description}) return tools if __name__ __main__: tools asyncio.run(create_mcp_toolkit())MCPToolkit是一个强大的适配器它处理了协议转换、参数封装和异步调用等底层细节为我们提供了开箱即用的Tool对象列表。4.4 第四步构建并运行LangChain Agent现在我们有了标准的LangChain Tools构建Agent就与平常无异了。我们使用OpenAI的LLM和ReAct代理框架。# 文件run_agent_with_mcp.py import asyncio import os from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import PromptTemplate # 导入上一步编写的工具创建函数 from create_langchain_tools import create_mcp_toolkit async def main(): # 1. 创建LLM确保已设置OPENAI_API_KEY环境变量 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 使用gpt-4o-mini成本更低效果足够 # 2. 获取MCP Tools tools await create_mcp_toolkit() # 3. 定义Agent的提示词模板 prompt PromptTemplate.from_template( 你是一个有帮助的AI助手可以访问文件系统来回答用户问题。 你可以使用以下工具 {tools} 请严格按照以下格式回答 问题用户输入的问题 思考你需要思考如何一步步解决问题 行动要使用的工具名称 行动输入工具的输入必须是有效的JSON格式 观察工具返回的结果 ... (这个思考/行动/观察循环可以重复多次) 最终答案根据观察得出的最终答案 如果用户的问题不需要使用工具或者工具无法解决请直接给出答案。 开始 问题{input} 思考{agent_scratchpad} ) # 4. 创建ReAct Agent agent create_react_agent(llm, tools, prompt) # 5. 创建Agent执行器 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 开启详细日志方便观察Agent的思考过程 handle_parsing_errorsTrue, # 处理解析错误 max_iterations5, # 限制最大迭代次数防止死循环 ) # 6. 运行Agent测试一个需要文件操作的问题 # 首先确保当前目录下有一个测试文件例如 test.txt内容为“Hello, MCP!” test_question 请读取当前目录下名为test.txt的文件并告诉我它的内容是什么 print(f用户问题: {test_question}) print(- * 50) try: result await agent_executor.ainvoke({input: test_question}) print(\n *50) print(f最终答案: {result[output]}) except Exception as e: print(f执行过程中出现错误: {e}) if __name__ __main__: asyncio.run(main())4.5 第五步扩展 - 连接多个Skill ServersMCP的真正威力在于可以轻松集成多个独立的Skill。假设我们还有一个提供SQLite查询的Skill Server。启动SQLite Skill Server(需要先有一个SQLite数据库文件例如example.db)# 在另一个终端或后台进程 mcp run sqlite --db-path ./example.db在LangChain中连接多个Servers# 文件connect_multiple_servers.py (部分关键代码) import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import MCPToolkit async def create_tools_from_multiple_servers(): all_tools [] # 定义多个Server的参数 servers [ StdioServerParameters(commandpython, args[-m, mcp_server_filesystem, .]), StdioServerParameters(commandpython, args[-m, mcp_server_sqlite, --db-path, ./example.db]), ] for server_params in servers: try: async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() toolkit await MCPToolkit.from_session(session).ainit() server_tools toolkit.get_tools() all_tools.extend(server_tools) print(f从Server {server_params.args[-1]} 加载了 {len(server_tools)} 个工具。) except Exception as e: print(f连接Server {server_params.args} 失败: {e}) continue print(f总计加载了 {len(all_tools)} 个工具。) return all_tools # 后续构建Agent的代码与单Server类似只是tools来源变成了all_tools这样你的Agent就同时具备了文件操作和数据库查询的能力而这两部分能力是由两个完全独立、可单独维护的服务提供的。5. 完整示例构建一个多功能文档查询Agent让我们整合以上步骤构建一个更实用的示例一个能同时查询文件内容和数据库信息的智能助手。项目结构multi_skill_agent/ ├── data/ │ ├── reports/ # 存放一些文本报告 │ │ └── q1_report.txt │ └── example.db # SQLite数据库文件 ├── skills/ # (可选) 放置自定义Skill的代码 ├── main.py # 主程序入口 └── requirements.txtrequirements.txt:langchain0.1.0 langchain-openai0.0.5 langchain-mcp-adapters0.1.0 mcp0.1.0 mcp-server-filesystem mcp-server-sqlitemain.py - 完整实现:import asyncio import sys import os from pathlib import Path from typing import List from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import PromptTemplate from langchain_core.tools import BaseTool from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import MCPToolkit class MultiSkillAgent: def __init__(self, openai_api_key: str None, model: str gpt-4o-mini): 初始化Agent设置LLM和工具列表 if openai_api_key: os.environ[OPENAI_API_KEY] openai_api_key self.llm ChatOpenAI(modelmodel, temperature0) self.tools: List[BaseTool] [] self.agent_executor: AgentExecutor None async def connect_skill_server(self, server_params: StdioServerParameters) - List[BaseTool]: 连接一个MCP Skill Server并返回其工具列表 server_tools [] try: print(f正在连接MCP Server: {server_params.command} { .join(server_params.args)}) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() toolkit await MCPToolkit.from_session(session).ainit() server_tools toolkit.get_tools() print(f 成功加载 {len(server_tools)} 个工具。) except Exception as e: print(f 连接失败: {e}) return server_tools async def initialize(self, skill_configs: list): 初始化Agent连接所有配置的Skill Servers print(开始初始化Multi-Skill Agent...) for config in skill_configs: tools await self.connect_skill_server(config) self.tools.extend(tools) if not self.tools: print(警告未加载到任何工具Agent将无法执行操作。) return print(f总计加载了 {len(self.tools)} 个工具。) # 创建Agent提示词可根据工具特性优化 prompt self._create_agent_prompt() # 创建ReAct Agent agent create_react_agent(self.llm, self.tools, prompt) # 创建执行器 self.agent_executor AgentExecutor( agentagent, toolsself.tools, verboseTrue, handle_parsing_errorsTrue, max_iterations8, early_stopping_methodgenerate, ) print(Agent初始化完成) def _create_agent_prompt(self) - PromptTemplate: 创建Agent的提示词模板 tool_descriptions \n.join([f- {tool.name}: {tool.description} for tool in self.tools]) template f 你是一个强大的AI助手可以调用多种工具来帮助用户解决问题。 你可以使用的工具如下 {tool_descriptions} 请严格按照以下格式回应 问题用户的问题 思考分析问题决定是否需要以及使用哪个工具 行动要使用的工具名称必须是上述列表中的一个 行动输入工具的输入必须是有效的JSON对象 观察工具返回的结果 ...这个思考/行动/观察循环可以重复多次但请尽量简洁 最终答案根据所有观察给出清晰、完整的最终答案 如果问题不需要工具或工具无法解决请直接推理并给出最终答案。 请确保“行动输入”是严格的JSON例如对于需要path参数的工具输入应为{{path: /some/file.txt}} 现在开始处理第一个问题。 问题{{input}} {{agent_scratchpad}} return PromptTemplate.from_template(template) async def run(self, query: str) - str: 运行Agent处理查询 if not self.agent_executor: return Agent未正确初始化。 print(f\n{*60}) print(f处理查询: {query}) print(*60) try: result await self.agent_executor.ainvoke({input: query}) return result.get(output, 未获得有效输出。) except Exception as e: return f执行过程中出现错误: {e} async def main(): 主函数配置并运行Agent # 1. 定义要连接的Skill Servers # 假设项目根目录下有一个data文件夹里面有报告和数据库 base_dir Path(__file__).parent data_dir base_dir / data skill_configs [ # 文件系统Skill根目录设置为data/reports StdioServerParameters( commandsys.executable, # 使用当前Python解释器 args[-m, mcp_server_filesystem, str(data_dir / reports)] ), # SQLite Skill连接到example.db StdioServerParameters( commandsys.executable, args[-m, mcp_server_sqlite, --db-path, str(data_dir / example.db)] ), ] # 2. 创建并初始化Agent agent MultiSkillAgent() await agent.initialize(skill_configs) # 3. 准备测试数据和查询 # 首先确保数据存在这里模拟创建 reports_dir data_dir / reports reports_dir.mkdir(parentsTrue, exist_okTrue) (reports_dir / q1_report.txt).write_text(2024年第一季度销售额为120万元主要增长来自华东地区。) import sqlite3 db_path data_dir / example.db if not db_path.exists(): conn sqlite3.connect(db_path) cursor conn.cursor() cursor.execute(CREATE TABLE IF NOT EXISTS products (id INTEGER PRIMARY KEY, name TEXT, price REAL)) cursor.execute(INSERT INTO products (name, price) VALUES (Laptop, 999.99), (Mouse, 29.99)) conn.commit() conn.close() print(已创建示例数据库和表。) # 4. 运行测试查询 test_queries [ 读取reports文件夹下的q1_report.txt文件总结一下销售额情况。, 查询数据库里所有产品的名称和价格。, 综合一下我们第一季度销售额不错数据库里最贵的产品是什么它的价格占季度销售额的比例大概是多少, ] for query in test_queries: answer await agent.run(query) print(f\n 最终答案:\n{answer}\n) print(-*60) if __name__ __main__: asyncio.run(main())6. 运行结果与效果验证运行上述main.py脚本你将看到类似以下的输出清晰地展示了Agent的思考链Chain of Thought和工具调用过程开始初始化Multi-Skill Agent... 正在连接MCP Server: /usr/local/bin/python -m mcp_server_filesystem /path/to/data/reports 成功加载 2 个工具。 正在连接MCP Server: /usr/local/bin/python -m mcp_server_sqlite --db-path /path/to/data/example.db 成功加载 2 个工具。 总计加载了 4 个工具。 Agent初始化完成 处理查询: 读取reports文件夹下的q1_report.txt文件总结一下销售额情况。 进入新的Agent执行链... 思考用户想要读取一个文件并总结销售额。我需要先找到并读取这个文件。 行动list_files 行动输入{path: .} 观察[., q1_report.txt] 思考我看到了 q1_report.txt 文件。现在需要读取它的内容。 行动read_file 行动输入{path: q1_report.txt} 观察2024年第一季度销售额为120万元主要增长来自华东地区。 思考我已经读取了文件内容。现在需要总结销售额情况。内容显示第一季度销售额120万元主要增长来自华东地区。我可以直接给出总结。 最终答案根据文件内容2024年第一季度销售额为120万元主要增长动力来自华东地区。 最终答案: 根据文件内容2024年第一季度销售额为120万元主要增长动力来自华东地区。 ------------------------------------------------------------ 处理查询: 查询数据库里所有产品的名称和价格。 进入新的Agent执行链... 思考用户想查询数据库。我需要使用SQLite工具。有一个sql_query工具可以用来执行SQL查询。 行动sql_query 行动输入{query: SELECT name, price FROM products} 观察[{name: Laptop, price: 999.99}, {name: Mouse, price: 29.99}] 思考查询成功返回了产品列表。我可以直接呈现这个结果。 最终答案数据库中共有2个产品 1. Laptop - 价格: 999.99 2. Mouse - 价格: 29.99 最终答案: 数据库中共有2个产品 1. Laptop - 价格: 999.99 2. Mouse - 价格: 29.99 ------------------------------------------------------------效果验证要点工具发现Agent成功从两个独立的Server加载了4个工具2个文件操作2个数据库操作。正确路由对于文件查询Agent正确选择了list_files和read_file工具对于数据库查询正确选择了sql_query工具。参数构造Agent能根据工具的描述自动构造正确的JSON格式输入如{path: q1_report.txt}{query: SELECT ...}。复杂任务处理在第三个综合查询中Agent展示了多步推理和工具组合的能力先读取文件获取销售额再查询数据库获取产品价格最后进行简单的数学计算和总结。解耦验证你可以独立更新mcp_server_sqlite的版本或者替换成另一个提供相同工具sql_query但内部实现不同的Server而main.py中的Agent代码完全不需要修改。7. 常见问题与排查思路在实际集成MCP与LangChain的过程中你可能会遇到以下典型问题问题现象可能原因排查方式解决方案连接MCP Server失败1. Server命令路径错误。2. 所需Python模块未安装。3. Server进程已崩溃。1. 在终端手动运行mcp run filesystem看是否报错。2. 检查pip list | grep mcp-server。3. 查看Python异常堆栈信息。1. 确保command和args参数正确。2. 使用pip install安装对应的server包。3. 使用try...except包裹连接代码并添加重试逻辑。Agent找不到或错误调用工具1. 工具名称不匹配。2. 输入参数格式错误。3. LLM的提示词未清晰描述工具。1. 在初始化时打印所有加载的工具名称和描述。2. 开启Agent的verboseTrue模式观察其“行动输入”。3. 检查提示词模板中的{tools}是否被正确替换。1. 确保工具描述清晰。可在提示词中强化工具名称。2. 使用handle_parsing_errorsTrue让Agent有机会纠正格式错误。3. 考虑使用更结构化的输出解析器如OpenAI Functions。Agent陷入循环或迭代次数过多1. 工具返回结果未满足Agent预期。2. 任务过于复杂或模糊。3.max_iterations设置过高。观察verbose日志看思考步骤是否在重复或无效循环。1. 优化工具返回的信息使其更直接、结构化。2. 在提示词中明确任务边界和停止条件。3. 合理设置max_iterations如5-10并使用early_stopping_method。性能问题响应慢1. 每次调用都新建连接。2. LLM本身响应慢。3. Skill Server处理耗时。1. 检查是否在每次工具调用时都创建了新的Session。2. 测试直接调用LLM的延迟。3. 单独测试Skill Server的响应时间。1.复用Session确保ClientSession在整个Agent生命周期内保持连接而不是每次调用都新建。2. 考虑使用更快的LLM或模型。3. 对耗时Skill做异步优化或缓存。权限或安全错误1. 文件系统Skill访问了无权目录。2. SQL查询包含危险操作。1. 检查Skill Server启动时设置的根目录。2. 审查Agent生成的SQL语句。1. 在启动Skill Server时严格限制其访问范围如只读模式、指定目录。2. 对于数据库Skill可使用只有SELECT权限的账户或在Server层进行SQL过滤。8. 最佳实践与工程建议将MCP架构投入生产环境或复杂项目时遵循以下最佳实践可以避免很多坑8.1 Skill设计与开发单一职责每个Skill Server应专注于一个特定的领域如文件、数据库、API网关、搜索引擎。这符合微服务的设计哲学便于独立开发、测试和部署。清晰的工具描述在实现Skill时为每个工具提供准确、详细的description和inputSchema。LLM依赖这些描述来选择工具模糊的描述会导致调用错误。健壮的错误处理Skill内部应该有完善的错误处理机制并将错误信息以结构化的方式返回给客户端帮助Agent理解失败原因例如“文件未找到” vs “权限被拒绝”。版本化与兼容性对Skill进行版本管理。当工具接口需要变更时考虑通过新增工具如search_v2而非修改现有工具来保持向后兼容。8.2 Agent应用开发Session管理这是性能关键。务必在Agent应用启动时创建并维护与各个Skill Server的持久化Session连接池避免为每次工具调用建立新连接的开销。提示词工程精心设计Agent的提示词。除了列出工具还应说明工具的适用场景和调用约束。例如“sql_query工具仅用于查询不支持INSERT/UPDATE”。超时与重试为工具调用设置合理的超时时间并实现重试机制特别是对于网络不稳定的远程HTTP Skill。可观测性记录详细的日志包括Agent的思考过程、工具调用请求和响应。这对于调试复杂问题和优化提示词至关重要。8.3 安全与权限最小权限原则以最低必要权限运行Skill Server。文件系统Skill应限制在特定目录数据库Skill应使用只有查询权限的账号。输入验证与净化在Skill Server端对输入参数进行严格验证。防止路径遍历../../../etc/passwd、SQL注入等攻击。网络隔离如果Skill Server以HTTP模式运行在网络上务必将其置于内网并通过防火墙或API网关控制访问。避免将敏感能力的Server暴露在公网。审计日志记录所有工具调用的审计日志包括调用者、参数、结果和时间戳便于安全审查和问题追溯。8.4 部署与运维容器化将每个Skill Server和Agent应用分别容器化Docker便于统一部署、资源隔离和水平扩展。服务发现在生产环境中Skill Server的地址可能是动态的。可以考虑使用简单的服务发现机制如环境变量、配置中心、Consul等让Agent能动态发现可用的Skill。健康检查为Skill Server实现健康检查端点Agent应用或编排系统可以定期检查其状态并进行故障转移或重启。监控与告警监控Skill Server的资源使用情况CPU、内存、请求延迟和错误率。设置告警以便在服务异常时及时介入。通过遵循这些实践你可以构建出一个既灵活强大又稳定可靠、易于维护的基于MCP和LangChain的AI应用系统。9. 总结与后续学习方向本文深入探讨了将LangChain Agent与MCP协议及Skills集成的技术原理与实践方法。我们从一个具体的开发痛点出发——工具与Agent的紧耦合——介绍了MCP如何通过标准化协议实现解耦并一步步演示了如何启动Skill Server、在LangChain中发现并封装工具、最终构建一个能协同使用多种能力的智能Agent。关键收获架构升级MCP不仅仅是一个工具调用库它更是一种架构模式。它促使我们将AI应用中的“能力提供者”Skills和“能力调度者”Agent分离让系统更模块化、更易扩展。开发效率一旦协议标准化团队可以并行开发Skills和Agent应用。前端Agent提示词、UI和后端各种能力服务的开发者可以更独立地工作。生态潜力一个开放的MCP Skill生态正在形成。未来我们或许可以像安装插件一样为我们的Agent轻松添加图像识别、代码执行、邮件发送等成千上万种能力而无需修改核心代码。后续你可以深入探索的方向开发自定义Skill本文使用了官方示例Server。尝试用Python或Node.js等MCP支持的语言开发一个你自己的Skill例如调用一个内部API、操作一个特定的云服务。探索更多现成Skill在MCP的官方仓库和社区中寻找更多有趣的Skill如github、brave-search、notion等极大地扩展Agent的能力边界。深入LangGraph对于需要复杂工作流、状态管理和循环的AgentLangChain的LangGraph框架是更强大的选择。研究如何将MCP Tools集成到LangGraph的图中。性能优化研究如何对频繁调用的Skill结果进行缓存如何并行调用多个独立工具以及如何对Agent的思考过程进行优化以减少不必要的工具调用成本考量。生产化部署将本文的示例代码改造为真正的生产服务考虑如何做版本管理、灰度发布、监控告警和自动扩缩容。技术的价值在于解决真实问题。MCP与LangChain的结合为构建下一代模块化、可插拔的AI智能体应用提供了坚实的技术基础。建议你将本文的示例代码作为起点亲手实践并改造将其应用到你的具体业务场景中真正体验它所带来的“全方位工作效率提升”。 LangChain Agent集成MCP协议：构建模块化AI应用架构实践
如果你正在构建基于大模型的智能应用可能会遇到这样的困境模型本身能力强大但让它稳定、可靠地调用外部工具如数据库、API、文件系统却异常困难。传统的Agent开发往往需要为每个工具编写繁琐的适配代码一旦工具更新或新增整个Agent的逻辑就可能需要重构。这不仅是开发效率的问题更关乎系统的可维护性和扩展性。最近一个名为MCPModel Context Protocol的协议开始在AI应用开发社区中受到关注。它并非一个具体的框架而是一个标准化的“通信协议”旨在解决大模型与外部工具Skills之间连接混乱的痛点。当它与成熟的Agent框架如LangChain结合时能带来一种全新的开发范式将工具能力的定义与Agent的调度逻辑彻底解耦。本文要探讨的核心正是如何将LangChain Agent接入MCP协议并利用Skills来构建更强大、更易维护的AI应用。我们将不止步于“是什么”而是深入剖析“为什么”MCPSkills的组合值得关注并通过一个从零到一的完整实践展示它如何“全方位提升工作效率”。你会发现这不仅仅是多了一个工具调用选项更是对AI应用架构的一次重要升级。1. 这篇文章真正要解决的问题在深入代码之前我们必须先厘清一个根本问题为什么在已有LangChain Tools的情况下我们还需要关注MCP和SkillsLangChain的Tools机制已经非常成熟它允许开发者将函数包装成工具供Agent调用。然而随着项目复杂度提升Tools模式暴露出几个核心痛点紧耦合与高维护成本Tools的逻辑如API调用、数据处理通常直接写在Agent的应用代码中。当工具接口变更、新增工具或工具逻辑需要优化时你必须修改并重新部署整个Agent应用。能力发现与管理的缺失Agent在运行时无法动态感知到有哪些可用的Tools除非你硬编码进去。在微服务或分布式环境下管理成百上千个分散的工具变得极其困难。开发协作的壁垒工具开发者例如负责数据库的团队和Agent应用开发者负责业务逻辑的团队需要紧密协作。工具的任何改动都可能成为跨团队沟通和联调的负担。安全与权限控制的复杂性如何控制Agent只能访问特定的工具或数据在紧耦合架构下这通常需要在每个工具函数内部进行重复的权限校验。MCP协议的核心价值正是为了解决上述痛点。它定义了一套标准让工具在MCP语境下称为“Skills”或“Servers”能够以独立进程或服务的形式存在并通过统一的协议向“客户端”如你的LangChain Agent宣告自己的能力。这带来了几个关键转变解耦Skills独立开发、部署、版本化管理。Agent无需关心Skill的内部实现只需知道如何调用它。动态发现Agent可以在启动时或运行时通过MCP协议从多个Skill Server发现可用的工具列表。标准化通信所有Skill都通过相同的JSON-RPC over stdio/HTTP协议与Agent通信降低了集成复杂度。集中化管理可以有一个中心化的Skill仓库或注册中心方便管理和复用。因此本文要解决的不是“如何让Agent多调用一个工具”的简单问题而是“如何构建一个松耦合、易扩展、好维护的AI应用架构”的系统性问题。如果你正在开发或维护一个需要集成多种能力如数据分析、文档处理、系统操作的复杂Agent那么MCPSkills的实践将为你提供一条更清晰的路径。2. 基础概念与核心原理在开始实践前我们需要准确理解几个关键概念及其之间的关系。2.1 LangChain Agent智能调度中枢LangChain Agent是一个基于大语言模型LLM的推理引擎。它接收用户输入或系统指令通过LLM进行思考ReAct模式等决定下一步该执行哪个工具Tool获取工具执行结果后再进行下一轮思考直至完成任务或达到停止条件。它是整个智能工作流的“大脑”和“调度器”。2.2 MCP (Model Context Protocol)工具连接的标准“插座”你可以把MCP想象成电子设备中的“USB协议”或“蓝牙协议”。它本身不提供任何具体功能如读文件、查数据库但它定义了一套所有工具Skill和设备Client都必须遵守的通信规范。核心思想标准化大模型或Agent与外部上下文、工具之间的交互方式。核心组件MCP Server (Skill Provider)提供具体能力的服务端。例如一个“文件系统Skill Server”可以提供read_file,write_file等能力。它启动后会按照MCP协议向客户端宣告自己有哪些“工具Tools”可用。MCP Client (Skill Consumer)消费这些能力的客户端。我们的LangChain Agent就需要扮演这个角色。它会连接一个或多个MCP Server获取工具列表并在需要时按照协议调用它们。通信方式通常采用JSON-RPCoverstdio标准输入输出或HTTP。Stdio模式非常适合将Skill作为本地子进程运行简单高效HTTP模式则适用于远程服务。2.3 Skills即插即用的标准化工具在MCP体系下Skill就是实现了MCP Server协议的具体工具实例。一个Skill Server可以提供多个相关的工具Tools。例如filesystemSkill提供read_file,list_files,search_files等工具。sqlSkill提供execute_query,list_tables等工具。githubSkill提供search_issues,create_pr等工具。Skills与LangChain Tools的关系它们的目标一致——为Agent提供外部能力。但Skills通过MCP协议实现了标准化和独立化。在LangChain中我们可以通过一个适配器将远程MCP Skill“转换”成本地LangChain Tool从而被Agent无缝调用。2.4 核心原理与工作流程整个系统的工作流程可以概括为以下几步启动Skill Servers启动一个或多个独立的MCP Skill服务进程如filesystem server。LangChain Agent连接MCP在LangChain应用中通过MCP Client库连接到这些Server。发现可用工具MCP Client从所有连接的Server获取它们提供的工具列表及其描述包括名称、参数schema等。封装为LangChain Tools将这些远程工具描述动态创建为LangChain Tool对象。Agent调度执行用户提问 - AgentLLM根据问题选择最合适的Tool - 通过MCP协议调用对应的Skill Server - 获取结果 - 继续思考或输出最终答案。这个架构的关键优势在于第1步Skill开发/部署和第2-5步Agent应用开发可以完全独立进行。只要协议一致它们就能协同工作。3. 环境准备与前置条件为了完成本次实践你需要准备以下环境。我们将以一个相对简单的场景为例构建一个能读取本地文件并回答问题的Agent。操作系统本文示例基于 macOS/Linux 环境Windows用户建议使用 WSL2 以获得最佳体验。Python环境请确保已安装 Python 3.10 或更高版本。推荐使用conda或venv创建独立的虚拟环境。# 创建并激活虚拟环境以venv为例 python -m venv langchain-mcp-env source langchain-mcp-env/bin/activate # Linux/macOS # langchain-mcp-env\Scripts\activate # Windows核心依赖安装我们将安装 LangChain 相关库以及官方提供的 MCP 客户端和示例 Servers。# 升级pip pip install --upgrade pip # 安装LangChain及其OpenAI集成我们使用OpenAI的模型作为Agent大脑 pip install langchain langchain-openai # 安装MCP相关的核心库 # mcp 是官方Python SDK用于构建Client和Server # langchain-mcp-adapters 是LangChain与MCP的桥接库 pip install mcp langchain-mcp-adapters # 安装一些可能用到的工具库以及示例MCP Servers # 这里安装 filesystem 和 sqlite 两个示例server pip install mcp[cli] mcp-server-filesystem mcp-server-sqlite模型API密钥本文使用 OpenAI GPT-4 或 GPT-3.5-turbo 作为Agent的LLM你需要准备一个有效的OPENAI_API_KEY。如果你没有可以考虑使用其他LangChain支持的模型如 Anthropic Claude、本地部署的Ollama等但后续代码需要相应调整。# 在终端中设置环境变量临时 export OPENAI_API_KEYyour-api-key-here # 或者写入 ~/.bashrc 或 ~/.zshrc 永久生效验证基础环境python -c import langchain, mcp; print(LangChain版本:, langchain.__version__); print(MCP版本:, mcp.__version__)4. 核心流程拆解从启动Skill到构建Agent让我们将“构建一个接入MCP Skills的LangChain Agent”这个目标拆解为五个清晰的步骤。4.1 第一步启动独立的MCP Skill ServerSkill Server是能力的提供者。我们首先让一个提供文件操作能力的Server在后台运行起来。MCP官方和社区提供了许多现成的Server。启动一个文件系统Skill Server 这个Server会暴露读取文件、列出目录等工具。我们通过Python的subprocess模块或直接在另一个终端窗口启动它。# 在新的终端窗口执行或使用 在后台运行 mcp run filesystem # 默认情况下它会将当前工作目录(.)作为根目录并监听 stdio。运行后这个进程会阻塞等待通过标准输入stdin接收MCP协议指令。我们的LangChain应用将作为客户端连接到这个进程。4.2 第二步在LangChain中连接MCP Server并发现Tools这是关键的一步。我们需要在Python代码中创建一个MCP客户端连接到上一步启动的Server进程然后获取它提供的所有工具。# 文件connect_mcp.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client # 定义要连接的MCP Server这里以filesystem为例 server_params StdioServerParameters( commandpython, # 解释器 args[-m, mcp_server_filesystem, .], # 模块和参数以当前目录为根 # 注意这里直接调用模块与上面mcp run命令等效 ) async def list_available_tools(): 连接MCP Server并列出所有可用工具 async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # 初始化连接 await session.initialize() # 获取Server提供的工具列表 tools await session.list_tools() print(f连接到MCP Server成功) print(f发现 {len(tools.tools)} 个工具) for tool in tools.tools: print(f - 名称: {tool.name}) print(f 描述: {tool.description}) if tool.inputSchema: print(f 输入参数: {tool.inputSchema}) print() return tools.tools # 运行异步函数 if __name__ __main__: tools asyncio.run(list_available_tools())运行这段代码你应该能看到类似以下的输出这证明你的LangChain应用已经成功与MCP Server握手并发现了它提供的工具如read_file,list_files等。连接到MCP Server成功 发现 2 个工具 - 名称: read_file 描述: Read the contents of a file at the given path. 输入参数: {type: object, properties: {path: {type: string, description: The path of the file to read.}}, required: [path]} - 名称: list_files 描述: List files and directories in a given directory path. 输入参数: {type: object, properties: {path: {type: string, description: The directory path to list.}}, required: [path]}4.3 第三步将MCP Tools适配为LangChain Tools发现工具只是第一步我们需要将这些符合MCP协议的工具描述转换成LangChain Agent能够识别和调用的Tool对象。langchain-mcp-adapters库提供了这个桥梁。# 文件create_langchain_tools.py import asyncio from langchain_mcp_adapters.tools import MCPToolkit from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def create_mcp_toolkit(): 创建并返回一个包含所有MCP Tools的LangChain Toolkit server_params StdioServerParameters( commandpython, args[-m, mcp_server_filesystem, .], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 关键步骤使用 MCPToolkit 将 session 转换为 LangChain 工具集 toolkit await MCPToolkit.from_session(session).ainit() # 获取LangChain格式的Tool列表 tools toolkit.get_tools() print(f成功创建了 {len(tools)} 个LangChain Tools。) for tool in tools: print(fTool: {tool.name} - {tool.description}) return tools if __name__ __main__: tools asyncio.run(create_mcp_toolkit())MCPToolkit是一个强大的适配器它处理了协议转换、参数封装和异步调用等底层细节为我们提供了开箱即用的Tool对象列表。4.4 第四步构建并运行LangChain Agent现在我们有了标准的LangChain Tools构建Agent就与平常无异了。我们使用OpenAI的LLM和ReAct代理框架。# 文件run_agent_with_mcp.py import asyncio import os from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import PromptTemplate # 导入上一步编写的工具创建函数 from create_langchain_tools import create_mcp_toolkit async def main(): # 1. 创建LLM确保已设置OPENAI_API_KEY环境变量 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 使用gpt-4o-mini成本更低效果足够 # 2. 获取MCP Tools tools await create_mcp_toolkit() # 3. 定义Agent的提示词模板 prompt PromptTemplate.from_template( 你是一个有帮助的AI助手可以访问文件系统来回答用户问题。 你可以使用以下工具 {tools} 请严格按照以下格式回答 问题用户输入的问题 思考你需要思考如何一步步解决问题 行动要使用的工具名称 行动输入工具的输入必须是有效的JSON格式 观察工具返回的结果 ... (这个思考/行动/观察循环可以重复多次) 最终答案根据观察得出的最终答案 如果用户的问题不需要使用工具或者工具无法解决请直接给出答案。 开始 问题{input} 思考{agent_scratchpad} ) # 4. 创建ReAct Agent agent create_react_agent(llm, tools, prompt) # 5. 创建Agent执行器 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 开启详细日志方便观察Agent的思考过程 handle_parsing_errorsTrue, # 处理解析错误 max_iterations5, # 限制最大迭代次数防止死循环 ) # 6. 运行Agent测试一个需要文件操作的问题 # 首先确保当前目录下有一个测试文件例如 test.txt内容为“Hello, MCP!” test_question 请读取当前目录下名为test.txt的文件并告诉我它的内容是什么 print(f用户问题: {test_question}) print(- * 50) try: result await agent_executor.ainvoke({input: test_question}) print(\n *50) print(f最终答案: {result[output]}) except Exception as e: print(f执行过程中出现错误: {e}) if __name__ __main__: asyncio.run(main())4.5 第五步扩展 - 连接多个Skill ServersMCP的真正威力在于可以轻松集成多个独立的Skill。假设我们还有一个提供SQLite查询的Skill Server。启动SQLite Skill Server(需要先有一个SQLite数据库文件例如example.db)# 在另一个终端或后台进程 mcp run sqlite --db-path ./example.db在LangChain中连接多个Servers# 文件connect_multiple_servers.py (部分关键代码) import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import MCPToolkit async def create_tools_from_multiple_servers(): all_tools [] # 定义多个Server的参数 servers [ StdioServerParameters(commandpython, args[-m, mcp_server_filesystem, .]), StdioServerParameters(commandpython, args[-m, mcp_server_sqlite, --db-path, ./example.db]), ] for server_params in servers: try: async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() toolkit await MCPToolkit.from_session(session).ainit() server_tools toolkit.get_tools() all_tools.extend(server_tools) print(f从Server {server_params.args[-1]} 加载了 {len(server_tools)} 个工具。) except Exception as e: print(f连接Server {server_params.args} 失败: {e}) continue print(f总计加载了 {len(all_tools)} 个工具。) return all_tools # 后续构建Agent的代码与单Server类似只是tools来源变成了all_tools这样你的Agent就同时具备了文件操作和数据库查询的能力而这两部分能力是由两个完全独立、可单独维护的服务提供的。5. 完整示例构建一个多功能文档查询Agent让我们整合以上步骤构建一个更实用的示例一个能同时查询文件内容和数据库信息的智能助手。项目结构multi_skill_agent/ ├── data/ │ ├── reports/ # 存放一些文本报告 │ │ └── q1_report.txt │ └── example.db # SQLite数据库文件 ├── skills/ # (可选) 放置自定义Skill的代码 ├── main.py # 主程序入口 └── requirements.txtrequirements.txt:langchain0.1.0 langchain-openai0.0.5 langchain-mcp-adapters0.1.0 mcp0.1.0 mcp-server-filesystem mcp-server-sqlitemain.py - 完整实现:import asyncio import sys import os from pathlib import Path from typing import List from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import PromptTemplate from langchain_core.tools import BaseTool from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import MCPToolkit class MultiSkillAgent: def __init__(self, openai_api_key: str None, model: str gpt-4o-mini): 初始化Agent设置LLM和工具列表 if openai_api_key: os.environ[OPENAI_API_KEY] openai_api_key self.llm ChatOpenAI(modelmodel, temperature0) self.tools: List[BaseTool] [] self.agent_executor: AgentExecutor None async def connect_skill_server(self, server_params: StdioServerParameters) - List[BaseTool]: 连接一个MCP Skill Server并返回其工具列表 server_tools [] try: print(f正在连接MCP Server: {server_params.command} { .join(server_params.args)}) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() toolkit await MCPToolkit.from_session(session).ainit() server_tools toolkit.get_tools() print(f 成功加载 {len(server_tools)} 个工具。) except Exception as e: print(f 连接失败: {e}) return server_tools async def initialize(self, skill_configs: list): 初始化Agent连接所有配置的Skill Servers print(开始初始化Multi-Skill Agent...) for config in skill_configs: tools await self.connect_skill_server(config) self.tools.extend(tools) if not self.tools: print(警告未加载到任何工具Agent将无法执行操作。) return print(f总计加载了 {len(self.tools)} 个工具。) # 创建Agent提示词可根据工具特性优化 prompt self._create_agent_prompt() # 创建ReAct Agent agent create_react_agent(self.llm, self.tools, prompt) # 创建执行器 self.agent_executor AgentExecutor( agentagent, toolsself.tools, verboseTrue, handle_parsing_errorsTrue, max_iterations8, early_stopping_methodgenerate, ) print(Agent初始化完成) def _create_agent_prompt(self) - PromptTemplate: 创建Agent的提示词模板 tool_descriptions \n.join([f- {tool.name}: {tool.description} for tool in self.tools]) template f 你是一个强大的AI助手可以调用多种工具来帮助用户解决问题。 你可以使用的工具如下 {tool_descriptions} 请严格按照以下格式回应 问题用户的问题 思考分析问题决定是否需要以及使用哪个工具 行动要使用的工具名称必须是上述列表中的一个 行动输入工具的输入必须是有效的JSON对象 观察工具返回的结果 ...这个思考/行动/观察循环可以重复多次但请尽量简洁 最终答案根据所有观察给出清晰、完整的最终答案 如果问题不需要工具或工具无法解决请直接推理并给出最终答案。 请确保“行动输入”是严格的JSON例如对于需要path参数的工具输入应为{{path: /some/file.txt}} 现在开始处理第一个问题。 问题{{input}} {{agent_scratchpad}} return PromptTemplate.from_template(template) async def run(self, query: str) - str: 运行Agent处理查询 if not self.agent_executor: return Agent未正确初始化。 print(f\n{*60}) print(f处理查询: {query}) print(*60) try: result await self.agent_executor.ainvoke({input: query}) return result.get(output, 未获得有效输出。) except Exception as e: return f执行过程中出现错误: {e} async def main(): 主函数配置并运行Agent # 1. 定义要连接的Skill Servers # 假设项目根目录下有一个data文件夹里面有报告和数据库 base_dir Path(__file__).parent data_dir base_dir / data skill_configs [ # 文件系统Skill根目录设置为data/reports StdioServerParameters( commandsys.executable, # 使用当前Python解释器 args[-m, mcp_server_filesystem, str(data_dir / reports)] ), # SQLite Skill连接到example.db StdioServerParameters( commandsys.executable, args[-m, mcp_server_sqlite, --db-path, str(data_dir / example.db)] ), ] # 2. 创建并初始化Agent agent MultiSkillAgent() await agent.initialize(skill_configs) # 3. 准备测试数据和查询 # 首先确保数据存在这里模拟创建 reports_dir data_dir / reports reports_dir.mkdir(parentsTrue, exist_okTrue) (reports_dir / q1_report.txt).write_text(2024年第一季度销售额为120万元主要增长来自华东地区。) import sqlite3 db_path data_dir / example.db if not db_path.exists(): conn sqlite3.connect(db_path) cursor conn.cursor() cursor.execute(CREATE TABLE IF NOT EXISTS products (id INTEGER PRIMARY KEY, name TEXT, price REAL)) cursor.execute(INSERT INTO products (name, price) VALUES (Laptop, 999.99), (Mouse, 29.99)) conn.commit() conn.close() print(已创建示例数据库和表。) # 4. 运行测试查询 test_queries [ 读取reports文件夹下的q1_report.txt文件总结一下销售额情况。, 查询数据库里所有产品的名称和价格。, 综合一下我们第一季度销售额不错数据库里最贵的产品是什么它的价格占季度销售额的比例大概是多少, ] for query in test_queries: answer await agent.run(query) print(f\n 最终答案:\n{answer}\n) print(-*60) if __name__ __main__: asyncio.run(main())6. 运行结果与效果验证运行上述main.py脚本你将看到类似以下的输出清晰地展示了Agent的思考链Chain of Thought和工具调用过程开始初始化Multi-Skill Agent... 正在连接MCP Server: /usr/local/bin/python -m mcp_server_filesystem /path/to/data/reports 成功加载 2 个工具。 正在连接MCP Server: /usr/local/bin/python -m mcp_server_sqlite --db-path /path/to/data/example.db 成功加载 2 个工具。 总计加载了 4 个工具。 Agent初始化完成 处理查询: 读取reports文件夹下的q1_report.txt文件总结一下销售额情况。 进入新的Agent执行链... 思考用户想要读取一个文件并总结销售额。我需要先找到并读取这个文件。 行动list_files 行动输入{path: .} 观察[., q1_report.txt] 思考我看到了 q1_report.txt 文件。现在需要读取它的内容。 行动read_file 行动输入{path: q1_report.txt} 观察2024年第一季度销售额为120万元主要增长来自华东地区。 思考我已经读取了文件内容。现在需要总结销售额情况。内容显示第一季度销售额120万元主要增长来自华东地区。我可以直接给出总结。 最终答案根据文件内容2024年第一季度销售额为120万元主要增长动力来自华东地区。 最终答案: 根据文件内容2024年第一季度销售额为120万元主要增长动力来自华东地区。 ------------------------------------------------------------ 处理查询: 查询数据库里所有产品的名称和价格。 进入新的Agent执行链... 思考用户想查询数据库。我需要使用SQLite工具。有一个sql_query工具可以用来执行SQL查询。 行动sql_query 行动输入{query: SELECT name, price FROM products} 观察[{name: Laptop, price: 999.99}, {name: Mouse, price: 29.99}] 思考查询成功返回了产品列表。我可以直接呈现这个结果。 最终答案数据库中共有2个产品 1. Laptop - 价格: 999.99 2. Mouse - 价格: 29.99 最终答案: 数据库中共有2个产品 1. Laptop - 价格: 999.99 2. Mouse - 价格: 29.99 ------------------------------------------------------------效果验证要点工具发现Agent成功从两个独立的Server加载了4个工具2个文件操作2个数据库操作。正确路由对于文件查询Agent正确选择了list_files和read_file工具对于数据库查询正确选择了sql_query工具。参数构造Agent能根据工具的描述自动构造正确的JSON格式输入如{path: q1_report.txt}{query: SELECT ...}。复杂任务处理在第三个综合查询中Agent展示了多步推理和工具组合的能力先读取文件获取销售额再查询数据库获取产品价格最后进行简单的数学计算和总结。解耦验证你可以独立更新mcp_server_sqlite的版本或者替换成另一个提供相同工具sql_query但内部实现不同的Server而main.py中的Agent代码完全不需要修改。7. 常见问题与排查思路在实际集成MCP与LangChain的过程中你可能会遇到以下典型问题问题现象可能原因排查方式解决方案连接MCP Server失败1. Server命令路径错误。2. 所需Python模块未安装。3. Server进程已崩溃。1. 在终端手动运行mcp run filesystem看是否报错。2. 检查pip list | grep mcp-server。3. 查看Python异常堆栈信息。1. 确保command和args参数正确。2. 使用pip install安装对应的server包。3. 使用try...except包裹连接代码并添加重试逻辑。Agent找不到或错误调用工具1. 工具名称不匹配。2. 输入参数格式错误。3. LLM的提示词未清晰描述工具。1. 在初始化时打印所有加载的工具名称和描述。2. 开启Agent的verboseTrue模式观察其“行动输入”。3. 检查提示词模板中的{tools}是否被正确替换。1. 确保工具描述清晰。可在提示词中强化工具名称。2. 使用handle_parsing_errorsTrue让Agent有机会纠正格式错误。3. 考虑使用更结构化的输出解析器如OpenAI Functions。Agent陷入循环或迭代次数过多1. 工具返回结果未满足Agent预期。2. 任务过于复杂或模糊。3.max_iterations设置过高。观察verbose日志看思考步骤是否在重复或无效循环。1. 优化工具返回的信息使其更直接、结构化。2. 在提示词中明确任务边界和停止条件。3. 合理设置max_iterations如5-10并使用early_stopping_method。性能问题响应慢1. 每次调用都新建连接。2. LLM本身响应慢。3. Skill Server处理耗时。1. 检查是否在每次工具调用时都创建了新的Session。2. 测试直接调用LLM的延迟。3. 单独测试Skill Server的响应时间。1.复用Session确保ClientSession在整个Agent生命周期内保持连接而不是每次调用都新建。2. 考虑使用更快的LLM或模型。3. 对耗时Skill做异步优化或缓存。权限或安全错误1. 文件系统Skill访问了无权目录。2. SQL查询包含危险操作。1. 检查Skill Server启动时设置的根目录。2. 审查Agent生成的SQL语句。1. 在启动Skill Server时严格限制其访问范围如只读模式、指定目录。2. 对于数据库Skill可使用只有SELECT权限的账户或在Server层进行SQL过滤。8. 最佳实践与工程建议将MCP架构投入生产环境或复杂项目时遵循以下最佳实践可以避免很多坑8.1 Skill设计与开发单一职责每个Skill Server应专注于一个特定的领域如文件、数据库、API网关、搜索引擎。这符合微服务的设计哲学便于独立开发、测试和部署。清晰的工具描述在实现Skill时为每个工具提供准确、详细的description和inputSchema。LLM依赖这些描述来选择工具模糊的描述会导致调用错误。健壮的错误处理Skill内部应该有完善的错误处理机制并将错误信息以结构化的方式返回给客户端帮助Agent理解失败原因例如“文件未找到” vs “权限被拒绝”。版本化与兼容性对Skill进行版本管理。当工具接口需要变更时考虑通过新增工具如search_v2而非修改现有工具来保持向后兼容。8.2 Agent应用开发Session管理这是性能关键。务必在Agent应用启动时创建并维护与各个Skill Server的持久化Session连接池避免为每次工具调用建立新连接的开销。提示词工程精心设计Agent的提示词。除了列出工具还应说明工具的适用场景和调用约束。例如“sql_query工具仅用于查询不支持INSERT/UPDATE”。超时与重试为工具调用设置合理的超时时间并实现重试机制特别是对于网络不稳定的远程HTTP Skill。可观测性记录详细的日志包括Agent的思考过程、工具调用请求和响应。这对于调试复杂问题和优化提示词至关重要。8.3 安全与权限最小权限原则以最低必要权限运行Skill Server。文件系统Skill应限制在特定目录数据库Skill应使用只有查询权限的账号。输入验证与净化在Skill Server端对输入参数进行严格验证。防止路径遍历../../../etc/passwd、SQL注入等攻击。网络隔离如果Skill Server以HTTP模式运行在网络上务必将其置于内网并通过防火墙或API网关控制访问。避免将敏感能力的Server暴露在公网。审计日志记录所有工具调用的审计日志包括调用者、参数、结果和时间戳便于安全审查和问题追溯。8.4 部署与运维容器化将每个Skill Server和Agent应用分别容器化Docker便于统一部署、资源隔离和水平扩展。服务发现在生产环境中Skill Server的地址可能是动态的。可以考虑使用简单的服务发现机制如环境变量、配置中心、Consul等让Agent能动态发现可用的Skill。健康检查为Skill Server实现健康检查端点Agent应用或编排系统可以定期检查其状态并进行故障转移或重启。监控与告警监控Skill Server的资源使用情况CPU、内存、请求延迟和错误率。设置告警以便在服务异常时及时介入。通过遵循这些实践你可以构建出一个既灵活强大又稳定可靠、易于维护的基于MCP和LangChain的AI应用系统。9. 总结与后续学习方向本文深入探讨了将LangChain Agent与MCP协议及Skills集成的技术原理与实践方法。我们从一个具体的开发痛点出发——工具与Agent的紧耦合——介绍了MCP如何通过标准化协议实现解耦并一步步演示了如何启动Skill Server、在LangChain中发现并封装工具、最终构建一个能协同使用多种能力的智能Agent。关键收获架构升级MCP不仅仅是一个工具调用库它更是一种架构模式。它促使我们将AI应用中的“能力提供者”Skills和“能力调度者”Agent分离让系统更模块化、更易扩展。开发效率一旦协议标准化团队可以并行开发Skills和Agent应用。前端Agent提示词、UI和后端各种能力服务的开发者可以更独立地工作。生态潜力一个开放的MCP Skill生态正在形成。未来我们或许可以像安装插件一样为我们的Agent轻松添加图像识别、代码执行、邮件发送等成千上万种能力而无需修改核心代码。后续你可以深入探索的方向开发自定义Skill本文使用了官方示例Server。尝试用Python或Node.js等MCP支持的语言开发一个你自己的Skill例如调用一个内部API、操作一个特定的云服务。探索更多现成Skill在MCP的官方仓库和社区中寻找更多有趣的Skill如github、brave-search、notion等极大地扩展Agent的能力边界。深入LangGraph对于需要复杂工作流、状态管理和循环的AgentLangChain的LangGraph框架是更强大的选择。研究如何将MCP Tools集成到LangGraph的图中。性能优化研究如何对频繁调用的Skill结果进行缓存如何并行调用多个独立工具以及如何对Agent的思考过程进行优化以减少不必要的工具调用成本考量。生产化部署将本文的示例代码改造为真正的生产服务考虑如何做版本管理、灰度发布、监控告警和自动扩缩容。技术的价值在于解决真实问题。MCP与LangChain的结合为构建下一代模块化、可插拔的AI智能体应用提供了坚实的技术基础。建议你将本文的示例代码作为起点亲手实践并改造将其应用到你的具体业务场景中真正体验它所带来的“全方位工作效率提升”。