在构建大模型应用时你是否遇到过这样的困境想让大模型调用数据库查询数据却发现连接和查询逻辑异常复杂且难以复用想集成一个内部API工具却要写大量胶水代码来处理参数解析和结果格式化或者团队开发了多个工具但每个工具与大模型的交互方式都不同导致Agent的开发和维护成本居高不下。如果你正被这些问题困扰那么Model Context Protocol (MCP)就是你一直在寻找的解决方案。本文将带你深入理解 MCP 协议并手把手教你如何将其与 LangChain 集成构建一个标准、可扩展的大模型工具连接通道。无论你是想接入自定义数据源还是编排复杂的工具链本文提供的完整代码和实战案例都能让你快速上手。1. MCP 协议大模型与外部世界的标准接口在深入代码之前我们必须先理解 MCP 协议解决了什么根本问题。1.1 什么是 MCP 协议Model Context Protocol (MCP)即模型上下文协议是一个开放标准旨在为大语言模型LLM与外部工具、数据源之间提供一套统一的、标准化的通信接口。你可以把它想象成大模型世界的USB 协议或HTTP 协议。在没有 MCP 之前每个工具如数据库客户端、API 封装、文件读取器都需要为不同的大模型框架如 LangChain、LlamaIndex编写特定的适配器。这导致了重复劳动同样的工具逻辑需要为不同框架重写。兼容性差工具生态碎片化难以共享和复用。开发复杂开发者需要深入理解每个框架的 Tool/Agent 定义方式。MCP 的核心思想是“一次定义到处运行”。工具提供者只需按照 MCP 标准实现一个 Server任何支持 MCP 的客户端如 LangChain、Claude Desktop、Cursor 等都可以直接发现并使用这些工具无需额外适配。1.2 MCP 的核心组件与工作原理MCP 架构主要包含三个角色MCP Server服务器实际提供工具或数据源能力的后端服务。例如一个提供数据库查询能力的服务或一个提供天气API调用的服务。MCP Client客户端希望使用这些工具的应用或框架。例如LangChain 应用、一个 IDE 插件。MCP Transport传输层定义 Server 和 Client 之间如何通信。目前主要支持stdio标准输入输出和SSE服务器发送事件两种方式其中 stdio 模式对于本地集成最为简单和常见。工作流程简述初始化Client 启动并按照约定如通过命令行参数或配置文件启动或连接到一个 MCP Server。能力协商Client 与 Server 建立连接后Server 会向 Client 宣告自己提供了哪些“工具”Tools和“资源”Resources。工具Tools可以被调用的函数例如query_database,fetch_weather。资源Resources可供读取的静态或动态数据源例如file:///path/to/data.json Client 可以读取这些资源的内容作为上下文。调用与响应Client例如 LangChain Agent根据需要向 Server 发起工具调用请求。Server 执行实际逻辑如运行 SQL、调用 API并将结果返回给 Client。上下文管理Client 也可以请求读取 Resource 的内容将其注入到大模型的提示词中提供更丰富的上下文信息。通过这套协议LangChain 这样的框架就不再需要关心工具的具体实现只需成为一个标准的 MCP Client就能接入整个 MCP 生态中的无数工具。2. 环境准备与项目搭建接下来我们将从零开始构建一个集成 MCP 的 LangChain 应用。我们会先创建一个提供自定义工具的 MCP Server然后创建一个 LangChain Client 来调用它。2.1 环境与工具清单确保你的开发环境满足以下要求操作系统macOS, Linux, 或 Windows (WSL2 推荐)。Python 版本 3.10 LangChain 和 MCP 相关库对 Python 版本有一定要求。包管理工具pip 或 poetry。核心 Python 库langchain-core/langchain: LangChain 核心库。langchain-mcp-adapters: LangChain 官方提供的 MCP 集成适配器。mcp: MCP 协议的 Python SDK用于快速开发 MCP Server。openai: 用于调用大模型 API本文以 OpenAI 兼容 API 为例。2.2 创建项目并安装依赖首先创建一个新的项目目录并初始化虚拟环境。# 创建项目目录 mkdir langchain-mcp-demo cd langchain-mcp-demo # 创建虚拟环境 (可选但强烈推荐) python -m venv venv # 激活虚拟环境 # macOS/Linux: source venv/bin/activate # Windows: # venv\Scripts\activate # 安装核心依赖 pip install langchain-core langchain langchain-mcp-adapters mcp openai如果你的网络环境访问 PyPI 较慢可以考虑使用国内镜像源例如pip install langchain-core langchain langchain-mcp-adapters mcp openai -i https://pypi.tuna.tsinghua.edu.cn/simple安装完成后可以通过pip list检查关键包是否安装成功。2.3 项目结构规划在开始编码前我们先规划一下项目结构这有助于理解后续的代码组织。langchain-mcp-demo/ ├── mcp_server.py # 自定义的 MCP 服务器 ├── langchain_client.py # LangChain 客户端应用 ├── requirements.txt # 项目依赖列表 └── README.md # 项目说明我们将按照这个结构来编写代码。3. 构建你的第一个 MCP ServerMCP Server 是能力提供方。我们将创建一个简单的 Server它提供两个工具一个计算器工具和一个获取系统时间的工具。3.1 编写 MCP Server 代码创建文件mcp_server.py。# mcp_server.py import asyncio from datetime import datetime from mcp import ClientSession, StdioServerParameters from mcp.server import Server from mcp.server.models import Tool import mcp.server.stdio # 创建 MCP Server 实例 server Server(demo-mcp-server) # 1. 定义第一个工具计算器 server.list_tools() async def handle_list_tools() - list[Tool]: 向客户端宣告本 Server 提供的所有工具 return [ Tool( namecalculator, description一个简单的计算器执行基础数学运算。支持加()、减(-)、乘(*)、除(/)。, inputSchema{ type: object, properties: { expression: { type: string, description: 数学表达式例如 3 5 * 2。注意乘号是*除号是/。 } }, required: [expression] } ), Tool( nameget_current_time, description获取服务器当前的日期和时间。, inputSchema{ type: object, properties: { format: { type: string, description: 时间格式字符串例如 %Y-%m-%d %H:%M:%S。如果留空则使用默认格式。, default: %Y-%m-%d %H:%M:%S } }, required: [] # 这个参数不是必须的 } ) ] # 2. 实现工具的执行逻辑 server.call_tool() async def handle_call_tool(name: str, arguments: dict) - list[dict]: 处理客户端发起的工具调用请求 if name calculator: return await handle_calculator(arguments) elif name get_current_time: return await handle_get_current_time(arguments) else: raise ValueError(f未知的工具: {name}) async def handle_calculator(arguments: dict) - list[dict]: 执行计算器逻辑 expression arguments.get(expression, ) if not expression: return [{type: text, text: 错误未提供表达式。}] # 警告在生产环境中直接 eval 是极其危险的这里仅用于演示。 # 真实场景应使用安全的表达式解析库如 ast.literal_eval 或自定义解析器。 try: # 替换常见的数学符号确保 eval 能理解 # 注意这只是简单演示不支持复杂函数或变量。 result eval(expression, {__builtins__: {}}, {}) return [{type: text, text: f表达式 {expression} 的计算结果是: {result}}] except Exception as e: return [{type: text, text: f计算错误: {e}}] async def handle_get_current_time(arguments: dict) - list[dict]: 获取当前时间 format_str arguments.get(format, %Y-%m-%d %H:%M:%S) now datetime.now() formatted_time now.strftime(format_str) return [{type: text, text: f当前服务器时间是: {formatted_time}}] # 3. 主函数启动 Server async def main(): 启动 MCP Server使用 stdio 传输层 # 配置 stdio 传输参数 server_params StdioServerParameters( commandpython, # 解释器命令 args[mcp_server.py], # 脚本参数这里就是它自己 envNone # 环境变量 ) # 使用 stdio 运行 Server这会处理与 Client 的所有通信 async with mcp.server.stdio.stdio_server(server_params) as (read_stream, write_stream): await server.run( read_stream, write_stream, # 这里可以传递 Server 的初始化参数本例中不需要 # server_initialization_parameters... ) if __name__ __main__: asyncio.run(main())代码关键点解析server.list_tools(): 这个装饰器标记的函数用于响应 Client 的“列出所有工具”请求。它返回一个Tool对象的列表每个对象定义了工具的名称、描述和输入参数模式JSON Schema。这是 Client 发现工具能力的依据。server.call_tool(): 这个装饰器标记的函数是工具调用的总入口。它根据传入的name参数将请求分发给具体的工具处理函数如handle_calculator。工具响应格式: 每个工具处理函数返回一个列表列表中的元素是字典必须包含type和content字段。type通常为textcontent就是返回给大模型的文本结果。MCP 协议支持更复杂的返回类型如图片但文本是最通用的。安全警告: 示例中使用了eval()来执行数学表达式这在生产环境是绝对禁止的因为它会执行任意代码造成严重安全漏洞。此处仅用于最简演示。真实项目应使用ast.literal_eval仅限字面量或numexpr、sympy等安全的数学表达式库。传输层: 我们使用stdio_server来启动 Server。这意味着 Server 将通过标准输入stdin接收请求通过标准输出stdout发送响应。这是本地集成最简单的方式。3.2 测试 MCP Server为了验证 Server 是否能正常工作我们可以使用 MCP SDK 自带的 CLI 工具进行测试。首先确保你的mcp库已安装。打开一个终端运行你的 Server 脚本python mcp_server.py此时脚本会启动并等待连接你可能看不到任何输出这是正常的stdio模式在等待客户端连接。打开另一个终端使用mcp命令进行测试# 首先查看有哪些可用命令 mcp --help # 使用 mcp inspect 命令来连接并检查 Server 提供的工具 # 你需要指定传输方式为 stdio并告诉它如何启动我们的 Server mcp inspect stdio --command python --args mcp_server.py如果一切正常mcp inspect命令会输出类似以下的内容显示 Server 提供的工具列表及其输入模式Tools: - calculator: 一个简单的计算器... - get_current_time: 获取服务器当前的日期和时间...这证明你的 MCP Server 已经成功启动并可以对外提供服务了。按CtrlC停止测试。4. 在 LangChain 中集成 MCP Client现在我们已经有了一个功能完整的 MCP Server。下一步就是创建一个 LangChain 应用作为 MCP Client 来调用这些工具。4.1 编写 LangChain Client 代码创建文件langchain_client.py。# langchain_client.py import asyncio import os from typing import List from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_openai import ChatOpenAI from langchain_mcp_adapters import MCPClient, MCPServer # 注意你需要设置你的 OpenAI API 密钥 # 方式1设置环境变量推荐 # export OPENAI_API_KEYyour-api-key-here # 方式2在代码中设置不推荐提交到版本库 # os.environ[OPENAI_API_KEY] your-api-key-here async def main(): 主函数演示 LangChain 如何通过 MCP 协议调用外部工具。 print( 启动 LangChain MCP 客户端 ) # 1. 创建 MCP Server 配置 # 这里我们配置一个本地的 stdio Server指向我们刚才写的 mcp_server.py demo_server MCPServer( namedemo-server, # 给这个 Server 起个名字 # 关键配置指定如何启动 Server 进程 commandpython, args[/绝对路径/到/你的/langchain-mcp-demo/mcp_server.py], # 请替换为你的实际路径 # 也可以使用相对路径但确保工作目录正确 # args[mcp_server.py], ) # 2. 创建 MCP Client并加载 Server # MCPClient 是 LangChain 与 MCP Server 通信的桥梁 async with MCPClient(servers[demo_server]) as client: print(f已连接 MCP Server: {demo_server.name}) # 3. 从 MCP Client 获取工具列表并转换为 LangChain Tool 对象 # 这是关键一步将 MCP 协议定义的工具适配成 LangChain Agent 能识别的格式。 tools await client.get_tools() print(f从 Server 加载了 {len(tools)} 个工具: {[t.name for t in tools]}) # 4. 初始化大语言模型 (LLM) # 使用 ChatOpenAI你也可以替换为其他 LangChain 支持的 LLM llm ChatOpenAI(modelgpt-3.5-turbo-1106, temperature0) # 如果你使用其他兼容 OpenAI API 的服务可以这样配置 # llm ChatOpenAI(base_urlhttps://api.xxx.com/v1, api_keyyour-key, modelyour-model) # 5. 构建 Agent 的提示词模板 # 这个模板告诉 Agent 它的角色、可用的工具以及对话历史的位置。 prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手可以调用工具来帮助用户解决问题。当你需要计算或获取当前时间时请务必调用相应的工具。), (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), # 用于存放 Agent 的思考过程和工具调用记录 ]) # 6. 创建 Agent # create_openai_tools_agent 会创建一个基于 OpenAI Function Calling 格式的 Agent。 # 它将 LLM、提示词和工具绑定在一起。 agent create_openai_tools_agent(llm, tools, prompt) # 7. 创建 Agent 执行器 # AgentExecutor 负责运行 Agent处理工具调用循环直到得到最终答案。 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) # 8. 运行测试用例 test_queries [ “请计算一下 (15 7) * 3 等于多少”, “现在几点了请告诉我格式如 ‘2024-01-01 12:00:00’。”, “先获取当前时间然后计算从今天零点到现在过了多少秒提示你需要先获取时间然后手动计算或再调用一次计算器” ] for query in test_queries: print(f\n--- 用户提问: {query} ---) try: # 调用 Agent 执行器处理查询 result await agent_executor.ainvoke({input: query}) print(fAgent 最终回答: {result[output]}) except Exception as e: print(f执行过程中出现错误: {e}) print(\n 演示结束 ) if __name__ __main__: # 运行异步主函数 asyncio.run(main())代码关键点解析MCPServer配置: 这是配置如何启动/连接 MCP Server 的核心。command和args指定了启动 Server 进程的命令就像在命令行中运行python mcp_server.py一样。务必替换args中的路径为你本地mcp_server.py的实际绝对路径这是最常见的错误来源。MCPClient上下文管理器: 使用async with来管理 Client 的生命周期。在进入上下文时它会启动配置的所有 Server 进程并建立连接在退出时会优雅地关闭连接和进程。client.get_tools(): 这是魔法发生的地方。这个方法会与 MCP Server 通信获取其声明的所有工具列表并自动将它们转换成 LangChain 原生的Tool对象。这样这些工具就可以像任何其他 LangChain Tool 一样被 Agent 使用。Agent 构建流程: 这是标准的 LangChain Agent 构建模式准备 LLM - 准备提示词 - 准备工具 - 组装成 Agent - 放入执行器。MCP 工具在此流程中无缝集成。异步调用: 由于 MCP 通信和 LLM 调用通常是 I/O 密集型的我们使用异步函数 (async def) 和ainvoke来提高效率。确保主入口使用asyncio.run(main())。4.2 运行与验证在运行 Client 之前请确保已经正确设置了OPENAI_API_KEY环境变量。已经将langchain_client.py中MCPServer的args路径修改正确。然后在项目根目录下运行python langchain_client.py预期输出verbose模式会显示详细思考过程 启动 LangChain MCP 客户端 已连接 MCP Server: demo-server 从 Server 加载了 2 个工具: [calculator, get_current_time] --- 用户提问: 请计算一下 (15 7) * 3 等于多少 --- 进入新的 Agent 执行链... 我可能需要计算这个表达式。我有一个计算器工具。 动作调用工具 calculator 动作输入{expression: (15 7) * 3} 观察表达式 (15 7) * 3 的计算结果是: 66 思考我得到了计算结果。 最终答案66 Agent 最终回答: 66 --- 用户提问: 现在几点了请告诉我格式如 ‘2024-01-01 12:00:00’。 --- 进入新的 Agent 执行链... 用户想知道当前时间并且指定了格式。我有一个获取时间的工具。 动作调用工具 get_current_time 动作输入{format: %Y-%m-%d %H:%M:%S} 观察当前服务器时间是: 2024-05-27 14:30:25 思考我已经获取了当前时间。 最终答案当前时间是 2024-05-27 14:30:25。 Agent 最终回答: 当前时间是 2024-05-27 14:30:25。你会看到 LangChain Agent 成功识别了用户意图自动选择了正确的 MCP 工具calculator或get_current_time并输出了正确结果。这标志着 LangChain 与 MCP Server 的集成成功5. 进阶实战接入真实数据源SQLite数据库上面的例子演示了基础工具集成。现在我们来解决一个更实际的场景让大模型通过 MCP 查询数据库。我们将创建一个新的 MCP Server提供 SQLite 数据库查询工具。5.1 创建数据库与示例数据首先创建一个新的 Python 脚本create_db.py来初始化数据库。# create_db.py import sqlite3 # 连接数据库如果不存在则创建 conn sqlite3.connect(example.db) cursor conn.cursor() # 创建用户表 cursor.execute( CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, email TEXT NOT NULL UNIQUE, age INTEGER, department TEXT ) ) # 插入示例数据 users_data [ (张三, zhangsanexample.com, 28, 技术部), (李四, lisiexample.com, 34, 市场部), (王五, wangwuexample.com, 25, 技术部), (赵六, zhaoliuexample.com, 41, 人事部), ] cursor.executemany(INSERT INTO users (name, email, age, department) VALUES (?, ?, ?, ?), users_data) # 创建订单表 cursor.execute( CREATE TABLE IF NOT EXISTS orders ( order_id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, product_name TEXT NOT NULL, amount REAL NOT NULL, order_date TEXT, FOREIGN KEY (user_id) REFERENCES users (id) ) ) # 插入示例订单数据 orders_data [ (1, 笔记本电脑, 6999.99, 2024-05-20), (1, 鼠标, 89.50, 2024-05-21), (2, 办公椅, 1200.00, 2024-05-18), (3, 键盘, 450.00, 2024-05-22), (4, 显示器, 1999.00, 2024-05-19), ] cursor.executemany(INSERT INTO orders (user_id, product_name, amount, order_date) VALUES (?, ?, ?, ?), orders_data) # 提交更改并关闭连接 conn.commit() conn.close() print(数据库 example.db 及示例数据已创建成功)运行这个脚本python create_db.py这将在当前目录下生成一个example.db文件。5.2 编写数据库 MCP Server创建文件mcp_server_sql.py。这个 Server 将提供一个安全的数据库查询工具。# mcp_server_sql.py import asyncio import sqlite3 import json from typing import Any from mcp import ClientSession, StdioServerParameters from mcp.server import Server from mcp.server.models import Tool import mcp.server.stdio # 数据库文件路径 DB_PATH example.db server Server(sqlite-mcp-server) def execute_sql_query(sql: str, parameters: tuple ()) - list[dict[str, Any]]: 执行 SQL 查询并返回结果列表字典格式 conn sqlite3.connect(DB_PATH) # 设置 row_factory 以返回字典 conn.row_factory sqlite3.Row cursor conn.cursor() try: cursor.execute(sql, parameters) rows cursor.fetchall() # 将 Row 对象转换为字典列表 result [dict(row) for row in rows] return result except sqlite3.Error as e: # 返回错误信息 return [{error: fSQL 执行错误: {e}}] finally: conn.close() server.list_tools() async def handle_list_tools() - list[Tool]: return [ Tool( namequery_database, description执行安全的 SQLite 查询。可以查询用户表(users)和订单表(orders)。请提供清晰、准确的 SQL SELECT 语句。, inputSchema{ type: object, properties: { sql: { type: string, description: 要执行的 SQL SELECT 查询语句。例如SELECT * FROM users WHERE department ? }, parameters: { type: array, items: {type: string}, description: SQL 查询中的参数值列表用于防止 SQL 注入。例如[技术部], default: [] } }, required: [sql] } ) ] server.call_tool() async def handle_call_tool(name: str, arguments: dict) - list[dict]: if name query_database: return await handle_query_database(arguments) else: raise ValueError(f未知的工具: {name}) async def handle_query_database(arguments: dict) - list[dict]: sql arguments.get(sql, ).strip() params arguments.get(parameters, []) # 基础安全校验只允许 SELECT 查询 if not sql.upper().startswith(SELECT): return [{type: text, text: 错误此工具仅支持 SELECT 查询语句以保障数据安全。}] # 执行查询 results execute_sql_query(sql, tuple(params)) if results and error in results[0]: # 返回错误信息 return [{type: text, text: results[0][error]}] else: # 将结果格式化为易读的字符串 if not results: output_text 查询成功但未返回任何数据。 else: # 使用 json.dumps 美化输出indent2 使格式更友好 output_text f查询成功返回 {len(results)} 条记录\njson\n{json.dumps(results, indent2, ensure_asciiFalse)}\n return [{type: text, text: output_text}] async def main(): server_params StdioServerParameters( commandpython, args[mcp_server_sql.py], ) async with mcp.server.stdio.stdio_server(server_params) as (read_stream, write_stream): await server.run(read_stream, write_stream) if __name__ __main__: asyncio.run(main())安全与设计要点SQL 注入防护工具设计为接受sql和parameters两个参数使用参数化查询 (cursor.execute(sql, parameters)) 来从根本上防止 SQL 注入。权限限制在handle_query_database函数中我们通过sql.upper().startswith(SELECT)强制只允许SELECT语句禁止INSERT,UPDATE,DELETE等写操作。这是生产环境中保护数据的重要措施。结果格式化将数据库结果字典列表通过json.dumps(indent2)格式化为美观的 JSON 字符串并包裹在 Markdown 代码块中便于大模型和开发者阅读。错误处理捕获sqlite3.Error并返回友好的错误信息而不是让整个 Server 崩溃。5.3 更新 LangChain Client 以使用数据库 Server现在修改langchain_client.py或新建一个文件使其连接到新的数据库 Server。# langchain_client_db.py import asyncio from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_openai import ChatOpenAI from langchain_mcp_adapters import MCPClient, MCPServer async def main(): print( 启动 LangChain MCP 数据库查询客户端 ) # 配置数据库 MCP Server sql_server MCPServer( namesqlite-server, commandpython, args[/绝对路径/到/你的/langchain-mcp-demo/mcp_server_sql.py], # 替换路径 ) async with MCPClient(servers[sql_server]) as client: tools await client.get_tools() print(f从 Server 加载了 {len(tools)} 个工具: {[t.name for t in tools]}) llm ChatOpenAI(modelgpt-3.5-turbo-1106, temperature0) # 更新提示词引导 Agent 使用数据库工具 prompt ChatPromptTemplate.from_messages([ (system, 你是一个数据分析助手可以访问公司的 SQLite 数据库。 数据库中有 users 表字段id, name, email, age, department和 orders 表字段order_id, user_id, product_name, amount, order_date。 当用户询问关于用户、订单或部门的数据时你需要编写正确的 SQL SELECT 语句来查询数据库。 请确保 SQL 语句语法正确并使用参数化查询来防止错误。 如果用户的问题无法通过查询数据库回答请如实告知。), (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) agent create_openai_tools_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) # 测试数据库查询 test_queries [ “技术部有哪些员工列出他们的名字和邮箱。”, “查询所有订单的总金额。”, “找出年龄大于30岁的用户并看看他们下了哪些订单。” ] for query in test_queries: print(f\n--- 用户提问: {query} ---) try: result await agent_executor.ainvoke({input: query}) print(fAgent 最终回答: {result[output]}) except Exception as e: print(f执行过程中出现错误: {e}) print(\n 演示结束 ) if __name__ __main__: asyncio.run(main())运行这个客户端python langchain_client_db.py预期输出片段 进入新的 Agent 执行链... 用户想了解技术部的员工信息。我需要查询 users 表。 动作调用工具 query_database 动作输入{sql: SELECT name, email FROM users WHERE department ?, parameters: [技术部]} 观察查询成功返回 2 条记录 json [ { name: 张三, email: zhangsanexample.com }, { name: 王五, email: wangwuexample.com } ]思考我已经获取了技术部员工的姓名和邮箱。 最终答案技术部共有2名员工张三zhangsanexample.com和王五wangwuexample.com。至此你已经成功构建了一个可以通过自然语言查询数据库的智能助手大模型自动将你的问题转换成了参数化的 SQL 查询并通过 MCP 协议安全地执行。 ## 6. 常见问题与排查思路 在集成 MCP 与 LangChain 的过程中你可能会遇到一些典型问题。下面是一个快速排查指南。 | 问题现象 | 可能原因 | 解决思路 | | :--- | :--- | :--- | | **运行 Client 时报错 FileNotFoundError 或 Permission denied** | MCPServer 配置中的 command 或 args 路径不正确。 | 1. 检查 args 中的 Python 脚本路径是否为**绝对路径**。br2. 确保该路径下的文件存在且有执行权限。br3. 尝试在终端中手动运行 python /path/to/your/mcp_server.py 看是否能启动。 | | **Client 启动后卡住无任何输出** | 1. MCP Server 进程启动失败或崩溃。br2. Server 和 Client 之间的 stdio 通信出现问题。 | 1. 单独运行 Server 脚本 (python mcp_server.py)看是否有 Python 语法错误。br2. 检查 Server 代码中是否有同步的 input() 或导致进程阻塞的操作。br3. 确保使用 async with MCPClient(...) 来正确管理生命周期。 | | **client.get_tools() 返回空列表或报错** | 1. Server 未正确实现 list_tools 接口。br2. 传输层协议不一致。 | 1. 使用 mcp inspect 命令测试 Server 是否能正常返回工具列表。br2. 检查 Server 代码中 server.list_tools() 装饰的函数是否正确返回了 Tool 对象列表。 | | **Agent 无法正确调用工具或调用参数错误** | 1. 工具定义的 inputSchema 与大模型生成的参数不匹配。br2. 提示词Prompt未清晰指导 Agent 使用工具。 | 1. 在工具的 description 和 inputSchema 的 description 字段中尽可能详细地描述工具功能和参数格式。br2. 优化 System Prompt明确告诉 Agent 在什么情况下使用哪个工具以及参数应该如何构造。br3. 开启 verboseTrue 查看 Agent 的思考链检查是哪一步出了问题。 | | **数据库查询工具执行失败** | 1. SQL 语法错误。br2. 数据库文件路径错误或权限不足。br3. 违反了安全规则如尝试执行非 SELECT 语句。 | 1. 查看 Server 返回的错误信息。br2. 检查 DB_PATH 变量指向的数据库文件是否存在。br3. 在 Server 代码中增加更详细的日志打印接收到的 SQL 和参数。 | | **ModuleNotFoundError: No module named mcp** | mcp 或 langchain-mcp-adapters 包未安装。 | 使用 pip install mcp langchain-mcp-adapters 确保安装成功。注意langchain-mcp-adapters 可能需要较新版本的 langchain-core。 | ## 7. 最佳实践与工程建议 将 MCP 用于生产环境或复杂项目时请遵循以下建议 ### 7.1 MCP Server 设计原则 1. **单一职责**一个 MCP Server 应专注于一类功能。例如database-mcp-server 只处理数据库操作weather-mcp-server 只处理天气查询。这便于维护和部署。 2. **安全性第一** * **输入验证与消毒**对所有来自 Client 的输入进行严格的验证和消毒防止注入攻击。 * **最小权限原则**Server 进程应以最低必要的系统权限运行。数据库 Server 应使用只有读取权限的数据库用户。 * **操作限制**像数据库 Server 示例一样明确禁止危险操作如 DROP, DELETE。 3. **健壮性与错误处理** * Server 内部应有完善的 try...except 块捕获异常并返回结构化的错误信息给 Client而不是让进程崩溃。 * 考虑添加请求超时机制防止长时间运行的操作阻塞 Client。 4. **清晰的文档**在工具的 description 和参数的 description 字段中提供清晰、无歧义的说明这能极大提高大模型调用工具的准确率。 ### 7.2 LangChain Client 集成建议 1. **配置管理**不要将 Server 的启动命令如 Python 路径、脚本路径硬编码在代码中。应使用配置文件如 YAML、JSON或环境变量来管理便于在不同环境开发、测试、生产中切换。 2. **资源管理**务必使用 async with MCPClient(...) as client: 上下文管理器确保 Server 进程在 Client 退出时被正确清理避免僵尸进程。 3. **工具过滤与包装**client.get_tools() 返回所有工具。你可能需要根据当前 Agent 的职责对工具列表进行过滤或重命名避免工具冲突或误导 Agent。 4. **提示词工程**System Prompt 是指导 Agent 正确使用 MCP 工具的关键。详细描述每个工具的用途、适用场景和参数格式。你可以考虑在 Prompt 中动态插入当前可用的工具列表及其描述。 ### 7.3 部署与运维 1. **进程管理**在生产环境中MCP Server 可能需要作为常驻服务运行而不是由 Client 每次启动。可以考虑使用 sse 传输模式将 Server 部署为 HTTP 服务Client 通过 URL 连接。这提供了更好的可管理性和可扩展性。 2. **监控与日志**为 MCP Server 和 LangChain Client 添加详细的日志记录包括工具调用请求、参数、执行结果和耗时。这对于调试和性能分析至关重要。 3. **版本兼容性**MCP 协议和 langchain-mcp-adapters 仍在快速发展中。锁定你的依赖版本在 requirements.txt 中使用 并在升级时仔细测试。 ### 7.4 扩展方向 1. **连接更多工具**探索 MCP 社区已有的 Server 实现例如 * 文件系统操作 * Git 仓库查询 * JIRA/Confluence 等办公软件 * 内部业务系统 API 2. **构建复杂工作流**结合 LangGraph 或 LangChain 的 Plan-and-Execute 模式利用多个 MCP 工具编排复杂、多步骤的任务。 3. **开发可视化界面**基于 LangChain 和 MCP你可以快速构建一个聊天界面让非技术人员也能通过自然语言安全地查询数据库、生成报告等。 通过本文的实战演练你已经掌握了使用 MCP 协议为 LangChain 应用扩展能力的核心方法。从理解协议原理到亲手编写 MCP Server 提供计算器和时间查询工具再到集成安全的数据库查询能力这条路径清晰地展示了如何通过标准化协议打破大模型与外部系统之间的壁垒。记住MCP 的核心价值在于“标准化”和“解耦”。开始将你项目中的内部工具改造成 MCP Server 吧你会发现 Agent 的构建和维护工作将变得前所未有的清晰和高效。如果在实践中遇到问题回顾第 6 节的排查思路并牢记第 7 节的最佳实践这将帮助你构建出稳定、安全、强大的智能应用。