最近在尝试将大模型与外部数据源、工具进行深度集成时你是否也遇到过这样的困境每个工具都需要写一套特定的适配代码Agent调用逻辑复杂且难以复用不同项目间的工具链无法共享随着Model Context ProtocolMCP的出现这些问题有了标准化的解决方案。本文将带你从零开始深入MCP协议原理并手把手完成LangChain与MCP的集成实战构建一个可扩展、标准化的外部工具连接通道彻底搞定数据源接入与工具编排的核心难题。1. 背景与核心概念为什么需要MCP在构建基于大模型的智能应用时让模型能够调用外部工具如数据库查询、API调用、文件操作是释放其潜力的关键。然而传统的集成方式存在几个显著痛点传统方式的挑战紧耦合工具逻辑与Agent核心代码深度绑定任何工具变更都可能引发连锁修改。协议不统一不同工具如SQL数据库、HTTP API、本地脚本有各自的调用方式和数据格式需要大量胶水代码。复用性差为一个项目开发的工具链很难直接迁移到另一个项目或另一个AI框架中。开发效率低开发者需要花费大量时间在工具协议的对接上而非业务逻辑本身。MCPModel Context Protocol的解决方案MCP是一个开放协议旨在为大模型与外部工具、数据源之间提供一套标准化的“通信语言”。你可以把它想象成AI世界的“USB协议”或“HTTP协议”。它定义了工具Tools的标准描述格式工具能做什么、需要什么参数、返回什么结果。资源Resources的统一访问方式如何读取文件、数据库表、API端点等结构化或非结构化数据。提示词模板Prompts的共享机制如何预定义和复用复杂的提示词。核心价值一次开发多处运行用MCP标准封装一个工具如“查询天气”它可以被任何支持MCP的AI框架如LangChain、Claude Desktop、Cursor等直接调用。关注点分离工具提供者专注于实现工具功能AI应用开发者专注于编排和提示工程。生态互联一个繁荣的MCP Server生态正在形成你可以轻松找到连接GitHub、Notion、Slack等服务的现成Server直接集成。LangChain与MCP的关系LangChain作为一个流行的AI应用开发框架其LangGraph等模块擅长于构建复杂的Agent工作流。通过集成MCPLangChain可以将MCP Server提供的标准化工具和资源无缝地纳入到自己的Agent工具包中从而极大地扩展了Agent的能力边界并降低了集成复杂度。2. 环境准备与版本说明在开始实战之前请确保你的开发环境已就绪。本文将使用Python作为主要开发语言。基础环境操作系统macOS / Linux / Windows (WSL2推荐)Python版本 3.10 本文示例基于Python 3.10包管理工具pip 或 poetry核心依赖库及版本我们将创建两个核心部分一个自定义的MCP Server以及一个使用LangChain调用该Server的客户端。请先创建一个新的项目目录。mkdir langchain-mcp-demo cd langchain-mcp-demo建议使用虚拟环境隔离依赖python -m venv venv # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate安装核心依赖。请注意MCP和LangChain的相关库迭代较快以下版本为撰写本文时的稳定版本实际开发时请关注官方文档更新。# 安装MCP核心SDK和标准库 pip install mcp[cli]1.0.0 # 安装LangChain及其MCP集成包 pip install langchain0.1.0 langchain-core0.1.0 pip install langchain-mcp-adapters0.1.0 # 这是LangChain官方提供的MCP适配器 # 安装其他实用库 pip install pydantic2.0.0 httpx0.25.0版本兼容性说明MCP协议本身在快速演进LangChain的MCP适配器也处于早期阶段。如果遇到导入错误或API变更请优先查阅以下官方资源MCP Python SDK: https://github.com/modelcontextprotocol/python-sdkLangChain MCP Adapters: https://python.langchain.com/docs/integrations/protocols/mcp/3. MCP协议核心原理拆解要有效使用MCP必须理解其核心组件和通信模型。MCP采用客户端-服务器Client-Server架构通常通过stdio标准输入输出或SSEServer-Sent Events进行通信。3.1 核心组件MCP Server工具和资源的提供者。它封装了具体的业务逻辑例如执行一个Shell命令、查询数据库、调用第三方API。一个Server可以提供多个工具和资源。MCP Client工具和资源的消费者。通常是AI应用框架如LangChain Agent它向Server发起请求获取工具列表、调用工具或读取资源。TransportClient和Server之间的通信通道。stdio是最简单常用的方式适合本地集成SSE则可用于远程服务。3.2 核心概念Tool工具一个可执行的操作。每个工具都有唯一的name、description供LLM理解用途和inputSchema参数定义基于JSON Schema。Resource资源一个可读取的数据实体如文本文件、数据库表的一行。用URI标识例如file:///path/to/data.json或db://mydb/users?id1。Prompt提示词预定义的提示词模板可以包含变量方便Client快速获取并填充使用。3.3 通信流程简析初始化Client启动Server进程建立通信连接。能力交换Client发送initialize请求Server回复其提供的工具、资源、提示词列表。工具调用Client发送tools/call请求包含工具名和参数。Server执行对应逻辑返回tools/call响应包含执行结果或错误信息。资源读取Client发送resources/read请求包含资源URI。Server返回资源内容。理解了这个模型我们就知道我们的任务是1. 实现一个MCP Server暴露我们自定义的工具。2. 在LangChain中配置Client连接这个Server并将其工具加载到Agent中。4. 实战一构建你的第一个MCP Server我们将创建一个提供两个简单工具的MCP Server一个计算器工具和一个获取系统时间的工具。4.1 项目结构在项目根目录下创建如下文件结构langchain-mcp-demo/ ├── mcp_server.py # MCP Server 主逻辑 ├── client_app.py # LangChain 客户端应用 ├── requirements.txt # 依赖文件 └── README.md4.2 实现MCP Server (mcp_server.py)我们将使用mcpSDK 的Server类来快速构建。#!/usr/bin/env python3 一个简单的自定义MCP Server示例。 提供计算器和获取系统时间两个工具。 import asyncio import json import subprocess from datetime import datetime from typing import Any from mcp import ClientSession, Server, StdioServerParameters from mcp.types import Tool, TextContent, CallToolResult import pydantic # 1. 定义工具的参数模型使用Pydantic class CalculatorInput(pydantic.BaseModel): 计算器工具的输入参数 a: float pydantic.Field(description第一个操作数) b: float pydantic.Field(description第二个操作数) operation: str pydantic.Field( description运算类型支持 add(加), subtract(减), multiply(乘), divide(除), pattern^(add|subtract|multiply|divide)$ ) class GetTimeInput(pydantic.BaseModel): 获取时间工具的输入参数 format: str pydantic.Field( description时间格式字符串例如 %Y-%m-%d %H:%M:%S。默认为标准格式。, default%Y-%m-%d %H:%M:%S ) # 2. 创建Server实例 server Server(my-custom-tools-server) # 3. 注册工具 server.list_tools() async def handle_list_tools() - list[Tool]: 返回此Server提供的工具列表 return [ Tool( namecalculator, description执行简单的四则运算。, inputSchemaCalculatorInput.model_json_schema(), ), Tool( nameget_current_time, description获取当前的系统时间。, inputSchemaGetTimeInput.model_json_schema(), ), ] # 4. 实现工具调用处理逻辑 server.call_tool() async def handle_call_tool(name: str, arguments: dict[str, Any]) - CallToolResult: 根据工具名和参数执行对应的工具 if name calculator: # 验证并解析参数 inputs CalculatorInput(**arguments) a, b, op inputs.a, inputs.b, inputs.operation # 执行计算 if op add: result a b elif op subtract: result a - b elif op multiply: result a * b elif op divide: if b 0: return CallToolResult( content[TextContent(typetext, text错误除数不能为零)], isErrorTrue ) result a / b else: return CallToolResult( content[TextContent(typetext, textf不支持的运算类型: {op})], isErrorTrue ) return CallToolResult( content[TextContent(typetext, textf计算结果: {result})] ) elif name get_current_time: inputs GetTimeInput(**arguments) now datetime.now() formatted_time now.strftime(inputs.format) return CallToolResult( content[TextContent(typetext, textf当前时间: {formatted_time})] ) else: # 如果收到未知工具名返回错误 return CallToolResult( content[TextContent(typetext, textf未知工具: {name})], isErrorTrue ) # 5. 主函数启动Server使用stdio传输 async def main(): # 配置stdio传输参数 params StdioServerParameters( commandpython, # 解释器命令 args[mcp_server.py], # 脚本参数这里就是自身 ) async with await server.run_stdio(params) as (read_stream, write_stream): # 这里Server会持续运行处理来自Client的请求 await asyncio.Future() # 永久等待 if __name__ __main__: asyncio.run(main())代码关键点解析参数模型使用Pydantic定义工具输入这不仅提供了类型验证和自动文档生成其model_json_schema()方法还能直接生成MCP协议要求的JSON Schema。装饰器注册server.list_tools()和server.call_tool()是注册处理函数的简洁方式。错误处理在divide操作中检查除零错误并通过返回isErrorTrue来告知Client调用失败。传输层StdioServerParameters指定了Server的启动命令。当Client启动时它会执行这个命令并与之通过标准输入输出通信。4.3 测试MCP Server我们可以使用MCP CLI工具来测试Server是否正常工作。首先确保已安装mcp[cli]。创建一个简单的测试脚本test_server.py#!/usr/bin/env python3 import asyncio from mcp import ClientSession, StdioServerParameters import json async def test_server(): # 配置与我们的Server连接 params StdioServerParameters( commandpython, args[mcp_server.py], ) # 创建Client会话 async with ClientSession(params) as session: # 1. 初始化会话 await session.initialize() # 2. 列出可用工具 tools await session.list_tools() print(可用工具:) for tool in tools: print(f - {tool.name}: {tool.description}) # 3. 调用计算器工具 print(\n调用计算器工具 (23 7):) result await session.call_tool( calculator, arguments{a: 23, b: 7, operation: add} ) print(f结果: {result.content[0].text}) # 4. 调用获取时间工具 print(\n调用获取时间工具:) result await session.call_tool( get_current_time, arguments{format: %Y年%m月%d日 %H时%M分} ) print(f结果: {result.content[0].text}) if __name__ __main__: asyncio.run(test_server())运行测试python test_server.py如果一切正常你将看到类似以下输出可用工具: - calculator: 执行简单的四则运算。 - get_current_time: 获取当前的系统时间。 调用计算器工具 (23 7): 结果: 计算结果: 30.0 调用获取时间工具: 结果: 当前时间: 2024年05月27日 14时30分至此一个功能完整的自定义MCP Server已经构建完成。5. 实战二在LangChain中集成MCP Server现在我们将把这个MCP Server提供的工具集成到LangChain的Agent中让大模型能够自主调用它们。5.1 创建LangChain客户端 (client_app.py)我们将使用langchain-mcp-adapters包中的MCPToolkit来简化集成。#!/usr/bin/env python3 LangChain客户端集成自定义MCP Server并构建一个可以调用工具的Agent。 import asyncio import os from typing import List, Optional from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_openai import ChatOpenAI # 使用OpenAI模型也可替换为其他 from langchain_mcp_adapters import MCPToolkit, MCPServer from langchain_core.tools import BaseTool # 注意需要设置你的OpenAI API Key os.environ[OPENAI_API_KEY] your-openai-api-key-here async def create_mcp_tools() - List[BaseTool]: 创建并连接到MCP Server获取其工具列表并转换为LangChain Tool对象。 # 1. 定义MCP Server配置使用stdio连接 server MCPServer.stdio( commandpython, args[mcp_server.py], # env{}, # 可以设置环境变量 # cwd., # 可以设置工作目录 ) # 2. 创建MCP工具包 toolkit MCPToolkit.from_server(server) # 3. 获取工具列表这是一个异步操作 # 注意这里我们直接使用toolkit.tools内部会处理连接和初始化 # 在实际异步环境中可能需要确保连接已建立。 tools toolkit.get_tools() print(f成功从MCP Server加载了 {len(tools)} 个工具:) for tool in tools: print(f - {tool.name}: {tool.description}) return tools def create_agent(tools: List[BaseTool]): 使用加载的工具创建一个LangChain Agent。 # 1. 选择LLM # 使用gpt-3.5-turbo或gpt-4根据实际情况调整 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 2. 构建Agent提示词 # 这是一个标准的工具调用Agent提示词模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个有用的助手可以调用工具来帮助用户解决问题。请根据用户的问题决定是否需要调用工具以及调用哪个工具。如果你知道答案也可以直接回答。), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) # 3. 创建Agent agent create_tool_calling_agent( llmllm, toolstools, promptprompt, ) # 4. 创建Agent执行器 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 开启详细日志方便观察Agent的思考过程 handle_parsing_errorsTrue, # 优雅处理解析错误 max_iterations5, # 限制最大迭代次数防止死循环 ) return agent_executor async def main(): 主函数加载工具创建Agent并运行一个交互循环。 print(正在初始化MCP工具...) try: tools await create_mcp_tools() except Exception as e: print(f连接MCP Server失败: {e}) print(请确保 mcp_server.py 正在运行或路径正确。) return print(\n正在创建LangChain Agent...) agent_executor create_agent(tools) print(\nAgent已就绪你可以输入问题来测试。输入 quit 或 exit 退出。) print(*50) # 简单的交互循环 while True: try: user_input input(\n你的问题: ).strip() if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input: continue # 调用Agent执行 # 注意AgentExecutor的invoke是同步方法但在异步上下文中直接调用。 result await agent_executor.ainvoke({input: user_input}) print(f\n助手: {result[output]}) except KeyboardInterrupt: print(\n\n程序被中断。) break except Exception as e: print(f\n处理过程中出现错误: {e}) if __name__ __main__: # 运行异步主函数 asyncio.run(main())5.2 运行与验证在运行客户端之前请确保你的MCP Server (mcp_server.py) 没有在其他地方运行因为stdio通信要求独占。然后执行python client_app.py程序启动后会先连接MCP Server并加载工具然后进入交互界面。你可以尝试问一些问题输入: “请计算 125 乘以 38 等于多少”预期观察: Agent会识别出需要调用calculator工具并自动填充参数{a: 125, b: 38, operation: multiply}然后输出计算结果。输入: “现在几点了用中文格式告诉我。”预期观察: Agent会调用get_current_time工具并可能尝试使用中文格式字符串最终返回当前时间。在verboseTrue模式下你会在控制台看到详细的思考过程、工具调用和结果这非常有助于调试和理解Agent的工作流。6. 进阶实战集成官方与社区MCP Server自定义Server只是开始MCP的强大之处在于丰富的生态。我们可以轻松集成官方或社区维护的Server为Agent赋予更强大的能力。6.1 集成“文件系统”ServerMCP官方提供了一个基础的filesystemServer允许Agent读写指定目录的文件。首先你需要安装它# 通过pip安装官方Server pip install mcp-server-filesystem然后修改你的client_app.py中的create_mcp_tools函数添加这个Serverasync def create_mcp_tools() - List[BaseTool]: tools [] # 1. 连接我们自定义的Server custom_server MCPServer.stdio( commandpython, args[mcp_server.py], ) custom_toolkit MCPToolkit.from_server(custom_server) tools.extend(custom_toolkit.get_tools()) # 2. 连接官方文件系统Server # 假设我们允许Agent访问当前目录下的一个data文件夹 fs_server MCPServer.stdio( commandnpx, # 使用npx运行Node.js包 args[-y, modelcontextprotocol/server-filesystem, ./data], # 指定目录 ) fs_toolkit MCPToolkit.from_server(fs_server) tools.extend(fs_toolkit.get_tools()) print(f成功从多个MCP Server加载了 {len(tools)} 个工具。) return tools重要安全提示文件系统工具权限很高在生产环境中务必严格限制其可访问的目录如./data并考虑在Server层面进行额外的安全审计和权限控制避免Agent执行危险操作。6.2 集成“天气预报”Server示例假设社区有一个提供天气预报的MCP Server它可能通过SSE提供服务。集成方式如下# 假设天气预报Server运行在 http://localhost:8080/sse weather_server MCPServer.sse(http://localhost:8080/sse) weather_toolkit MCPToolkit.from_server(weather_server) tools.extend(weather_toolkit.get_tools())通过这种方式你可以像搭积木一样将不同来源、不同功能的工具组合起来构建一个能力超强的超级Agent。7. 常见问题与排查思路在集成MCP和LangChain的过程中你可能会遇到以下典型问题问题现象可能原因排查步骤与解决方案启动Client时连接Server失败1. Server脚本路径错误。2. Python环境不一致。3. Server脚本本身有语法错误。1. 检查StdioServerParameters中的command和args。2. 确保Client和Server使用相同的虚拟环境。3. 单独运行python mcp_server.py看是否有报错。Agent无法识别或错误调用工具1. 工具描述(description)不清晰LLM无法理解。2. 参数Schema定义有误或太复杂。3. LLM温度(temperature)过高导致输出不稳定。1. 优化工具描述使用简单、明确的动词开头。2. 使用Pydantic简化参数模型确保JSON Schema正确生成。3. 将temperature设为0或较低值增加确定性。4. 开启verbose模式观察Agent的思考链看是哪一步出了问题。工具调用结果不符合预期1. Server端工具逻辑有bug。2. Agent传递的参数类型错误。3. 网络或资源问题对于访问外部API的工具。1. 首先使用test_server.py这样的脚本直接测试Server确保工具本身正确。2. 检查Server端日志看收到的参数是什么。3. 在Server端增加详细的日志输出便于调试。遇到ModuleNotFoundError: mcpmcp或langchain-mcp-adapters包未正确安装。1. 确认在正确的虚拟环境中。2. 使用pip list | grep mcp和pip list | grep langchain检查包是否存在。3. 尝试重新安装pip install --upgrade mcp[cli] langchain-mcp-adapters。性能问题或响应慢1. 每次调用都重新启动Server进程stdio模式。2. 工具本身执行慢如查询大数据库。3. LLM API调用延迟。1. 考虑使用SSE模式或进程池复用Server连接。2. 在Server端对耗时操作进行优化或缓存。3. 为Agent设置合理的超时(max_execution_time)和最大迭代次数(max_iterations)。如何管理多个Server的依赖每个Server可能依赖不同的Python环境或系统工具。1. 为每个Server使用独立的虚拟环境或容器如Docker。2. 在StdioServerParameters中通过env参数传递特定的环境变量。3. 使用像uv或poetry这样的高级包管理工具管理多项目依赖。8. 最佳实践与工程建议将MCP用于生产级项目时遵循以下最佳实践可以提升稳定性、安全性和可维护性。8.1 工具设计原则单一职责每个工具只做一件事并把它做好。避免创建“瑞士军刀”式的复杂工具。描述清晰工具的description字段是LLM理解其用途的唯一依据。使用“动词宾语”的格式如“计算两个数的和”而非“计算器”。强类型参数充分利用Pydantic定义参数进行类型、范围、枚举值验证。这能减少LLM传参错误并在Server端提供安全保障。防御性编程在Server端工具实现中对所有输入进行校验对可能失败的操作进行异常捕获并返回友好的错误信息给Client。8.2 安全与权限最小权限原则每个MCP Server只应拥有完成其功能所需的最小权限。例如文件系统Server只允许访问特定子目录。输入净化与校验对于执行命令、访问数据库、调用外部API的工具必须严格校验和净化输入防止注入攻击。生产环境隔离考虑在Docker容器中运行MCP Server实现与主应用的环境隔离和资源限制。审计与日志记录所有工具调用的详细信息谁、何时、调用什么、参数、结果便于事后审计和问题排查。8.3 性能与可维护性连接池与长连接对于高频调用的工具避免为每次请求都创建新的Server进程。研究使用SSE长连接或Client端连接池。版本化管理将你的自定义MCP Server作为独立的项目或包进行版本化管理。明确其与MCP协议版本、LangChain版本的兼容性。配置化不要将Server的连接参数如命令、路径、URL硬编码在Client中。使用配置文件或环境变量来管理便于不同环境开发、测试、生产的切换。健康检查为重要的MCP Server实现健康检查端点Client可以在启动时或定期检查Server是否可用。8.4 与LangChain生态结合利用LangGraph对于复杂的工作流可以将MCP工具与LangGraph结合构建有状态、可循环、带条件分支的智能体。工具路由Routing当工具很多时可以考虑实现一个“路由工具”或使用ToolExecutor根据问题类型智能选择最合适的工具而不是让LLM从海量工具中挑选。Fallback机制当某个MCP Server不可用时Client应有降级方案例如使用备用Server、调用本地替代函数或直接告知用户服务暂时不可用。通过本文的讲解和实战你已经掌握了使用MCP协议标准化大模型外部工具连接的核心方法并成功在LangChain中落地集成。从自定义工具开发到生态工具集成再到生产级的最佳实践这套方法论能显著提升你构建AI智能体的效率与能力。接下来你可以探索更多的官方和社区MCP Server或将你团队内部的系统封装成MCP Server逐步构建起一个强大、可复用的AI工具生态。