ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

MCP协议:AI与外部工具的标准接口设计与实践

MCP协议:AI与外部工具的标准接口设计与实践 1. 项目概述当AI需要“伸手”时最近在折腾各种AI应用和智能体Agent时我遇到了一个非常具体且普遍的痛点如何让大模型“伸手”去操作外部世界比如我想让一个帮我写周报的AI能自动去Jira拉取我本周的任务列表或者让一个数据分析助手能直接查询公司内部的数据库。这听起来像是AI Agent的标配能力但实际落地时你会发现连接外部工具和数据源的过程异常繁琐——每个工具都要写特定的适配代码处理不同的认证、参数和错误格式就像给每个新设备都重写一遍驱动程序。直到我深入研究了MCPModel Context Protocol协议才感觉找到了那个“通用接口”。这个协议的目标非常明确为AI应用与外部工具、数据源之间建立一套标准化、可插拔的连接规范。你可以把它想象成AI世界的“USB协议”。在物理世界USB接口让键盘、鼠标、U盘可以即插即用在AI世界MCP协议的目标是让数据库、API、文件系统乃至一个命令行工具都能以统一的方式被AI模型安全、高效地调用。这个想法让我非常兴奋。它解决的不仅仅是技术集成问题更是AI应用开发范式的转变。过去我们总是围绕某个特定的大模型API来构建功能工具是硬编码进去的未来我们可以围绕MCP来构建让模型能力与工具能力解耦。一个具备强大推理能力的模型搭配上一系列通过MCP协议接入的专业工具搜索引擎、代码执行器、绘图工具等其解决问题的能力将呈指数级增长。接下来我将结合自己的实践拆解MCP协议的核心思想、实现细节并分享如何从零开始构建和集成一个MCP服务器。2. MCP协议核心设计思想拆解要理解MCP不能只把它看作又一个RPC或API规范。它的设计从头到尾都贯穿着为“AI调用”服务的特殊考量。2.1 核心目标标准化工具调用与上下文管理MCP协议的核心目标可以概括为两点标准化工具调用定义一套统一的模型AI与服务器工具提供方之间的通信方式包括如何发现工具、如何描述工具、如何调用工具以及如何返回结果。这消除了“方言”让AI无需学习每个工具独特的“说话方式”。高效的上下文管理AI模型尤其是大语言模型有上下文窗口的限制。MCP协议设计了一套资源Resources和提示Prompts的声明与读取机制允许服务器主动向模型声明“我这里有这些资料资源和预制问题提示”模型可以根据需要按需读取而不是一次性吞下所有可能用到的信息极大优化了上下文的使用效率。这背后的逻辑是AI模型客户端和工具服务器是平等的、松耦合的双方。服务器向客户端“广告”自己的能力工具列表和可提供的信息资源列表客户端根据当前任务选择调用合适的工具或读取相关资源并将结果整合到自己的思考与输出中。2.2 协议栈与通信模式MCP协议建立在JSON-RPC 2.0之上。选择JSON-RPC是因为它轻量、简单、跨语言支持广泛非常适合这种需要频繁、双向通信的场景。通信是全双工的通常通过标准输入输出stdio、WebSocket或SSEServer-Sent Events进行。这在实践中意味着你可以将一个MCP服务器作为一个独立的进程启动AI应用客户端通过管道与其通信就像在本地调用一个命令行工具一样自然。一个典型的会话流程如下初始化握手客户端与服务器建立连接后交换initialize请求与响应协商协议版本和基础能力。能力通告服务器通过notifications或requests主动向客户端发送tools/list、resources/list、prompts/list等信息宣告“我有什么”。按需调用客户端在推理过程中如果决定使用某个工具就向服务器发送tools/call请求。服务器执行工具逻辑如运行一段代码、调用一个API并将结果返回。按需读取客户端如果需要了解某个资源的详情如一个文件的内容就发送resources/read请求。服务器返回资源内容。会话结束通过notifications优雅地结束会话。这种设计将主动权部分交给了服务器让它能动态地更新自己可提供的工具和资源列表非常灵活。2.3 与传统API集成的本质区别你可能会问这和直接让AI调用HTTP API有什么区别区别巨大主要体现在抽象层次和安全性上。面向意图而非面向语法传统API集成你需要告诉AI“要查天气请向https://api.weather.com/v1/forecast发送一个GET请求参数是cityBeijing认证头是Authorization: Bearer YOUR_KEY。” 这要求AI理解HTTP协议、URL结构、查询参数和头部信息。而在MCP中服务器会声明一个名为get_weather的工具描述是“获取指定城市的天气情况”输入参数是一个city字符串。AI只需要理解“获取天气”这个意图并知道要提供城市名即可。所有的网络细节、认证逻辑都被封装在服务器内部。统一的安全边界MCP服务器是一个独立的进程或服务。所有对外部系统数据库、第三方API、文件系统的访问权限都集中在这个服务器上。你可以对这个服务器进行严格的安全审计和权限控制比如它只能读取某个特定目录只能访问内网某些API。AI客户端本身不需要也不应该持有访问这些敏感资源的密钥。这相当于建立了一个安全的“工具沙箱”。动态性与上下文感知MCP的资源Resources概念非常强大。例如一个连接GitHub的MCP服务器可以将“当前用户打开的Issue列表”定义为一个资源。当AI客户端需要了解当前工作上下文时它可以读取这个资源获取实时、结构化的数据而不是依赖可能过时或冗长的聊天历史。3. 核心组件深度解析理解了设计思想我们再来拆解MCP协议的三个核心组件工具Tools、资源Resources和提示Prompts。它们是服务器向AI客户端“自我介绍”的核心内容。3.1 工具ToolsAI的“可执行函数”工具是MCP协议中最重要的概念。它是对一个可执行操作的抽象描述。一个工具定义通常包含以下部分name: 工具的唯一标识符如search_web。description: 对人类和AI都友好的描述说明这个工具是做什么的。这个描述至关重要它是AI决定是否调用该工具的主要依据。描述应清晰、简洁并包含关键输入参数的暗示。inputSchema: 定义调用此工具所需的参数遵循JSON Schema规范。这相当于函数的参数列表和类型声明。示例一个简单的文件读取工具定义{ name: read_file, description: 读取指定路径的文本文件内容。, inputSchema: { type: object, properties: { file_path: { type: string, description: 要读取的文件的绝对路径或相对于服务器工作目录的路径。 } }, required: [file_path] } }当AI客户端需要读取文件时它会发送一个如下的调用请求{ jsonrpc: 2.0, id: 1, method: tools/call, params: { name: read_file, arguments: { file_path: /home/user/document.txt } } }服务器执行读取操作后返回结果{ jsonrpc: 2.0, id: 1, result: { content: [ { type: text, text: 这是文件的内容... } ] } }实操心得工具描述的“艺术”编写description时要站在AI的角度思考。避免使用晦涩的技术术语。好的描述应直接回答“在什么情况下我应该使用这个工具” 例如calculate就不如计算两个数的加减乘除结果来得明确。同时在inputSchema的description里详细说明每个参数的格式和约束能显著减少AI调用出错的概率。3.2 资源Resources结构化的上下文信息资源代表服务器可以提供的一段信息或内容它有一个唯一的URI来标识。资源不是主动推送给AI的而是“挂在那里”供AI在需要时按需查询resources/read。资源的典型用途包括提供静态参考项目README文档、API使用手册。提供动态上下文当前服务器状态、用户最近的操作记录、实时数据摘要如“今日待办事项”。分块大型内容一本电子书可以按章节定义为多个资源AI可以只读取当前相关的章节节省上下文。资源与工具的关键区别资源是“只读”的信息源而工具是“可执行”的操作。AI读取资源不会改变外部状态但调用工具可能会。3.3 提示Prompts预制的问题模板提示是服务器预定义的一些问题或指令模板AI客户端可以读取并直接使用或稍作修改后用于与用户交互。这有点像“快捷提问”。例如一个代码仓库的MCP服务器可以提供一个名为explain_recent_change的提示其内容可能是“请解释最近一次提交SHA: {{commit_hash}}引入了哪些更改并评估其风险。” AI客户端可以读取这个提示将其中的{{commit_hash}}替换为实际的提交哈希然后用来询问用户或直接用于分析。提示功能在构建高度领域特定的AI助手时非常有用它允许工具提供方将领域内最常问的问题模式固化下来提升交互效率。4. 动手实现一个MCP服务器理论说得再多不如动手写一个。我们来实现一个最简单的MCP服务器一个系统信息查询服务器。它提供一个工具来获取当前系统的负载情况。我们将使用Python因为其生态中有很好的MCP SDK支持。这里我选择官方推荐的mcpSDK。4.1 环境准备与项目初始化首先创建一个新的Python虚拟环境并安装依赖。# 创建并激活虚拟环境根据你的系统选择 python -m venv .venv source .venv/bin/activate # Linux/macOS # .venv\Scripts\activate # Windows # 安装MCP SDK pip install mcp注意mcp库是一个底层SDK。社区也有更高级的封装如mcp-server但为了理解原理我们从基础的开始。4.2 构建系统信息查询工具我们的服务器将提供一个名为get_system_load的工具。# server.py import asyncio import json import psutil # 需要安装pip install psutil from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client # 1. 定义我们的工具 tools [ { name: get_system_load, description: 获取当前系统的CPU、内存和磁盘使用率。, inputSchema: { type: object, properties: {}, # 这个工具不需要输入参数 required: [] } } ] async def handle_tool_call(name: str, arguments: dict) - dict: 处理工具调用的核心函数 if name get_system_load: # 使用psutil获取系统信息 cpu_percent psutil.cpu_percent(interval0.1) memory_info psutil.virtual_memory() disk_usage psutil.disk_usage(/) result_text f **系统负载报告**: - **CPU使用率**: {cpu_percent}% - **内存使用**: {memory_info.used / (1024**3):.2f} GB / {memory_info.total / (1024**3):.2f} GB ({memory_info.percent}%) - **磁盘使用 (根目录)**: {disk_usage.used / (1024**3):.2f} GB / {disk_usage.total / (1024**3):.2f} GB ({disk_usage.percent}%) # MCP要求返回特定格式的内容列表 return { content: [{type: text, text: result_text}] } else: raise ValueError(f未知工具: {name}) async def main(): # 2. 创建服务器参数使用标准输入输出作为传输层 server_params StdioServerParameters( commandpython, # 解释器 args[-u, __file__], # 以非缓冲模式运行当前脚本 ) # 3. 启动客户端会话在这个模式下当前脚本既是客户端也是逻辑处理者 async with stdio_client(server_params) as (read_stream, write_stream): session ClientSession(read_stream, write_stream) # 4. 初始化握手 await session.initialize() # 5. 通知客户端我们有哪些工具 await session.notify_tools_list_changed(tools) # 6. 进入主循环监听请求 async for message in session.channel: if message.method tools/call: # 处理工具调用请求 tool_name message.params[name] tool_args message.params.get(arguments, {}) try: result await handle_tool_call(tool_name, tool_args) # 发送成功响应 await session.send_success_response(message.id, result) except Exception as e: # 发送错误响应 await session.send_error_response(message.id, str(e)) # 可以添加对其他请求如resources/read的处理 else: # 忽略或处理其他类型的消息 pass if __name__ __main__: asyncio.run(main())这个服务器通过标准输入输出与客户端通信。它声明了一个工具并在收到该工具的调用请求时执行psutil库的查询逻辑并格式化返回结果。4.3 与AI客户端如Claude Desktop集成测试单独运行这个服务器是没意义的我们需要一个AI客户端来调用它。一个流行的测试方式是使用Claude Desktop应用它内置了MCP客户端支持。配置Claude Desktop找到Claude Desktop的配置文件夹。macOS:~/Library/Application Support/Claude/claude_desktop_config.jsonWindows:%APPDATA%\Claude\claude_desktop_config.json编辑配置文件在mcpServers部分添加我们的服务器配置。{ mcpServers: { system-info-server: { command: /path/to/your/.venv/bin/python, args: [/path/to/your/server.py] } } }关键点command必须指向你虚拟环境中的Python解释器绝对路径确保psutil库可用。args是脚本的绝对路径。重启Claude Desktop保存配置并重启应用。测试在Claude的聊天框中你可以直接问“当前的系统负载怎么样” Claude会识别到可用的get_system_load工具并在后台调用它然后将工具返回的结果整合到它的回复中。你会看到类似“我正在调用系统工具来获取信息...”的提示然后得到一份格式良好的系统负载报告。5. 构建复杂MCP服务器的进阶实践实现一个工具只是开始。一个实用的MCP服务器通常需要集成多个工具管理资源并处理更复杂的逻辑。5.1 多工具集成与组织一个服务器可以提供多个相关工具。例如一个“开发者助手”服务器可能包含search_code: 在代码库中搜索。run_tests: 执行特定测试套件。deploy_preview: 部署一个预览环境。在代码组织上建议为每个工具定义一个独立的处理函数并使用字典或装饰器进行映射保持主循环的简洁。tool_handlers { get_system_load: handle_get_system_load, search_logs: handle_search_logs, restart_service: handle_restart_service, } async def dispatch_tool_call(name, arguments): handler tool_handlers.get(name) if handler: return await handler(arguments) else: raise ValueError(fTool not found: {name})5.2 状态管理与资源声明服务器可能需要维护一些内部状态。例如一个数据库查询服务器在初始化时建立了连接池这个连接池需要在多个工具调用间共享。class DatabaseServer: def __init__(self, connection_string): self.pool create_pool(connection_string) self.resources [{ uri: resource://database/schema, name: 当前数据库Schema摘要, description: 主要数据表的名称和列信息。, mimeType: text/plain }] async def get_tools(self): return [...工具列表...] async def read_resource(self, uri): if uri resource://database/schema: # 动态查询数据库生成schema摘要 schema_summary await self.generate_schema_summary() return {contents: [{type: text, text: schema_summary}]}资源可以是静态的也可以是像上面这样动态生成的。AI客户端在需要了解数据库结构时会读取这个资源服务器实时查询并返回。5.3 错误处理与健壮性健壮的MCP服务器必须考虑各种错误情况工具参数验证在inputSchema中定义严格的JSON Schema只是第一步。在工具处理函数内部仍需对参数进行业务逻辑验证。外部依赖失败调用第三方API、数据库查询可能失败。必须使用try...except进行捕获并返回结构化的错误信息给客户端而不是让整个服务器崩溃。异步超时对于可能长时间运行的工具要设置超时机制防止阻塞。async def handle_external_api_call(arguments): try: async with aiohttp.ClientSession(timeoutaiohttp.ClientTimeout(total30)) as session: async with session.get(https://api.example.com/data) as resp: if resp.status 200: data await resp.json() return format_result(data) else: # 返回明确的错误信息帮助AI理解 return { content: [{ type: text, text: f请求外部API失败状态码{resp.status}。可能的原因服务暂时不可用或参数有误。 }], isError: True # MCP响应中可以包含错误标志 } except asyncio.TimeoutError: return {content: [{type: text, text: 请求超时请稍后重试。}], isError: True} except Exception as e: # 记录日志但返回用户友好的信息 logger.error(fAPI调用异常: {e}) return {content: [{type: text, text: 处理您的请求时遇到内部错误。}], isError: True}6. 实战集成真实世界API——天气查询服务器让我们构建一个更有实用价值的MCP服务器集成一个免费的天气API。我们将使用wttr.in这个简单的服务。6.1 设计工具与选择API工具设计名称get_weather描述获取全球指定城市当前天气状况和未来几天的简要预报。输入参数city(字符串必需)days(数字可选默认为3表示预报天数)。API选择wttr.in提供简洁的APIhttps://wttr.in/{city}?formatj1返回JSON格式数据。它无需认证适合演示。6.2 服务器实现代码# weather_mcp_server.py import asyncio import aiohttp from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client import json async def fetch_weather(city: str, days: int 3) - str: 调用wttr.in API获取天气数据 url fhttps://wttr.in/{city} params {format: j1} if days: params[days] days async with aiohttp.ClientSession() as session: try: async with session.get(url, paramsparams, timeout10) as response: if response.status 200: data await response.json() return parse_weather_data(data, city) else: return f无法获取{city}的天气信息API返回状态码{response.status}。 except asyncio.TimeoutError: return f请求天气信息超时请检查网络或稍后重试。 except Exception as e: return f获取天气信息时发生错误{str(e)} def parse_weather_data(data: dict, city: str) - str: 解析wttr.in返回的JSON数据格式化成易读文本 current data[current_condition][0] forecast data[weather] summary f**{city} 当前天气**\n summary f- 温度: {current[temp_C]}°C (体感 {current[FeelsLikeC]}°C)\n summary f- 状况: {current[weatherDesc][0][value]}\n summary f- 湿度: {current[humidity]}%\n summary f- 风速: {current[windspeedKmph]} km/h\n summary f- 风向: {current[winddir16Point]}\n\n summary f**未来{len(forecast)}天预报**\n for day in forecast[:3]: # 只显示最近3天 date day[date] max_temp day[maxtempC] min_temp day[mintempC] condition day[hourly][4][weatherDesc][0][value] # 取中午时段的描述 summary f- {date}: {condition}, 气温 {min_temp}~{max_temp}°C\n return summary async def handle_tool_call(name: str, arguments: dict) - dict: if name get_weather: city arguments.get(city, ).strip() if not city: return { content: [{type: text, text: 请提供要查询的城市名称例如Beijing 或 London。}], isError: True } days min(max(int(arguments.get(days, 3)), 1), 7) # 限制在1-7天 weather_report await fetch_weather(city, days) return { content: [{type: text, text: weather_report}] } raise ValueError(f未知工具: {name}) tools [ { name: get_weather, description: 获取全球指定城市当前天气状况和未来几天的简要预报。, inputSchema: { type: object, properties: { city: { type: string, description: 城市名称支持英文名如London或拼音如Beijing。 }, days: { type: number, description: 预报天数默认为3范围1-7。, default: 3 } }, required: [city] } } ] async def main(): # ... 与之前示例相同的通信主循环框架 ... server_params StdioServerParameters(commandpython, args[-u, __file__]) async with stdio_client(server_params) as (read, write): session ClientSession(read, write) await session.initialize() await session.notify_tools_list_changed(tools) async for message in session.channel: if message.method tools/call: tool_name message.params[name] tool_args message.params.get(arguments, {}) try: result await handle_tool_call(tool_name, tool_args) await session.send_success_response(message.id, result) except Exception as e: await session.send_error_response(message.id, str(e)) if __name__ __main__: asyncio.run(main())6.3 配置与使用将上述代码保存为weather_mcp_server.py。安装依赖pip install aiohttp mcp。参照4.3节将其添加到Claude Desktop的MCP服务器配置中。重启Claude后你就可以直接问“上海明天天气怎么样” 或 “What‘s the weather in Paris for the next 5 days?”。Claude会自动调用get_weather工具并呈现结果。这个例子展示了如何将一个简单的公共API封装成AI可安全、规范调用的工具。你可以举一反三将公司内部的CRM、ERP、监控系统API都以此方式封装瞬间为你的AI助手赋予强大的“企业级”能力。7. 调试、问题排查与性能优化开发MCP服务器过程中难免会遇到问题。这里分享一些实用的调试和优化技巧。7.1 常见问题与排查清单问题现象可能原因排查步骤Claude Desktop无法加载服务器1. 配置文件路径错误。2. Python解释器或脚本路径错误。3. 虚拟环境依赖未安装。4. 服务器脚本启动即报错。1. 检查claude_desktop_config.json格式和路径。2. 在终端手动运行配置中的command和args看能否启动Python并执行脚本。3. 确保虚拟环境已激活且安装了mcp等必要包。4. 在脚本开头添加print(“Server starting...”)并查看Claude Desktop日志通常可在应用菜单中找到。AI不调用工具1. 工具描述不清晰。2. 工具名称或参数与AI理解不匹配。3. 服务器初始化失败工具列表未成功发送。1. 优化description使其更贴近自然语言查询意图。2. 使用更通用的工具名和参数名如city而非location_name。3. 在服务器初始化后添加日志确认notify_tools_list_changed被调用。工具调用返回错误1. 参数格式错误或缺失。2. 服务器端处理逻辑异常如API调用失败。3. 网络或权限问题。1. 在handle_tool_call函数内部首先打印或记录收到的arguments验证数据。2. 用try...except包裹核心逻辑并返回详细的错误信息。3. 单独测试服务器内部函数排除外部依赖问题。通信超时或中断1. 工具执行时间过长。2. 服务器进程崩溃。3. 标准输入输出缓冲区问题。1. 为长时间操作设置超时或设计为异步非阻塞模式。2. 增强服务器代码的健壮性捕获所有未处理异常。3. 确保Python以-u无缓冲模式运行。7.2 调试技巧使用独立测试客户端不依赖Claude Desktop自己写一个简单的测试客户端能极大提升开发效率。# test_client.py import asyncio import json import sys async def test_server(): # 启动服务器进程 proc await asyncio.create_subprocess_exec( sys.executable, -u, weather_mcp_server.py, stdinasyncio.subprocess.PIPE, stdoutasyncio.subprocess.PIPE, stderrasyncio.subprocess.PIPE ) async def send_request(method, paramsNone, id1): request {jsonrpc: 2.0, id: id, method: method} if params: request[params] params message json.dumps(request) \n proc.stdin.write(message.encode()) await proc.stdin.drain() # 读取响应 line await proc.stdout.readline() return json.loads(line.decode().strip()) # 1. 初始化 init_response await send_request(initialize, {protocolVersion: 0.1}) print(初始化响应:, init_response) # 2. 模拟客户端接收工具列表通知这里需要根据服务器实际发送的消息调整 # 通常服务器会主动发送通知我们这里简化直接调用工具列表请求如果协议支持 # 假设我们直接调用工具 print(\n--- 测试工具调用 ---) call_response await send_request(tools/call, { name: get_weather, arguments: {city: London, days: 2} }, id2) print(工具调用响应:, json.dumps(call_response, indent2, ensure_asciiFalse)) proc.terminate() await proc.wait() if __name__ __main__: asyncio.run(test_server())这个客户端模拟了MCP协议的基本交互你可以快速验证服务器的核心逻辑是否正确而无需反复重启Claude Desktop。7.3 性能优化与最佳实践连接池与资源复用对于需要连接数据库、外部API的服务器在初始化时创建连接池或会话并在整个服务器生命周期内复用避免为每个工具调用都建立新连接。异步编程务必使用asyncio等异步框架。任何I/O操作网络请求、文件读写、数据库查询都应该是异步的以防止阻塞整个服务器影响其他并发的工具调用请求。结果缓存对于耗时长、更新频率不高的操作如获取全量数据列表可以考虑在服务器内存中设置短期缓存如functools.lru_cache但要注意缓存失效策略。工具粒度设计工具不宜过大或过小。一个工具应完成一个逻辑上独立、完整的操作。例如“创建用户并发送欢迎邮件”最好拆分成create_user和send_welcome_email两个工具这样AI可以更灵活地组合使用。详细的错误信息工具返回的错误信息应尽可能对AI和最终用户友好。避免返回原始的异常堆栈而是转换为如“无法连接到数据库请检查网络或联系管理员”这样的自然语言描述。8. MCP生态与未来展望MCP协议由Anthropic公司提出并推动但其设计是开放和协议无关的。这意味着任何遵循该协议的客户端和服务器都可以互操作。目前除了Claude Desktop一些开源的AI应用框架如Continue、Cursor等也开始支持MCP。生态正在快速成长官方与社区服务器已经出现了许多开源的MCP服务器用于连接GitHub、Notion、Slack、PostgreSQL、甚至命令行终端。开发工具除了Python SDK社区也正在为Node.js、Rust、Go等语言开发SDK降低开发门槛。应用场景从个人效率助手管理待办、查询信息到专业领域Agent代码审查、客服答疑、数据分析MCP正在成为连接大模型与专业能力的“桥梁协议”。我个人的体会是MCP协议的价值在于它定义了一个清晰的“边界”。在这个边界内AI模型负责理解、规划和决策边界外专业的工具服务器负责安全、可靠地执行。这种分工协作的模式比试图让一个模型学会所有事情的“全能巨无霸”路径在当下看来更务实、更安全也更具可扩展性。它让AI应用真正开始像搭积木一样可以灵活地组合不同的能力模块。最后一个小技巧当你设计MCP工具时不妨把自己想象成在为一个“超级实习生”编写工作手册。这个实习生AI非常聪明但缺乏对具体系统的了解。你的工具描述就是给它的清晰指令卡告诉它“在什么情况下用什么参数调用哪个功能”。手册写得好实习生就能快速上手创造巨大价值。MCP协议就是这套手册的标准化格式。
返回列表