在AI应用开发领域如何让大模型稳定、可靠地调用外部工具和数据是每个开发者都会遇到的“拦路虎”。尤其是在构建复杂的智能体Agent时工具调用的稳定性、数据获取的规范性以及架构的可维护性常常成为项目从Demo走向生产的关键瓶颈。本文将围绕LangChain与Model Context Protocol (MCP)协议为你提供一套从核心概念到项目落地的完整实战指南。无论你是希望为现有应用集成AI能力还是计划从零构建一个功能强大的智能体本文涵盖的工具定义、拦截器实现、架构解析与综合案例都将为你提供可直接复用的代码与清晰的工程思路。1. 背景与核心概念为什么是LangChain MCP在深入代码之前我们需要理解这两个技术栈各自解决了什么问题以及它们结合带来的优势。LangChain是一个用于开发由语言模型驱动的应用程序的框架。它并非一个“开箱即用”的产品而是一个提供了丰富“积木”组件的工具箱。其核心价值在于标准化流程将与大模型交互的复杂流程如问答、摘要、工具调用抽象成可组合的链Chains、智能体Agents等。工具集成提供了统一的接口来定义和调用外部工具如搜索引擎、数据库、API这是构建智能体的基础。上下文管理帮助开发者高效地处理超出模型上下文长度的文本通过检索、摘要等方式管理输入。然而在LangChain中定义和使用工具传统上需要将工具的逻辑代码与应用程序代码紧密耦合。当工具数量增多、来源多样如不同团队的内部API、第三方服务时这会带来维护和部署的挑战。这正是Model Context Protocol (MCP)旨在解决的问题。MCP是一个开放协议它定义了大模型或AI应用与外部工具、数据源之间进行通信的标准方式。你可以把它想象成AI世界的“USB协议”或“HTTP for AI”。MCP的核心思想是解耦工具/数据作为服务器Server任何能提供计算、查询或数据获取能力的服务都可以实现为一个MCP服务器。它暴露出工具列表名称、描述、参数模式和数据源。AI应用作为客户端ClientLangChain应用或其他AI框架可以作为MCP客户端通过标准协议动态发现并调用这些远程工具无需在应用代码中硬编码工具逻辑。结合的优势架构清晰工具服务独立部署、维护和升级不影响主应用。语言无关工具服务器可以用任何语言编写Python, Node.js, Go等只要遵循MCP协议即可。动态发现客户端可以在运行时发现可用的工具实现更灵活的插件化架构。安全与权限可以在协议层对工具调用进行认证、授权和审计。本文将教你如何在LangChain中作为MCP客户端集成并利用MCP工具同时也会简要介绍如何创建一个简单的MCP服务器让你全面掌握这套技术栈。2. 环境准备与版本说明本教程的示例代码主要使用Python。请确保你的环境满足以下要求操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04 推荐)。大部分命令是通用的。Python版本 3.8。推荐使用3.10或3.11以获得最佳兼容性。包管理工具pip或conda。IDEVS Code, PyCharm 或任何你熟悉的代码编辑器。我们将创建两个项目一个MCP客户端项目主应用和一个MCP服务器项目工具服务。首先为客户端项目创建环境并安装核心依赖。MCP客户端项目依赖# 创建项目目录并进入 mkdir langchain-mcp-client cd langchain-mcp-client # 创建虚拟环境可选但推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate # 安装核心依赖 pip install langchain langchain-community langchain-openai # 安装MCP客户端集成库这是LangChain社区对MCP协议的支持 pip install langchain-mcp-adapters # 安装MCP协议的核心Python实现用于客户端通信 pip install mcp # 安装OpenAI SDK本例使用OpenAI模型 pip install openaiMCP服务器项目依赖我们将在后续章节创建。一个简单的MCP服务器可能只需要mcp库和实现工具逻辑所需的库如requests。版本兼容性说明 AI生态迭代迅速LangChain和MCP的版本更新可能带来API变化。本文示例基于以下较新且稳定的版本构建如果你的环境存在差异请参考官方文档调整。langchain0.1.0langchain-community0.0.10mcp0.1.0langchain-mcp-adapters(请关注其最新版本)关键点langchain-mcp-adapters是社区维护的适配器库它提供了将MCP服务器工具转换为LangChainTool对象的桥梁是本文实现的核心。3. MCP协议核心与LangChain集成原理拆解在动手编码前理解MCP的通信模型和LangChain的集成方式至关重要。3.1 MCP协议基础客户端-服务器模型MCP协议通常使用标准输入/输出stdio或HTTP作为传输层。在开发调试阶段stdio更为常见。其工作流程如下启动客户端你的LangChain应用启动一个子进程运行MCP服务器程序。握手通过stdio交换初始化消息协商协议版本。列表工具客户端向服务器发送tools/list请求服务器返回所有可用工具的定义包括名称、描述和JSON Schema格式的参数规范。调用工具客户端向服务器发送tools/call请求包含工具名和参数。服务器执行相应逻辑返回结果或错误。关闭通信结束。一个工具的定义在MCP中是一个结构化的JSON对象LangChain的适配器会将其自动映射为Tool对象。3.2 LangChain如何集成MCP工具langchain-mcp-adapters库的核心类是MCPClient和MCPToolkit。MCPClient负责与MCP服务器建立连接、管理通信。MCPToolkit一个高级封装它使用MCPClient获取工具列表并将其转换为一个LangChainToolkit。Toolkit可以很方便地提供给AgentExecutor使用。集成流程创建MCPClient连接到你的MCP服务器通过子进程命令。使用client.list_tools()获取远程工具定义。使用MCPToolkit.from_client(client)创建工具包。从工具包中获取get_tools()方法返回的Tool列表。将这些Tool列表赋予LangChain的智能体Agent。这样智能体在思考时就能“看到”这些来自远程服务器的工具并在需要时调用它们。4. 实战案例一构建一个天气查询MCP服务器让我们从创建一个最简单的MCP服务器开始。这个服务器提供一个get_weather工具接收城市名参数返回模拟的天气信息。项目结构mcp-weather-server/ ├── server.py # MCP服务器主程序 ├── pyproject.toml # 项目依赖声明可选 └── README.md第一步创建服务器代码 (server.py)#!/usr/bin/env python3 一个简单的MCP服务器提供天气查询工具。 import json import sys import asyncio from typing import Any, Dict, List from mcp import Server, NotificationOptions import mcp.server.stdio from mcp.server.models import ToolDefinition from pydantic import BaseModel, Field # 1. 定义工具的参数模型使用Pydantic class GetWeatherParams(BaseModel): 获取天气的参数 city_name: str Field(descriptionThe name of the city to get weather for, e.g., Beijing or New York.) # 2. 创建MCP服务器实例 server Server(weather-mcp-server) # 3. 使用装饰器注册工具 server.list_tools() async def handle_list_tools() - List[ToolDefinition]: 返回服务器提供的工具列表 return [ ToolDefinition( nameget_weather, descriptionGet the current weather information for a given city., inputSchemaGetWeatherParams.model_json_schema(), # 自动生成JSON Schema ) ] server.call_tool() async def handle_call_tool(name: str, arguments: Dict[str, Any]) - List[Dict[str, Any]]: 处理工具调用请求 if name get_weather: # 验证并解析参数 params GetWeatherParams(**arguments) city params.city_name # 模拟天气数据获取逻辑实际项目中这里会调用真实API # 例如response requests.get(fhttps://api.weatherapi.com/v1/current.json?keyYOUR_KEYq{city}) weather_data { city: city, temperature: 22°C, condition: Sunny, humidity: 65%, wind_speed: 15 km/h, forecast: Clear skies throughout the day. } # MCP要求返回一个列表每个元素是一个“文本”或“图像”内容块 return [{ type: text, text: json.dumps(weather_data, indent2, ensure_asciiFalse) }] else: raise ValueError(fUnknown tool: {name}) # 4. 主函数启动stdio服务器 async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, NotificationOptions()) if __name__ __main__: asyncio.run(main())代码解读参数模型使用Pydantic的BaseModel定义工具参数这能自动生成标准的JSON Schema供客户端理解如何调用。工具列表server.list_tools装饰的函数返回工具定义列表。ToolDefinition是关键它包含了工具名、描述和输入模式。工具调用server.call_tool装饰的函数处理所有工具调用。通过name区分不同的工具使用定义好的Pydantic模型验证arguments。返回格式MCP协议要求工具调用返回一个内容块列表。最常用的是{type: text, text: ...}。通信层使用mcp.server.stdio.stdio_server()来建立基于标准输入输出的通信通道。第二步安装服务器依赖并运行在mcp-weather-server目录下# 安装mcp库 pip install mcp pydantic # 运行服务器 python server.py服务器启动后它会等待来自客户端的连接通过stdio。我们接下来构建客户端来连接它。5. 实战案例二LangChain客户端调用MCP工具现在我们构建主应用客户端使用LangChain创建一个智能体来调用刚刚创建的天气查询工具。项目结构langchain-mcp-client/ ├── client_agent.py ├── .env # 存储API密钥需自己创建 └── requirements.txt第一步创建客户端代理代码 (client_agent.py)import asyncio import os from pathlib import Path from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_mcp_adapters import MCPClient, MCPToolkit from dotenv import load_dotenv # 加载环境变量用于OpenAI API Key load_dotenv() async def main(): # 0. 配置 # 确保已设置环境变量 OPENAI_API_KEY # 假设天气服务器脚本位于 ../mcp-weather-server/server.py server_script_path Path(__file__).parent.parent / mcp-weather-server / server.py # 1. 创建MCP客户端并连接服务器 # 通过子进程启动MCP服务器客户端通过stdio与其通信 async with MCPClient.create_with_stdio_server( command[python, str(server_script_path)] ) as client: # 2. 从MCP客户端创建工具包 toolkit await MCPToolkit.from_client(client) # 获取LangChain可用的Tool对象列表 tools await toolkit.get_tools() print(f成功从MCP服务器加载了 {len(tools)} 个工具:) for tool in tools: print(f - {tool.name}: {tool.description}) # 3. 初始化大语言模型 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 使用gpt-4o-mini可根据情况调整 # 4. 构建智能体提示词 prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手可以调用工具来获取信息。请根据用户问题决定是否需要调用工具。如果调用请严格按照工具要求的格式提供参数。), MessagesPlaceholder(variable_namechat_history, optionalTrue), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) # 5. 创建OpenAI Tools Agent agent create_openai_tools_agent(llm, tools, prompt) # 6. 创建代理执行器 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 7. 运行测试 print(\n--- 开始测试 ---) test_questions [ 北京今天的天气怎么样, 帮我查询一下纽约的天气情况。, 上海和伦敦的天气对比一下。 ] for question in test_questions: print(f\n用户: {question}) try: # 注意AgentExecutor的invoke是同步方法我们在async函数中需要用to_thread或直接调用 # 这里为简化在async函数中同步调用。生产环境应考虑更优的并发处理。 result await asyncio.to_thread(agent_executor.invoke, {input: question}) print(f助手: {result[output]}) except Exception as e: print(f执行过程中出现错误: {e}) if __name__ __main__: asyncio.run(main())第二步配置环境并运行在项目根目录创建.env文件填入你的OpenAI API KeyOPENAI_API_KEYsk-your-openai-api-key-here确保天气MCP服务器 (server.py) 没有在运行或者运行在另一个终端。运行客户端代理python client_agent.py预期输出与过程解析客户端脚本会启动一个子进程来运行server.py并通过stdio建立MCP连接。连接成功后客户端会列出从服务器发现的工具get_weather。针对每个测试问题AgentExecutor会驱动大模型进行思考。模型会识别出需要查询天气决定调用get_weather工具并生成符合JSON Schema的参数如{city_name: 北京}。LangChain将调用请求通过MCP客户端发送给服务器。服务器执行工具逻辑返回模拟的天气数据。MCP客户端将结果返回给LangChain模型根据结果生成最终的自然语言回复给用户。由于设置了verboseTrue你将在控制台看到详细的思考过程、工具调用和结果。关键点整个过程中你的主应用客户端完全不需要知道get_weather工具内部是如何实现的是模拟数据还是调用了真实API。它只通过标准的MCP协议进行交互实现了完美的解耦。6. 进阶实战实现工具调用拦截器Interceptor在生产环境中我们经常需要在工具调用前后执行一些通用逻辑例如日志记录记录谁在何时调用了什么工具参数和结果是什么。权限校验检查当前用户或会话是否有权调用该工具。参数清洗/转换对输入参数进行标准化处理。错误处理与重试对调用失败的工具进行降级或重试。性能监控统计工具调用的耗时。在LangChain with MCP的架构中我们可以在两个层面实现拦截器MCP服务器层面在工具的实现逻辑外层包裹拦截逻辑。LangChain Tool层面通过自定义Tool类或包装现有的Tool对象。这里我们展示第二种方式因为它更通用且不依赖于具体的MCP服务器实现。我们将创建一个LoggingInterceptorTool它包装原始的MCP Tool增加日志功能。创建interceptors.py文件from typing import Any, Optional, Type, Union from langchain_core.tools import BaseTool, Tool, StructuredTool from langchain_core.callbacks import CallbackManagerForToolRun, AsyncCallbackManagerForToolRun from pydantic import BaseModel import logging import time import json # 设置日志 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class LoggingInterceptorTool(BaseTool): 一个为工具添加日志和监控的拦截器包装器。 original_tool: BaseTool 被包装的原始工具对象。 def __init__(self, original_tool: BaseTool, **kwargs): # 继承原始工具的所有属性 kwargs.update({ name: original_tool.name, description: original_tool.description, args_schema: original_tool.args_schema, original_tool: original_tool, }) super().__init__(**kwargs) def _run( self, *args, **kwargs, ) - Any: 同步执行工具并添加拦截逻辑。 start_time time.time() tool_name self.original_tool.name # 1. 调用前日志 logger.info(f[Tool Call START] {tool_name} - Args: {kwargs}) try: # 2. 调用原始工具 result self.original_tool._run(*args, **kwargs) elapsed time.time() - start_time # 3. 调用成功日志 # 注意结果可能很大生产环境可能需要采样或脱敏 result_sample str(result)[:200] ... if len(str(result)) 200 else str(result) logger.info(f[Tool Call SUCCESS] {tool_name} - Time: {elapsed:.3f}s - Result sample: {result_sample}) return result except Exception as e: elapsed time.time() - start_time # 4. 调用失败日志 logger.error(f[Tool Call FAILED] {tool_name} - Time: {elapsed:.3f}s - Error: {e}, exc_infoTrue) raise # 重新抛出异常让上游处理 async def _arun( self, *args, **kwargs, ) - Any: 异步执行工具并添加拦截逻辑。 start_time time.time() tool_name self.original_tool.name logger.info(f[Async Tool Call START] {tool_name} - Args: {kwargs}) try: result await self.original_tool._arun(*args, **kwargs) elapsed time.time() - start_time result_sample str(result)[:200] ... if len(str(result)) 200 else str(result) logger.info(f[Async Tool Call SUCCESS] {tool_name} - Time: {elapsed:.3f}s - Result sample: {result_sample}) return result except Exception as e: elapsed time.time() - start_time logger.error(f[Async Tool Call FAILED] {tool_name} - Time: {elapsed:.3f}s - Error: {e}, exc_infoTrue) raise def wrap_tools_with_interceptor(tools: list[BaseTool]) - list[BaseTool]: 将工具列表中的每个工具用拦截器包装。 wrapped_tools [] for tool in tools: # 检查工具是否已经是包装过的避免重复包装 if isinstance(tool, LoggingInterceptorTool): wrapped_tools.append(tool) else: wrapped_tools.append(LoggingInterceptorTool(original_tooltool)) return wrapped_tools在客户端应用中使用拦截器修改之前的client_agent.py在获取工具后应用拦截器。# ... 之前的导入和代码 ... from interceptors import wrap_tools_with_interceptor # 导入拦截器 async def main(): # ... 前面的代码创建client, toolkit ... tools await toolkit.get_tools() # 新增应用拦截器包装工具 tools wrap_tools_with_interceptor(tools) # print(f成功从MCP服务器加载了 {len(tools)} 个工具 (已包装拦截器)...) # ... 后续创建agent和执行的代码不变 ...现在每次智能体调用get_weather工具时你都会在控制台看到详细的调用日志包括参数、执行时间和结果摘要。这为调试、监控和审计提供了极大便利。7. 常见问题与排查思路在集成LangChain与MCP的过程中你可能会遇到以下典型问题问题现象可能原因排查步骤与解决方案连接MCP服务器失败1. 服务器脚本路径错误。2. 服务器依赖未安装。3. 服务器脚本本身有语法错误。1. 检查server_script_path是否正确指向有效的.py文件。2. 在服务器目录下独立运行python server.py看是否能正常启动应等待输入不报错。3. 查看客户端报错信息通常是子进程启动失败。ModuleNotFoundError: No module named mcpMCP Python包未在客户端或服务器环境安装。1. 在客户端和服务器环境中分别执行pip list | grep mcp确认。2. 使用pip install mcp安装。注意langchain-mcp-adapters可能对mcp有最低版本要求。工具列表为空 (len(tools) 0)1. 服务器未正确实现list_tools处理函数。2. 客户端与服务器协议版本不兼容。3. 连接已建立但初始化消息交换失败。1. 确保服务器代码中server.list_tools()装饰的函数已正确定义并返回非空列表。2. 尝试使用mcp库的CLI工具测试服务器npx modelcontextprotocol/inspector python server.py(需要Node.js环境)。3. 在服务器代码中添加日志打印接收到的请求和发出的响应。工具调用时参数错误1. 客户端传递的参数格式不符合服务器定义的JSON Schema。2. 服务器端Pydantic模型验证失败。3. 工具名拼写错误。1. 在客户端设置verboseTrue观察Agent生成的调用参数是否正确。2. 检查服务器端ToolDefinition中的inputSchema是否与GetWeatherParams的schema一致。3. 确保工具名在调用 (handle_call_tool) 和定义 (handle_list_tools) 中完全一致。Agent不调用工具直接回答1. 模型如GPT-4o-mini能力不足以理解任务需要调用工具。2. 提示词Prompt未清晰指示模型使用工具。3. 工具描述不够清晰模型无法匹配。1. 尝试使用能力更强的模型如gpt-4o。2. 优化系统提示词明确告知模型“你可以使用以下工具”并列出工具名和描述。3. 完善工具的描述 (description)使其更贴近自然语言问题。例如“查询城市天气”比“get_weather”更好。异步运行时事件循环冲突在Jupyter Notebook或已有事件循环的环境中运行异步代码。1. 使用asyncio.run(main())作为入口点。2. 如果在已有循环中使用await main()并确保顶层有asyncio.get_event_loop().run_until_complete()。3. 检查是否有其他库如某些旧版OpenAI干扰了事件循环。8. 最佳实践与工程建议将LangChain与MCP用于生产级项目时请考虑以下建议1. 工具设计与命名规范描述清晰工具的描述 (description) 应尽可能详细、自然帮助大模型准确理解其用途。例如“根据城市名称查询当前温度、天气状况和湿度”优于“获取天气”。参数标准化使用Pydantic模型严格定义参数并利用Field(description...)为每个字段添加描述。这既是文档也能被某些客户端用于生成UI。命名有含义工具名使用蛇形命名法snake_case并体现其核心功能如calculate_monthly_revenue。2. MCP服务器部署与运维独立进程每个MCP服务器应作为独立进程运行便于资源隔离、独立扩缩容和故障恢复。健康检查为MCP服务器实现健康检查端点如果使用HTTP传输或信号处理对于stdio便于容器编排平台如K8s管理。版本管理工具接口名称、参数变更时需考虑版本兼容性。可以通过工具名加后缀如get_weather_v2或协议层支持版本来管理。错误处理服务器端工具实现必须有完善的异常捕获和日志记录返回给客户端的错误信息应友好且不泄露内部细节。3. 客户端LangChain应用架构连接池对于HTTP模式的MCP服务器考虑使用连接池避免频繁建立连接的开销。超时与重试为工具调用设置合理的超时时间并对可重试的错误如网络抖动实现重试机制。这可以在拦截器中实现。工具路由一个客户端可以连接多个MCP服务器。设计一个“工具路由层”根据工具名前缀或元数据将调用分发到不同的后端服务器。配置化将MCP服务器的连接信息命令、URL放在配置文件中而不是硬编码在代码里。4. 安全与权限传输安全如果MCP服务器部署在远程应使用SSH隧道、TLSHTTPS等方式加密通信。认证与授权在MCP协议之上实现认证层。客户端可以在调用工具时传递认证令牌如JWT服务器在handle_call_tool中验证令牌和权限。输入验证与清理除了Pydantic验证在工具实现内部对来自不可信客户端即大模型生成的参数进行严格的业务逻辑验证和清理防止注入攻击。5. 可观测性结构化日志如拦截器示例所示记录工具调用的关键指标工具名、参数、耗时、状态、错误码。使用JSON格式输出便于接入ELK等日志系统。指标监控在拦截器中增加指标上报如调用次数、耗时分布、错误率接入Prometheus等监控系统。链路追踪为每个用户会话或请求生成唯一的追踪ID并在所有工具调用中传递该ID便于在分布式系统中追踪完整请求链路。通过遵循以上实践你可以构建出健壮、可维护、可扩展的基于LangChain和MCP的智能体系统从容应对从简单Demo到复杂企业级应用的各种场景。