1. 项目概述为什么我们需要重新思考LLM与工具的关系最近在折腾大语言模型应用落地的朋友估计都绕不开一个词Agent。无论是想做个能自动处理邮件的助手还是想搞个能分析数据的智能体最终都得让LLM学会“使用工具”。但真上手了你会发现这事儿远没想象中那么简单。传统的Agent工具集成方式就像把一台高性能发动机硬塞进一辆老式马车的底盘——动力是有了但跑起来浑身别扭动不动就散架。我花了大量时间在几个实际项目中从简单的天气查询到复杂的企业级数据管道调度几乎把能踩的坑都踩了一遍。最终我意识到问题的核心在于耦合度。目前主流的做法无论是LangChain的Tool定义还是LlamaIndex的Function Calling本质上都是将工具的实现逻辑、调用协议乃至错误处理与LLM的提示词工程和推理流程深度捆绑在一起。这直接导致了两个让我头疼不已的痛点痛点一语言绑定的僵局。我的工具库是用Python写的性能好、生态丰富。但业务团队可能更熟悉Go或者Java他们想复用我的工具逻辑或者前端同事想用JavaScript直接调用几乎不可能。每次都要为了适配新的调用方用目标语言重写一遍工具逻辑这不仅是重复劳动更是维护的噩梦。一个工具逻辑的更新需要同步更新多个语言版本的实现一致性根本无法保证。痛点二LLM与工具的生死与共。在传统架构下工具服务的稳定性直接决定了整个Agent系统的可用性。如果工具服务因为网络、依赖或自身bug挂掉LLM的调用链会立刻中断返回一个令人沮丧的错误。更糟糕的是错误信息往往难以被LLM理解和处理导致整个对话流程卡死。你不得不花大量精力去构建健壮的错误处理和中途拦截机制这部分的代码复杂度甚至超过了业务逻辑本身。所以当我在社区里看到MCPModel Context Protocol这个协议开始被讨论时眼前顿时一亮。它的核心思想非常直接在LLM和工具之间建立一个标准化的、与语言无关的通信层。工具提供者只需按照协议暴露能力LLM或者说Agent框架只需按照协议去发现和调用双方不再需要关心对方是用什么语言实现的。于是我决定动手验证一下这个想法。目标很明确用尽可能少的代码实现一个MCP Server将我的Python工具库暴露出去同时再实现一个极简的MCP Client让LLM能通过这个Client调用工具。整个过程下来核心代码真的控制在了300行左右。结果令人振奋不仅成功解耦整个系统的可维护性和扩展性都上了一个台阶。下面我就把这套方案的完整设计思路、实操步骤以及踩坑心得毫无保留地分享出来。2. MCP协议核心思想与架构拆解在动手写代码之前我们必须先吃透MCP到底要解决什么问题以及它是如何设计的。你可以把它想象成计算机硬件里的USB协议。在USB出现之前鼠标、键盘、打印机各有各的接口你需要不同的线缆、不同的驱动插错了口甚至可能烧坏设备。USB协议出现后定义了一套标准的物理接口、电气信号和通信规范。从此设备制造商只需生产符合USB标准的设备电脑主板只需提供USB接口双方就能即插即用背后的复杂驱动和协商过程都被协议层消化了。MCP在LLM与工具的世界里扮演的正是这个“USB协议”的角色。它的设计目标非常清晰标准化Standardization定义工具如何向LLM描述自己名称、描述、参数格式以及LLM如何调用工具请求格式、响应格式。这解决了“工具描述语言”不统一的问题。解耦Decoupling工具的实现Server端和LLM对工具的调用Client端完全分离。工具可以用任何语言编写运行在任何地方LLM Agent也可以通过任何支持MCP协议的Client来调用这些工具。它们之间只通过标准的协议消息进行通信。可发现性DiscoverabilityClient能够动态地发现Server提供了哪些工具并获取其完整的调用规范无需提前硬编码。这使得工具的热插拔成为可能。2.1 MCP的核心组件与交互流程一个完整的MCP架构通常包含三个角色MCP Server工具提供方它封装了具体的工具逻辑例如“查询数据库”、“发送邮件”、“生成图表”。Server启动后会在一个网络端点如HTTP端口或Stdio等待连接。MCP Client工具调用方它代表LLM或Agent框架负责与Server建立连接获取工具列表并代表LLM发起工具调用。Client不包含任何具体的工具逻辑。传输层Transport负责在Server和Client之间传递标准的JSON-RPC消息。主流支持两种方式标准输入输出Stdio和HTTP。Stdio模式通常用于本地紧密集成的场景如一个CLI工具而HTTP模式则适用于跨网络、分布式的部署。它们之间的交互可以概括为以下几个核心步骤初始化连接InitializeClient向Server发起连接并交换各自的元信息如名称、版本、支持的能力。列出工具ListToolsClient调用tools/list方法Server返回一个工具描述列表。每个描述都包含工具的唯一名称、详细的功能描述、以及调用所需的参数JSON Schema。这是LLM理解工具能力的关键。调用工具CallTool当LLM决定使用某个工具时Client会向Server发起tools/call请求携带工具名和具体的参数。Server执行实际逻辑。返回结果Server执行完毕后将结果或错误信息封装成标准格式返回给Client。Client再将其呈现给LLM用于后续的推理或回答生成。整个过程中Client和Server都不需要知道对方的实现细节。Client不在乎工具是用Python还是Go写的Server也不在乎调用它的是LangChain Agent还是自定义的脚本。2.2 为什么MCP能根治两大痛点现在让我们回到开头的两个痛点看看MCP是如何解决的针对“语言绑定”痛点MCP Server可以用任何语言实现只要它遵循协议发送和接收JSON-RPC消息。这意味着你可以用Python写一个高性能的数据处理工具Server同时用Go写一个系统管理工具Server。你的LLM Agent通过MCP Client可以同时无缝调用它们无需任何语言适配层。团队协作时后端用Java算法用Python前端用JS都可以各自维护自己的MCP Server最终在Agent层面统一集成。针对“耦合故障”痛点由于通信是标准化的错误也被纳入了协议规范。工具Server可以返回结构化的错误信息如错误码、类型、详情。MCP Client可以统一处理这些错误例如进行重试、降级处理或者将友好的错误信息反馈给LLM让LLM决定下一步动作比如建议用户检查输入。更重要的是某个工具的故障甚至Server崩溃通常不会导致Client或LLM进程崩溃因为它们之间是松耦合的网络/进程间通信。你可以方便地实现Server的健康检查、熔断和负载均衡。理解了这些我们就能明白实现MCP的关键不在于复杂的业务逻辑而在于正确地实现协议规定的几个核心JSON-RPC方法。接下来我们就进入实战环节。3. 300行代码实现MCP Server与Client详解我们的目标是构建一个最小可行系统。假设我们有一个简单的“天气查询”工具和一个“单位换算”工具我们将用Python实现它们的MCP Server同时实现一个简单的MCP Client来演示调用。我们将选择Stdio传输层因为它最简单无需处理网络问题适合本地集成和演示。3.1 环境准备与依赖选择首先创建一个新的项目目录。我们不需要重量级的框架Python标准库的json和subprocess以及sys就足够了。但为了更规范地处理JSON-RPC和协议细节我们使用一个轻量级的库mcp。这是一个低级别的、用于构建MCP组件的Python SDK。# 创建项目目录并初始化虚拟环境推荐 mkdir mcp-demo cd mcp-demo python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate # 安装核心依赖 pip install mcpmcp库提供了编写Server和Client所需的底层抽象帮我们处理了协议消息的序列化/反序列化、生命周期管理等样板代码让我们能专注于工具逻辑本身。3.2 实现MCP Server约150行我们在项目根目录下创建server.py。#!/usr/bin/env python3 一个简单的MCP Server示例提供天气查询和单位换算工具。 使用Stdio传输。 import asyncio import json import sys from typing import Any, List from mcp import ClientSession, StdioServerParameters from mcp.server import Server from mcp.server.models import Tool import mcp.server.stdio # 模拟一个简单的天气查询函数 async def get_weather(city: str) - str: 根据城市名返回模拟天气信息。 # 这里应该是调用真实API例如OpenWeatherMap # 为了演示我们返回模拟数据 weather_data { 北京: 晴25°C北风2级, 上海: 多云28°C东南风3级, 深圳: 雷阵雨30°C南风1级, } return weather_data.get(city, f未找到{city}的天气信息。) # 模拟一个单位换算函数 async def convert_units(value: float, from_unit: str, to_unit: str) - str: 进行简单的单位换算。 conversions { (km, mile): lambda v: v * 0.621371, (mile, km): lambda v: v * 1.60934, (kg, lb): lambda v: v * 2.20462, (lb, kg): lambda v: v * 0.453592, (celsius, fahrenheit): lambda v: (v * 9/5) 32, (fahrenheit, celsius): lambda v: (v - 32) * 5/9, } key (from_unit.lower(), to_unit.lower()) if key in conversions: result conversions[key](value) return f{value} {from_unit} {result:.2f} {to_unit} else: return f不支持从 {from_unit} 到 {to_unit} 的换算。 async def main(): # 1. 创建MCP Server实例 server Server(demo-tools-server) # 2. 向Server注册工具 # 每个Tool对象定义了工具的名称、描述和输入参数模式 server.list_tools() async def handle_list_tools() - List[Tool]: return [ Tool( nameget_weather, description根据给定的城市名称查询该城市的当前天气情况。, inputSchema{ type: object, properties: { city: { type: string, description: 城市名称例如北京、上海、New York } }, required: [city] } ), Tool( nameconvert_units, description进行常见的单位换算例如公里与英里、公斤与磅、摄氏度与华氏度之间的转换。, inputSchema{ type: object, properties: { value: {type: number, description: 需要换算的数值}, from_unit: {type: string, description: 原始单位例如km, mile, kg, lb, celsius, fahrenheit}, to_unit: {type: string, description: 目标单位} }, required: [value, from_unit, to_unit] } ) ] # 3. 绑定工具名称到具体的处理函数 server.call_tool() async def handle_call_tool(name: str, arguments: dict) - list: if name get_weather: city arguments.get(city) if not city: raise ValueError(参数 city 是必需的。) result_text await get_weather(city) # 返回格式需符合MCP协议content是一个列表 return [{ type: text, text: result_text }] elif name convert_units: value arguments.get(value) from_unit arguments.get(from_unit) to_unit arguments.get(to_unit) if None in (value, from_unit, to_unit): raise ValueError(参数 value, from_unit, to_unit 都是必需的。) result_text await convert_units(float(value), from_unit, to_unit) return [{ type: text, text: result_text }] else: raise ValueError(f未知的工具: {name}) # 4. 使用Stdio传输层运行Server # 这允许通过标准输入输出与Client通信 async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, # 可选的Server初始化信息 server_info{ name: Demo Tools Server, version: 0.1.0 } ) if __name__ __main__: asyncio.run(main())代码关键点解析工具定义Tool对象这是LLM理解工具的“说明书”。description字段至关重要LLM会根据它来判断在什么场景下使用这个工具。inputSchema必须严格按照JSON Schema格式定义它规定了调用时必须传入哪些参数以及参数的类型和格式。清晰的Schema能极大减少LLM调用出错的概率。工具实现函数get_weather和convert_units是实际执行业务逻辑的函数。它们可以是同步或异步的可以调用任何其他库或服务。这里为了演示用了模拟数据。请求路由handle_call_tool这个装饰器函数是Server的核心路由器。它根据传入的name找到对应的工具并从arguments字典中提取参数调用真正的业务函数。Stdio传输mcp.server.stdio.stdio_server()创建了基于标准输入输出的通信通道。这意味着我们的Server可以作为一个独立的命令行进程启动Client通过管道与之通信。这是最简单、最轻量的集成方式。注意在实际生产环境中get_weather函数应该调用真实的天气API并做好错误处理如网络超时、API限流、无效城市名等。这些错误应该在函数内部捕获并转化为用户或LLM可理解的错误信息通过MCP协议返回。3.3 实现MCP Client约100行接下来我们创建一个client.py它代表LLM或Agent框架负责与Server对话。#!/usr/bin/env python3 一个简单的MCP Client示例演示如何发现并调用Server提供的工具。 import asyncio import json import subprocess import sys from typing import Any, Dict, List from mcp import ClientSession from mcp.client.stdio import stdio_client async def main(): # 1. 启动MCP Server进程并建立Stdio连接 # 这里我们启动刚才写的server.py进程 server_process subprocess.Popen( [sys.executable, server.py], # 使用当前Python解释器运行server.py stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsys.stderr, # 将Server的错误输出到控制台便于调试 textTrue ) # 2. 创建MCP Client会话连接到Server进程的输入输出流 async with stdio_client(server_process.stdin, server_process.stdout) as (read_stream, write_stream): async with ClientSession(read_stream, write_stream) as session: # 3. 初始化连接与Server握手 await session.initialize() # 4. 列出Server提供的所有工具 print(正在从Server发现可用工具...) tools await session.list_tools() print(f发现 {len(tools)} 个工具:) for tool in tools: print(f - {tool.name}: {tool.description}) # 可以打印详细的参数模式供LLM或开发者参考 # print(f 参数模式: {json.dumps(tool.inputSchema, indent2, ensure_asciiFalse)}) # 5. 模拟LLM决策过程根据用户请求选择工具并调用 # 假设LLM分析了用户输入“北京天气怎么样”后决定调用get_weather user_query 北京天气怎么样 print(f\n用户查询: {user_query}) print(LLM决策: 调用 get_weather 工具) tool_name get_weather arguments {city: 北京} print(f调用工具: {tool_name} 参数: {arguments}) try: # 执行工具调用 response await session.call_tool(tool_name, arguments) # response是一个Content对象列表我们取第一个文本内容 if response and response.contents: tool_result response.contents[0].text print(f工具返回结果: {tool_result}) # 在实际Agent中这个结果会被喂回给LLM用于生成最终回答 final_answer f根据查询北京的天气情况是{tool_result} print(fAgent最终回答: {final_answer}) else: print(工具调用未返回有效内容。) except Exception as e: print(f工具调用失败: {e}) # 6. 再演示一个调用 print(\n--- 第二个例子 ---) user_query 把10公里换算成英里 print(f用户查询: {user_query}) print(LLM决策: 调用 convert_units 工具) tool_name convert_units arguments {value: 10, from_unit: km, to_unit: mile} print(f调用工具: {tool_name} 参数: {arguments}) try: response await session.call_tool(tool_name, arguments) if response and response.contents: tool_result response.contents[0].text print(f工具返回结果: {tool_result}) final_answer f换算结果是{tool_result} print(fAgent最终回答: {final_answer}) else: print(工具调用未返回有效内容。) except Exception as e: print(f工具调用失败: {e}) # 7. 会话结束等待Server进程退出 server_process.wait() print(\nClient会话结束。) if __name__ __main__: asyncio.run(main())代码关键点解析启动Server进程Client使用subprocess.Popen启动server.py并捕获其标准输入和输出。这模拟了将MCP Server作为一个独立子进程管理的场景。建立会话stdio_client和ClientSession管理了与Server的协议层通信包括初始化握手、消息发送和接收。工具发现session.list_tools()是Client获取Server能力目录的方法。在一个动态环境中Agent可以在启动时或定期调用此方法以感知工具的变化。工具调用session.call_tool()是核心调用方法。Client只需要知道工具名和符合Schema的参数即可完全无需了解工具的内部实现。返回的response对象结构是标准的包含了工具执行产生的文本、图片或其他类型的内容。错误处理调用被try...except包裹。MCP协议允许Server返回结构化的错误Client可以根据错误类型决定重试、降级或通知用户。3.4 运行演示在终端中确保处于虚拟环境然后直接运行Clientpython client.py你将看到类似以下的输出正在从Server发现可用工具... 发现 2 个工具: - get_weather: 根据给定的城市名称查询该城市的当前天气情况。 - convert_units: 进行常见的单位换算例如公里与英里、公斤与磅、摄氏度与华氏度之间的转换。 用户查询: 北京天气怎么样 LLM决策: 调用 get_weather 工具 调用工具: get_weather 参数: {city: 北京} 工具返回结果: 晴25°C北风2级 Agent最终回答: 根据查询北京的天气情况是晴25°C北风2级 --- 第二个例子 --- 用户查询: 把10公里换算成英里 LLM决策: 调用 convert_units 工具 调用工具: convert_units 参数: {value: 10, from_unit: km, to_unit: mile} 工具返回结果: 10.0 km 6.21 mile Agent最终回答: 换算结果是10.0 km 6.21 mile Client会话结束。至此一个完整的、解耦的MCP工具调用流程就完成了。整个核心通信逻辑Server Client的代码量正如标题所说在300行左右。4. 从Demo到生产关键配置与进阶实践上面的Demo跑通了基本流程但要想投入实际使用还有几个关键环节需要加固和优化。这部分才是体现工程经验价值的地方。4.1 传输层选择Stdio vs. HTTP (SSE)我们的Demo使用了Stdio它简单直接适合本地集成比如你的Agent应用和工具Server部署在同一台机器上或者作为单个应用的插件系统。它的生命周期管理也简单Client启动ServerClient退出时Server通常也结束。但对于微服务架构或远程工具HTTP通常基于Server-Sent Events, SSE是更合适的选择。MCP over HTTP允许你将工具Server部署在独立的容器或服务器上通过网络提供服务。多个Client可以连接同一个Server实现工具能力的共享和复用。如何切换为HTTP传输在Server端你需要一个支持SSE的HTTP框架比如FastAPI。核心是暴露两个端点GET /sse用于建立SSE长连接持续接收Client的请求。POST /message用于Client向Server发送具体的JSON-RPC请求。在Client端不再启动子进程而是直接连接到Server的HTTP URL。实操心得对于内部系统Stdio模式因其零网络延迟和简单的权限管理继承父进程权限而更高效。对于需要跨团队、跨网络共享的工具服务HTTP模式是必然选择。许多成熟的MCP SDK包括Pythonmcp库的高层API都同时支持两种模式切换通常只需更改几行配置代码。4.2 工具描述的“艺术”编写LLM友好的Schema工具能否被LLM正确调用一半取决于Tool对象中的description和inputSchema。写得好LLM调用精准写得差LLM要么不用要么乱用。优化技巧描述要具体且包含关键词不要写“查询天气”要写“根据给定的城市名称查询该城市的当前天气情况支持国内外主要城市”。这样LLM在理解用户意图“上海下雨了吗”时能更准确地匹配到“城市名称”和“天气”这两个关键点。参数Schema要严谨且自解释type必须准确string,number,integer,boolean,array,object。description字段务必填写对于city参数描述写成“城市名称例如北京、上海、New York”这相当于给了LLM几个示例few-shot能显著提升它填充参数的正确率。善用enum如果参数只有几个固定值一定要用enum列出。例如from_unit: {type: string, enum: [km, mile, kg, lb], description: ...}。这能从根本上杜绝LLM胡编乱造一个不支持的参数值。required数组要列全确保所有调用时必须的参数都在这里。一个反面教材# 差的Schema Tool( namesearch, description搜索信息, inputSchema{ type: object, properties: { q: {type: string} } } )LLM看到这个它不知道q代表什么也不知道该搜什么。它可能不会调用或者调用时乱写q的内容。一个优秀范例# 好的Schema Tool( namesearch_web, description使用搜索引擎查询最新的网络信息适合回答关于实时事件、新闻、最新知识的问题。, inputSchema{ type: object, properties: { query: { type: string, description: 搜索查询关键词应具体明确例如2024年巴黎奥运会最新金牌榜、Python asyncio 教程最新版 }, max_results: { type: integer, description: 返回的最大结果数量默认为5, default: 5 } }, required: [query] } )4.3 错误处理与健壮性设计生产环境的工具必须考虑各种失败情况。Server端错误处理在handle_call_tool函数内部要用try...except包裹业务逻辑。try: result await some_network_call(arguments) return [{type: text, text: result}] except NetworkTimeoutError: # 返回结构化的错误信息符合MCP协议 raise McpError( code-32000, message网络请求超时, data{suggestion: 请稍后重试或检查网络连接} ) except ValidationError as e: raise McpError( code-32602, message参数验证失败, data{details: str(e)} )MCP协议定义了标准的错误码范围如-32600到-32603是JSON-RPC标准错误-32000到-32099是自定义服务器错误使用它们有助于Client进行统一处理。Client端错误处理Client调用call_tool时可能会收到错误。一个健壮的Agent应该能处理这些错误而不是崩溃。try: response await session.call_tool(tool_name, arguments) # 处理成功响应 except McpError as e: if e.code -32000: # 自定义超时错误 # 策略1重试 # 策略2使用备用工具 # 策略3告知用户“查询超时请稍后再试” fallback_result 当前服务繁忙已为您提供缓存信息... elif e.code -32602: # 无效参数 # 尝试修正参数或直接向用户澄清 clarification f参数有误{e.data[details]}请确认您想查询的城市是 # 将clarification送回LLM让它重新生成问题或与用户交互 else: # 其他未知错误记录日志并降级处理 log_error(e) final_answer 工具暂时不可用请稍后尝试。超时与心跳对于HTTP/SSE连接必须设置合理的读写超时。此外MCP协议支持ping/pong消息作为心跳用于检测连接健康状态。在生产Client中实现心跳机制和自动重连逻辑是必要的。4.4 安全与权限考量当工具能力被暴露后安全就成为重中之重。身份认证与授权在HTTP模式下必须在Server端实现认证。可以在初始化连接时要求Client提供API Key或Token并在Server端进行验证。MCP协议本身不规定认证方式这需要你在传输层之上自己实现例如在HTTP头中添加Authorization。参数校验与净化永远不要相信Client传来的参数。即使在Schema中定义了类型Server端在执行业务逻辑前必须进行二次校验和净化。特别是涉及数据库查询、系统命令执行、文件操作的工具要严防注入攻击。访问范围控制不同的Client代表不同的用户或Agent可能拥有不同的工具调用权限。可以在Server端维护一个权限映射表在handle_call_tool中检查当前会话是否有权调用name指定的工具。5. 集成到现有LLM Agent框架的实战指南现在我们已经有了一个健壮的MCP Server。如何让它被现有的LangChain、LlamaIndex等框架使用呢原理很简单为这些框架实现一个MCP Client适配器。以LangChain为例我们需要创建一个自定义的Tool类这个类内部封装了与MCP Server的通信逻辑。from langchain.tools import BaseTool from pydantic import BaseModel, Field import asyncio # 假设我们有一个封装好的异步MCP Client from my_mcp_client import AsyncMCPClient class MCPTool(BaseTool): name: str description: str mcp_client: AsyncMCPClient args_schema: type None # 可以动态生成 def _run(self, **kwargs): # LangChain默认是同步的这里需要异步转同步仅示例生产环境应用异步Agent loop asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result loop.run_until_complete(self.mcp_client.call_tool(self.name, kwargs)) return result.contents[0].text if result.contents else finally: loop.close() async def _arun(self, **kwargs): # 异步版本用于支持异步Agent result await self.mcp_client.call_tool(self.name, kwargs) return result.contents[0].text if result.contents else # 使用方式 async def main(): client AsyncMCPClient(server_urlhttp://localhost:8080) await client.connect() # 动态发现工具并创建LangChain Tool列表 tools_info await client.list_tools() langchain_tools [] for tool_info in tools_info: # 根据tool_info.inputSchema动态创建Pydantic模型作为args_schema # ... (动态模型创建代码略) langchain_tools.append( MCPTool( nametool_info.name, descriptiontool_info.description, mcp_clientclient, args_schemaDynamicArgsSchemaModel # 动态生成的模型 ) ) # 将tools列表赋给你的LangChain Agent agent initialize_agent( llmyour_llm, toolslangchain_tools, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue ) result await agent.arun(北京和上海今天天气怎么样) print(result)核心思路利用MCP Client的list_tools方法在Agent初始化时动态地获取所有远程工具的描述和Schema然后为每个工具实例化一个LangChain的Tool对象。这样你的LangChain Agent就具备了调用远程MCP Server的能力而且当Server端新增或更新工具时Agent无需修改代码只需重新初始化或动态加载即可感知。对于LlamaIndex、AutoGen等其他框架集成模式大同小异核心都是实现一个符合框架要求的Tool抽象背后委托给MCP Client。6. 常见问题、排查技巧与性能优化在实际部署和调试MCP系统时你肯定会遇到各种问题。下面是我踩过坑后总结的一些排查思路和优化建议。6.1 问题排查速查表问题现象可能原因排查步骤Client连接Server失败1. Server进程未启动。2. 传输方式不匹配Client用HTTPServer用Stdio。3. 端口被占用或URL错误。1. 检查Server进程是否在运行ps aux | grep server.py。2. 确认Client和Server配置的传输层Stdio/HTTP一致。3. 对于HTTP用curl测试/sse端点是否可达。list_tools返回空列表1. Server的工具注册逻辑有误server.list_tools装饰器未正确返回数据。2. 初始化消息未正确交换。1. 在Server的handle_list_tools函数内打印日志确认其被调用和返回值。2. 检查Server启动日志看是否有错误。启用MCP库的调试日志如设置环境变量MCP_LOGdebug。call_tool返回“未知工具”错误1. Client传递的工具名与Server注册的名称大小写或拼写不一致。2. Server的工具路由逻辑handle_call_tool有bug。1. 对比Client调用时的tool_name和Serverlist_tools返回的名称。2. 在Server的handle_call_tool函数开始处打印接收到的name和arguments进行调试。LLM无法正确选择或调用工具1. 工具描述description不够清晰LLM无法理解其用途。2. 参数SchemainputSchema描述模糊或缺少示例。3. LLM的提示词Prompt中未充分引导其使用工具。1. 优化description包含明确的使用场景和关键词。2. 为每个参数添加详细的description和enum或examples。3. 在给LLM的System Prompt中明确指示其可以使用工具并简要说明工具能力。工具调用超时或无响应1. 工具函数本身执行缓慢如网络请求、复杂计算。2. Server或Client未设置合理的超时。3. 网络问题。1. 在工具函数内部添加超时控制。2. 在Client调用call_tool时设置超时参数如果SDK支持。3. 对于HTTP模式检查网络延迟和防火墙设置。Stdio模式下Server进程僵尸Client异常退出未正确关闭Server进程。在Client代码中使用try...finally确保server_process.terminate()或server_process.kill()被调用。更好的方式是使用asyncio的create_subprocess_shell并管理其生命周期。6.2 性能优化建议连接池HTTP模式如果Client需要频繁调用同一个Server不要为每次调用都建立新的HTTP/SSE连接。应该实现一个连接池维护一个可复用的长连接。批处理工具调用MCP协议支持tools/call的批量调用吗目前标准协议似乎是一次一个。但如果你的场景需要连续调用多个工具可以在Client端实现一个简单的批处理队列或者考虑在Server端暴露一个组合工具Composite Tool将多个操作封装成一次调用减少网络往返。Server无状态化尽可能将MCP Server设计为无状态的。这样便于水平扩展可以通过负载均衡部署多个Server实例。任何会话状态或用户上下文应该由Client持有并通过参数传递或者存储在外部服务如Redis中。监控与日志在Server和Client的关键节点添加结构化日志如工具调用开始/结束、参数、耗时、结果状态。这便于监控系统健康度和排查问题。可以集成像Prometheus这样的指标系统暴露工具调用次数、延迟、错误率等指标。6.3 一个真实的踩坑案例Schema变更的兼容性我在一个项目中最初为get_weather工具只定义了city参数。后来业务需要想增加一个country可选参数用于区分同名城市。我直接修改了Server端的Schema增加了这个参数。结果已经在线运行的Client特别是那些缓存了旧版工具列表的Agent在调用时仍然只传city参数导致Server端校验失败。教训与解决方案向后兼容添加新参数时尽量将其设为required: false并提供合理的默认值。在工具处理函数中优雅地处理旧客户端缺失该参数的情况。版本管理考虑在工具描述或Server初始化信息中加入版本号。Client可以在发现工具时感知到版本变化并决定是否更新本地缓存或采取其他行动。平滑升级采用蓝绿部署或金丝雀发布策略。先部署支持新旧两种参数格式的Server新版本然后逐步更新Client最后再移除对旧格式的支持。通过MCP协议将LLM与工具解耦不仅仅是技术架构的优化更是一种思维方式的转变。它让我们从“如何让LLM调用我的Python函数”这种紧耦合的思维中跳出来转向“如何向LLM生态提供一组标准的、可靠的服务”。当你开始用“协议”和“服务”的视角来设计工具时整个Agent系统的灵活性、可维护性和团队协作效率都会得到质的提升。这300行代码就是一个全新的起点。 基于MCP协议实现LLM与工具解耦：300行代码构建标准化Agent工具层
1. 项目概述为什么我们需要重新思考LLM与工具的关系最近在折腾大语言模型应用落地的朋友估计都绕不开一个词Agent。无论是想做个能自动处理邮件的助手还是想搞个能分析数据的智能体最终都得让LLM学会“使用工具”。但真上手了你会发现这事儿远没想象中那么简单。传统的Agent工具集成方式就像把一台高性能发动机硬塞进一辆老式马车的底盘——动力是有了但跑起来浑身别扭动不动就散架。我花了大量时间在几个实际项目中从简单的天气查询到复杂的企业级数据管道调度几乎把能踩的坑都踩了一遍。最终我意识到问题的核心在于耦合度。目前主流的做法无论是LangChain的Tool定义还是LlamaIndex的Function Calling本质上都是将工具的实现逻辑、调用协议乃至错误处理与LLM的提示词工程和推理流程深度捆绑在一起。这直接导致了两个让我头疼不已的痛点痛点一语言绑定的僵局。我的工具库是用Python写的性能好、生态丰富。但业务团队可能更熟悉Go或者Java他们想复用我的工具逻辑或者前端同事想用JavaScript直接调用几乎不可能。每次都要为了适配新的调用方用目标语言重写一遍工具逻辑这不仅是重复劳动更是维护的噩梦。一个工具逻辑的更新需要同步更新多个语言版本的实现一致性根本无法保证。痛点二LLM与工具的生死与共。在传统架构下工具服务的稳定性直接决定了整个Agent系统的可用性。如果工具服务因为网络、依赖或自身bug挂掉LLM的调用链会立刻中断返回一个令人沮丧的错误。更糟糕的是错误信息往往难以被LLM理解和处理导致整个对话流程卡死。你不得不花大量精力去构建健壮的错误处理和中途拦截机制这部分的代码复杂度甚至超过了业务逻辑本身。所以当我在社区里看到MCPModel Context Protocol这个协议开始被讨论时眼前顿时一亮。它的核心思想非常直接在LLM和工具之间建立一个标准化的、与语言无关的通信层。工具提供者只需按照协议暴露能力LLM或者说Agent框架只需按照协议去发现和调用双方不再需要关心对方是用什么语言实现的。于是我决定动手验证一下这个想法。目标很明确用尽可能少的代码实现一个MCP Server将我的Python工具库暴露出去同时再实现一个极简的MCP Client让LLM能通过这个Client调用工具。整个过程下来核心代码真的控制在了300行左右。结果令人振奋不仅成功解耦整个系统的可维护性和扩展性都上了一个台阶。下面我就把这套方案的完整设计思路、实操步骤以及踩坑心得毫无保留地分享出来。2. MCP协议核心思想与架构拆解在动手写代码之前我们必须先吃透MCP到底要解决什么问题以及它是如何设计的。你可以把它想象成计算机硬件里的USB协议。在USB出现之前鼠标、键盘、打印机各有各的接口你需要不同的线缆、不同的驱动插错了口甚至可能烧坏设备。USB协议出现后定义了一套标准的物理接口、电气信号和通信规范。从此设备制造商只需生产符合USB标准的设备电脑主板只需提供USB接口双方就能即插即用背后的复杂驱动和协商过程都被协议层消化了。MCP在LLM与工具的世界里扮演的正是这个“USB协议”的角色。它的设计目标非常清晰标准化Standardization定义工具如何向LLM描述自己名称、描述、参数格式以及LLM如何调用工具请求格式、响应格式。这解决了“工具描述语言”不统一的问题。解耦Decoupling工具的实现Server端和LLM对工具的调用Client端完全分离。工具可以用任何语言编写运行在任何地方LLM Agent也可以通过任何支持MCP协议的Client来调用这些工具。它们之间只通过标准的协议消息进行通信。可发现性DiscoverabilityClient能够动态地发现Server提供了哪些工具并获取其完整的调用规范无需提前硬编码。这使得工具的热插拔成为可能。2.1 MCP的核心组件与交互流程一个完整的MCP架构通常包含三个角色MCP Server工具提供方它封装了具体的工具逻辑例如“查询数据库”、“发送邮件”、“生成图表”。Server启动后会在一个网络端点如HTTP端口或Stdio等待连接。MCP Client工具调用方它代表LLM或Agent框架负责与Server建立连接获取工具列表并代表LLM发起工具调用。Client不包含任何具体的工具逻辑。传输层Transport负责在Server和Client之间传递标准的JSON-RPC消息。主流支持两种方式标准输入输出Stdio和HTTP。Stdio模式通常用于本地紧密集成的场景如一个CLI工具而HTTP模式则适用于跨网络、分布式的部署。它们之间的交互可以概括为以下几个核心步骤初始化连接InitializeClient向Server发起连接并交换各自的元信息如名称、版本、支持的能力。列出工具ListToolsClient调用tools/list方法Server返回一个工具描述列表。每个描述都包含工具的唯一名称、详细的功能描述、以及调用所需的参数JSON Schema。这是LLM理解工具能力的关键。调用工具CallTool当LLM决定使用某个工具时Client会向Server发起tools/call请求携带工具名和具体的参数。Server执行实际逻辑。返回结果Server执行完毕后将结果或错误信息封装成标准格式返回给Client。Client再将其呈现给LLM用于后续的推理或回答生成。整个过程中Client和Server都不需要知道对方的实现细节。Client不在乎工具是用Python还是Go写的Server也不在乎调用它的是LangChain Agent还是自定义的脚本。2.2 为什么MCP能根治两大痛点现在让我们回到开头的两个痛点看看MCP是如何解决的针对“语言绑定”痛点MCP Server可以用任何语言实现只要它遵循协议发送和接收JSON-RPC消息。这意味着你可以用Python写一个高性能的数据处理工具Server同时用Go写一个系统管理工具Server。你的LLM Agent通过MCP Client可以同时无缝调用它们无需任何语言适配层。团队协作时后端用Java算法用Python前端用JS都可以各自维护自己的MCP Server最终在Agent层面统一集成。针对“耦合故障”痛点由于通信是标准化的错误也被纳入了协议规范。工具Server可以返回结构化的错误信息如错误码、类型、详情。MCP Client可以统一处理这些错误例如进行重试、降级处理或者将友好的错误信息反馈给LLM让LLM决定下一步动作比如建议用户检查输入。更重要的是某个工具的故障甚至Server崩溃通常不会导致Client或LLM进程崩溃因为它们之间是松耦合的网络/进程间通信。你可以方便地实现Server的健康检查、熔断和负载均衡。理解了这些我们就能明白实现MCP的关键不在于复杂的业务逻辑而在于正确地实现协议规定的几个核心JSON-RPC方法。接下来我们就进入实战环节。3. 300行代码实现MCP Server与Client详解我们的目标是构建一个最小可行系统。假设我们有一个简单的“天气查询”工具和一个“单位换算”工具我们将用Python实现它们的MCP Server同时实现一个简单的MCP Client来演示调用。我们将选择Stdio传输层因为它最简单无需处理网络问题适合本地集成和演示。3.1 环境准备与依赖选择首先创建一个新的项目目录。我们不需要重量级的框架Python标准库的json和subprocess以及sys就足够了。但为了更规范地处理JSON-RPC和协议细节我们使用一个轻量级的库mcp。这是一个低级别的、用于构建MCP组件的Python SDK。# 创建项目目录并初始化虚拟环境推荐 mkdir mcp-demo cd mcp-demo python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate # 安装核心依赖 pip install mcpmcp库提供了编写Server和Client所需的底层抽象帮我们处理了协议消息的序列化/反序列化、生命周期管理等样板代码让我们能专注于工具逻辑本身。3.2 实现MCP Server约150行我们在项目根目录下创建server.py。#!/usr/bin/env python3 一个简单的MCP Server示例提供天气查询和单位换算工具。 使用Stdio传输。 import asyncio import json import sys from typing import Any, List from mcp import ClientSession, StdioServerParameters from mcp.server import Server from mcp.server.models import Tool import mcp.server.stdio # 模拟一个简单的天气查询函数 async def get_weather(city: str) - str: 根据城市名返回模拟天气信息。 # 这里应该是调用真实API例如OpenWeatherMap # 为了演示我们返回模拟数据 weather_data { 北京: 晴25°C北风2级, 上海: 多云28°C东南风3级, 深圳: 雷阵雨30°C南风1级, } return weather_data.get(city, f未找到{city}的天气信息。) # 模拟一个单位换算函数 async def convert_units(value: float, from_unit: str, to_unit: str) - str: 进行简单的单位换算。 conversions { (km, mile): lambda v: v * 0.621371, (mile, km): lambda v: v * 1.60934, (kg, lb): lambda v: v * 2.20462, (lb, kg): lambda v: v * 0.453592, (celsius, fahrenheit): lambda v: (v * 9/5) 32, (fahrenheit, celsius): lambda v: (v - 32) * 5/9, } key (from_unit.lower(), to_unit.lower()) if key in conversions: result conversions[key](value) return f{value} {from_unit} {result:.2f} {to_unit} else: return f不支持从 {from_unit} 到 {to_unit} 的换算。 async def main(): # 1. 创建MCP Server实例 server Server(demo-tools-server) # 2. 向Server注册工具 # 每个Tool对象定义了工具的名称、描述和输入参数模式 server.list_tools() async def handle_list_tools() - List[Tool]: return [ Tool( nameget_weather, description根据给定的城市名称查询该城市的当前天气情况。, inputSchema{ type: object, properties: { city: { type: string, description: 城市名称例如北京、上海、New York } }, required: [city] } ), Tool( nameconvert_units, description进行常见的单位换算例如公里与英里、公斤与磅、摄氏度与华氏度之间的转换。, inputSchema{ type: object, properties: { value: {type: number, description: 需要换算的数值}, from_unit: {type: string, description: 原始单位例如km, mile, kg, lb, celsius, fahrenheit}, to_unit: {type: string, description: 目标单位} }, required: [value, from_unit, to_unit] } ) ] # 3. 绑定工具名称到具体的处理函数 server.call_tool() async def handle_call_tool(name: str, arguments: dict) - list: if name get_weather: city arguments.get(city) if not city: raise ValueError(参数 city 是必需的。) result_text await get_weather(city) # 返回格式需符合MCP协议content是一个列表 return [{ type: text, text: result_text }] elif name convert_units: value arguments.get(value) from_unit arguments.get(from_unit) to_unit arguments.get(to_unit) if None in (value, from_unit, to_unit): raise ValueError(参数 value, from_unit, to_unit 都是必需的。) result_text await convert_units(float(value), from_unit, to_unit) return [{ type: text, text: result_text }] else: raise ValueError(f未知的工具: {name}) # 4. 使用Stdio传输层运行Server # 这允许通过标准输入输出与Client通信 async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, # 可选的Server初始化信息 server_info{ name: Demo Tools Server, version: 0.1.0 } ) if __name__ __main__: asyncio.run(main())代码关键点解析工具定义Tool对象这是LLM理解工具的“说明书”。description字段至关重要LLM会根据它来判断在什么场景下使用这个工具。inputSchema必须严格按照JSON Schema格式定义它规定了调用时必须传入哪些参数以及参数的类型和格式。清晰的Schema能极大减少LLM调用出错的概率。工具实现函数get_weather和convert_units是实际执行业务逻辑的函数。它们可以是同步或异步的可以调用任何其他库或服务。这里为了演示用了模拟数据。请求路由handle_call_tool这个装饰器函数是Server的核心路由器。它根据传入的name找到对应的工具并从arguments字典中提取参数调用真正的业务函数。Stdio传输mcp.server.stdio.stdio_server()创建了基于标准输入输出的通信通道。这意味着我们的Server可以作为一个独立的命令行进程启动Client通过管道与之通信。这是最简单、最轻量的集成方式。注意在实际生产环境中get_weather函数应该调用真实的天气API并做好错误处理如网络超时、API限流、无效城市名等。这些错误应该在函数内部捕获并转化为用户或LLM可理解的错误信息通过MCP协议返回。3.3 实现MCP Client约100行接下来我们创建一个client.py它代表LLM或Agent框架负责与Server对话。#!/usr/bin/env python3 一个简单的MCP Client示例演示如何发现并调用Server提供的工具。 import asyncio import json import subprocess import sys from typing import Any, Dict, List from mcp import ClientSession from mcp.client.stdio import stdio_client async def main(): # 1. 启动MCP Server进程并建立Stdio连接 # 这里我们启动刚才写的server.py进程 server_process subprocess.Popen( [sys.executable, server.py], # 使用当前Python解释器运行server.py stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsys.stderr, # 将Server的错误输出到控制台便于调试 textTrue ) # 2. 创建MCP Client会话连接到Server进程的输入输出流 async with stdio_client(server_process.stdin, server_process.stdout) as (read_stream, write_stream): async with ClientSession(read_stream, write_stream) as session: # 3. 初始化连接与Server握手 await session.initialize() # 4. 列出Server提供的所有工具 print(正在从Server发现可用工具...) tools await session.list_tools() print(f发现 {len(tools)} 个工具:) for tool in tools: print(f - {tool.name}: {tool.description}) # 可以打印详细的参数模式供LLM或开发者参考 # print(f 参数模式: {json.dumps(tool.inputSchema, indent2, ensure_asciiFalse)}) # 5. 模拟LLM决策过程根据用户请求选择工具并调用 # 假设LLM分析了用户输入“北京天气怎么样”后决定调用get_weather user_query 北京天气怎么样 print(f\n用户查询: {user_query}) print(LLM决策: 调用 get_weather 工具) tool_name get_weather arguments {city: 北京} print(f调用工具: {tool_name} 参数: {arguments}) try: # 执行工具调用 response await session.call_tool(tool_name, arguments) # response是一个Content对象列表我们取第一个文本内容 if response and response.contents: tool_result response.contents[0].text print(f工具返回结果: {tool_result}) # 在实际Agent中这个结果会被喂回给LLM用于生成最终回答 final_answer f根据查询北京的天气情况是{tool_result} print(fAgent最终回答: {final_answer}) else: print(工具调用未返回有效内容。) except Exception as e: print(f工具调用失败: {e}) # 6. 再演示一个调用 print(\n--- 第二个例子 ---) user_query 把10公里换算成英里 print(f用户查询: {user_query}) print(LLM决策: 调用 convert_units 工具) tool_name convert_units arguments {value: 10, from_unit: km, to_unit: mile} print(f调用工具: {tool_name} 参数: {arguments}) try: response await session.call_tool(tool_name, arguments) if response and response.contents: tool_result response.contents[0].text print(f工具返回结果: {tool_result}) final_answer f换算结果是{tool_result} print(fAgent最终回答: {final_answer}) else: print(工具调用未返回有效内容。) except Exception as e: print(f工具调用失败: {e}) # 7. 会话结束等待Server进程退出 server_process.wait() print(\nClient会话结束。) if __name__ __main__: asyncio.run(main())代码关键点解析启动Server进程Client使用subprocess.Popen启动server.py并捕获其标准输入和输出。这模拟了将MCP Server作为一个独立子进程管理的场景。建立会话stdio_client和ClientSession管理了与Server的协议层通信包括初始化握手、消息发送和接收。工具发现session.list_tools()是Client获取Server能力目录的方法。在一个动态环境中Agent可以在启动时或定期调用此方法以感知工具的变化。工具调用session.call_tool()是核心调用方法。Client只需要知道工具名和符合Schema的参数即可完全无需了解工具的内部实现。返回的response对象结构是标准的包含了工具执行产生的文本、图片或其他类型的内容。错误处理调用被try...except包裹。MCP协议允许Server返回结构化的错误Client可以根据错误类型决定重试、降级或通知用户。3.4 运行演示在终端中确保处于虚拟环境然后直接运行Clientpython client.py你将看到类似以下的输出正在从Server发现可用工具... 发现 2 个工具: - get_weather: 根据给定的城市名称查询该城市的当前天气情况。 - convert_units: 进行常见的单位换算例如公里与英里、公斤与磅、摄氏度与华氏度之间的转换。 用户查询: 北京天气怎么样 LLM决策: 调用 get_weather 工具 调用工具: get_weather 参数: {city: 北京} 工具返回结果: 晴25°C北风2级 Agent最终回答: 根据查询北京的天气情况是晴25°C北风2级 --- 第二个例子 --- 用户查询: 把10公里换算成英里 LLM决策: 调用 convert_units 工具 调用工具: convert_units 参数: {value: 10, from_unit: km, to_unit: mile} 工具返回结果: 10.0 km 6.21 mile Agent最终回答: 换算结果是10.0 km 6.21 mile Client会话结束。至此一个完整的、解耦的MCP工具调用流程就完成了。整个核心通信逻辑Server Client的代码量正如标题所说在300行左右。4. 从Demo到生产关键配置与进阶实践上面的Demo跑通了基本流程但要想投入实际使用还有几个关键环节需要加固和优化。这部分才是体现工程经验价值的地方。4.1 传输层选择Stdio vs. HTTP (SSE)我们的Demo使用了Stdio它简单直接适合本地集成比如你的Agent应用和工具Server部署在同一台机器上或者作为单个应用的插件系统。它的生命周期管理也简单Client启动ServerClient退出时Server通常也结束。但对于微服务架构或远程工具HTTP通常基于Server-Sent Events, SSE是更合适的选择。MCP over HTTP允许你将工具Server部署在独立的容器或服务器上通过网络提供服务。多个Client可以连接同一个Server实现工具能力的共享和复用。如何切换为HTTP传输在Server端你需要一个支持SSE的HTTP框架比如FastAPI。核心是暴露两个端点GET /sse用于建立SSE长连接持续接收Client的请求。POST /message用于Client向Server发送具体的JSON-RPC请求。在Client端不再启动子进程而是直接连接到Server的HTTP URL。实操心得对于内部系统Stdio模式因其零网络延迟和简单的权限管理继承父进程权限而更高效。对于需要跨团队、跨网络共享的工具服务HTTP模式是必然选择。许多成熟的MCP SDK包括Pythonmcp库的高层API都同时支持两种模式切换通常只需更改几行配置代码。4.2 工具描述的“艺术”编写LLM友好的Schema工具能否被LLM正确调用一半取决于Tool对象中的description和inputSchema。写得好LLM调用精准写得差LLM要么不用要么乱用。优化技巧描述要具体且包含关键词不要写“查询天气”要写“根据给定的城市名称查询该城市的当前天气情况支持国内外主要城市”。这样LLM在理解用户意图“上海下雨了吗”时能更准确地匹配到“城市名称”和“天气”这两个关键点。参数Schema要严谨且自解释type必须准确string,number,integer,boolean,array,object。description字段务必填写对于city参数描述写成“城市名称例如北京、上海、New York”这相当于给了LLM几个示例few-shot能显著提升它填充参数的正确率。善用enum如果参数只有几个固定值一定要用enum列出。例如from_unit: {type: string, enum: [km, mile, kg, lb], description: ...}。这能从根本上杜绝LLM胡编乱造一个不支持的参数值。required数组要列全确保所有调用时必须的参数都在这里。一个反面教材# 差的Schema Tool( namesearch, description搜索信息, inputSchema{ type: object, properties: { q: {type: string} } } )LLM看到这个它不知道q代表什么也不知道该搜什么。它可能不会调用或者调用时乱写q的内容。一个优秀范例# 好的Schema Tool( namesearch_web, description使用搜索引擎查询最新的网络信息适合回答关于实时事件、新闻、最新知识的问题。, inputSchema{ type: object, properties: { query: { type: string, description: 搜索查询关键词应具体明确例如2024年巴黎奥运会最新金牌榜、Python asyncio 教程最新版 }, max_results: { type: integer, description: 返回的最大结果数量默认为5, default: 5 } }, required: [query] } )4.3 错误处理与健壮性设计生产环境的工具必须考虑各种失败情况。Server端错误处理在handle_call_tool函数内部要用try...except包裹业务逻辑。try: result await some_network_call(arguments) return [{type: text, text: result}] except NetworkTimeoutError: # 返回结构化的错误信息符合MCP协议 raise McpError( code-32000, message网络请求超时, data{suggestion: 请稍后重试或检查网络连接} ) except ValidationError as e: raise McpError( code-32602, message参数验证失败, data{details: str(e)} )MCP协议定义了标准的错误码范围如-32600到-32603是JSON-RPC标准错误-32000到-32099是自定义服务器错误使用它们有助于Client进行统一处理。Client端错误处理Client调用call_tool时可能会收到错误。一个健壮的Agent应该能处理这些错误而不是崩溃。try: response await session.call_tool(tool_name, arguments) # 处理成功响应 except McpError as e: if e.code -32000: # 自定义超时错误 # 策略1重试 # 策略2使用备用工具 # 策略3告知用户“查询超时请稍后再试” fallback_result 当前服务繁忙已为您提供缓存信息... elif e.code -32602: # 无效参数 # 尝试修正参数或直接向用户澄清 clarification f参数有误{e.data[details]}请确认您想查询的城市是 # 将clarification送回LLM让它重新生成问题或与用户交互 else: # 其他未知错误记录日志并降级处理 log_error(e) final_answer 工具暂时不可用请稍后尝试。超时与心跳对于HTTP/SSE连接必须设置合理的读写超时。此外MCP协议支持ping/pong消息作为心跳用于检测连接健康状态。在生产Client中实现心跳机制和自动重连逻辑是必要的。4.4 安全与权限考量当工具能力被暴露后安全就成为重中之重。身份认证与授权在HTTP模式下必须在Server端实现认证。可以在初始化连接时要求Client提供API Key或Token并在Server端进行验证。MCP协议本身不规定认证方式这需要你在传输层之上自己实现例如在HTTP头中添加Authorization。参数校验与净化永远不要相信Client传来的参数。即使在Schema中定义了类型Server端在执行业务逻辑前必须进行二次校验和净化。特别是涉及数据库查询、系统命令执行、文件操作的工具要严防注入攻击。访问范围控制不同的Client代表不同的用户或Agent可能拥有不同的工具调用权限。可以在Server端维护一个权限映射表在handle_call_tool中检查当前会话是否有权调用name指定的工具。5. 集成到现有LLM Agent框架的实战指南现在我们已经有了一个健壮的MCP Server。如何让它被现有的LangChain、LlamaIndex等框架使用呢原理很简单为这些框架实现一个MCP Client适配器。以LangChain为例我们需要创建一个自定义的Tool类这个类内部封装了与MCP Server的通信逻辑。from langchain.tools import BaseTool from pydantic import BaseModel, Field import asyncio # 假设我们有一个封装好的异步MCP Client from my_mcp_client import AsyncMCPClient class MCPTool(BaseTool): name: str description: str mcp_client: AsyncMCPClient args_schema: type None # 可以动态生成 def _run(self, **kwargs): # LangChain默认是同步的这里需要异步转同步仅示例生产环境应用异步Agent loop asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result loop.run_until_complete(self.mcp_client.call_tool(self.name, kwargs)) return result.contents[0].text if result.contents else finally: loop.close() async def _arun(self, **kwargs): # 异步版本用于支持异步Agent result await self.mcp_client.call_tool(self.name, kwargs) return result.contents[0].text if result.contents else # 使用方式 async def main(): client AsyncMCPClient(server_urlhttp://localhost:8080) await client.connect() # 动态发现工具并创建LangChain Tool列表 tools_info await client.list_tools() langchain_tools [] for tool_info in tools_info: # 根据tool_info.inputSchema动态创建Pydantic模型作为args_schema # ... (动态模型创建代码略) langchain_tools.append( MCPTool( nametool_info.name, descriptiontool_info.description, mcp_clientclient, args_schemaDynamicArgsSchemaModel # 动态生成的模型 ) ) # 将tools列表赋给你的LangChain Agent agent initialize_agent( llmyour_llm, toolslangchain_tools, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue ) result await agent.arun(北京和上海今天天气怎么样) print(result)核心思路利用MCP Client的list_tools方法在Agent初始化时动态地获取所有远程工具的描述和Schema然后为每个工具实例化一个LangChain的Tool对象。这样你的LangChain Agent就具备了调用远程MCP Server的能力而且当Server端新增或更新工具时Agent无需修改代码只需重新初始化或动态加载即可感知。对于LlamaIndex、AutoGen等其他框架集成模式大同小异核心都是实现一个符合框架要求的Tool抽象背后委托给MCP Client。6. 常见问题、排查技巧与性能优化在实际部署和调试MCP系统时你肯定会遇到各种问题。下面是我踩过坑后总结的一些排查思路和优化建议。6.1 问题排查速查表问题现象可能原因排查步骤Client连接Server失败1. Server进程未启动。2. 传输方式不匹配Client用HTTPServer用Stdio。3. 端口被占用或URL错误。1. 检查Server进程是否在运行ps aux | grep server.py。2. 确认Client和Server配置的传输层Stdio/HTTP一致。3. 对于HTTP用curl测试/sse端点是否可达。list_tools返回空列表1. Server的工具注册逻辑有误server.list_tools装饰器未正确返回数据。2. 初始化消息未正确交换。1. 在Server的handle_list_tools函数内打印日志确认其被调用和返回值。2. 检查Server启动日志看是否有错误。启用MCP库的调试日志如设置环境变量MCP_LOGdebug。call_tool返回“未知工具”错误1. Client传递的工具名与Server注册的名称大小写或拼写不一致。2. Server的工具路由逻辑handle_call_tool有bug。1. 对比Client调用时的tool_name和Serverlist_tools返回的名称。2. 在Server的handle_call_tool函数开始处打印接收到的name和arguments进行调试。LLM无法正确选择或调用工具1. 工具描述description不够清晰LLM无法理解其用途。2. 参数SchemainputSchema描述模糊或缺少示例。3. LLM的提示词Prompt中未充分引导其使用工具。1. 优化description包含明确的使用场景和关键词。2. 为每个参数添加详细的description和enum或examples。3. 在给LLM的System Prompt中明确指示其可以使用工具并简要说明工具能力。工具调用超时或无响应1. 工具函数本身执行缓慢如网络请求、复杂计算。2. Server或Client未设置合理的超时。3. 网络问题。1. 在工具函数内部添加超时控制。2. 在Client调用call_tool时设置超时参数如果SDK支持。3. 对于HTTP模式检查网络延迟和防火墙设置。Stdio模式下Server进程僵尸Client异常退出未正确关闭Server进程。在Client代码中使用try...finally确保server_process.terminate()或server_process.kill()被调用。更好的方式是使用asyncio的create_subprocess_shell并管理其生命周期。6.2 性能优化建议连接池HTTP模式如果Client需要频繁调用同一个Server不要为每次调用都建立新的HTTP/SSE连接。应该实现一个连接池维护一个可复用的长连接。批处理工具调用MCP协议支持tools/call的批量调用吗目前标准协议似乎是一次一个。但如果你的场景需要连续调用多个工具可以在Client端实现一个简单的批处理队列或者考虑在Server端暴露一个组合工具Composite Tool将多个操作封装成一次调用减少网络往返。Server无状态化尽可能将MCP Server设计为无状态的。这样便于水平扩展可以通过负载均衡部署多个Server实例。任何会话状态或用户上下文应该由Client持有并通过参数传递或者存储在外部服务如Redis中。监控与日志在Server和Client的关键节点添加结构化日志如工具调用开始/结束、参数、耗时、结果状态。这便于监控系统健康度和排查问题。可以集成像Prometheus这样的指标系统暴露工具调用次数、延迟、错误率等指标。6.3 一个真实的踩坑案例Schema变更的兼容性我在一个项目中最初为get_weather工具只定义了city参数。后来业务需要想增加一个country可选参数用于区分同名城市。我直接修改了Server端的Schema增加了这个参数。结果已经在线运行的Client特别是那些缓存了旧版工具列表的Agent在调用时仍然只传city参数导致Server端校验失败。教训与解决方案向后兼容添加新参数时尽量将其设为required: false并提供合理的默认值。在工具处理函数中优雅地处理旧客户端缺失该参数的情况。版本管理考虑在工具描述或Server初始化信息中加入版本号。Client可以在发现工具时感知到版本变化并决定是否更新本地缓存或采取其他行动。平滑升级采用蓝绿部署或金丝雀发布策略。先部署支持新旧两种参数格式的Server新版本然后逐步更新Client最后再移除对旧格式的支持。通过MCP协议将LLM与工具解耦不仅仅是技术架构的优化更是一种思维方式的转变。它让我们从“如何让LLM调用我的Python函数”这种紧耦合的思维中跳出来转向“如何向LLM生态提供一组标准的、可靠的服务”。当你开始用“协议”和“服务”的视角来设计工具时整个Agent系统的灵活性、可维护性和团队协作效率都会得到质的提升。这300行代码就是一个全新的起点。