ARTICLE DETAIL

资讯详情

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

从零实现MCP协议:为AI Agent打造标准化工具接口

从零实现MCP协议:为AI Agent打造标准化工具接口 1. 项目缘起当AI需要“手”和“眼睛”最近在折腾AI Agent发现一个挺普遍的问题大模型本身是个“超级大脑”天文地理无所不知但让它干点具体事比如去数据库里查个数据、调用个内部API、或者操作一下本地文件它就有点“抓瞎”了。你得像教小孩一样给它写一堆复杂的函数描述Function Calling还得处理各种授权、参数解析和错误处理一个Agent项目大半精力都花在给AI“造工具”上了。这让我想起了早期的计算机没有操作系统和驱动程序每个程序都得自己管理硬件。现在AI应用开发似乎也陷入了类似的境地。直到我遇到了MCPModel Context Protocol。简单说MCP就是一个标准协议它能让AI模型比如Claude、GPT安全、标准化地“连接”到外部工具、数据源和系统就像给AI装上了统一的“USB接口”。网上关于MCP的概念讨论很多但真正“从零开始”、在协议层面动手接一个的实践分享却很少。大家要么在用现成的MCP Server要么在等大厂出方案。但我觉得理解协议本身亲手实现一次才是掌握它的最好方式。这次我就以把一个简单的本地文件搜索工具变成AI能力为例带你走一遍完整的MCP接入流程。你会发现它没那么神秘核心就是一套清晰的HTTPSSE通信规范。2. 拆解MCP它到底是什么又解决了什么在开始写代码之前我们必须先搞清楚MCP协议到底规定了什么。你可以把它想象成AI世界里的“RESTful API”或“gRPC”但它服务的对象是AI模型而非人类用户直接调用的前端。MCP的核心思想是标准化工具描述与调用。在没有MCP之前如果你想在AI Agent中使用一个工具比如一个查询天气的API你需要为这个工具编写一个函数并在代码中实现它。用特定的格式如OpenAI的Function Calling格式向大模型描述这个工具的名称、参数、说明。在收到大模型的调用请求后解析JSON调用函数处理异常再将结果格式化返回给模型。 这个过程不仅繁琐而且工具描述格式各异无法在不同模型和平台间通用。MCP通过定义一套标准的协议将上述过程解耦和标准化MCP Server工具提供方负责实际执行工具操作如读文件、查数据库。它启动后会通过标准方式向客户端“宣告”自己有哪些能力Tools、能提供哪些数据Resources。MCP ClientAI应用方通常是集成了AI模型的应用程序如Claude Desktop、自定义的Agent框架。它负责连接Server获取工具列表并在模型需要时按照协议格式调用Server上的工具。SSEServer-Sent Events通信Client和Server之间通过HTTP和SSE进行双向通信。Client发送JSON-RPC格式的请求Server返回流式或非流式响应。那么MCP具体解决了哪些痛点呢工具发现与描述的标准化Server启动后自动广播能力Client无需硬编码工具信息。安全边界清晰AI模型只能调用Server明确暴露的工具且Server运行在独立的进程或环境中与核心应用隔离提升了安全性。开发效率与复用性一旦一个工具被实现为MCP Server它可以被任何兼容MCP的Client如Claude、Cursor、Windmill直接使用无需为每个平台重复适配。复杂的工具组合成为可能模型可以链式调用多个MCP Server提供的工具完成复杂任务而开发者只需关注单个工具的实现。理解了这些我们再来看协议的具体内容。MCP的通信基于JSON-RPC 2.0所有消息都是JSON对象。核心的“方法”包括initialize连接建立后的握手。tools/list客户端获取服务器提供的所有工具列表。tools/call客户端调用某个工具。notifications服务器主动通知客户端如工具执行进度。我们接下来的实践就将围绕实现这些核心方法展开。3. 实战准备环境与第一个MCP Server理论说得再多不如动手写一行代码。我们的目标是创建一个最简单的MCP Server它提供一个工具search_files可以根据关键词搜索当前目录下的文本文件内容。3.1 环境搭建与依赖选择首先确保你安装了Python 3.8。我们将使用官方推荐的mcpSDK 来简化开发它处理了底层的协议通信和类型校验。# 创建一个新的项目目录并进入 mkdir mcp-file-search-server cd mcp-file-search-server # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate # 安装 mcp 库 pip install mcp注意mcp库是一个快速上手的SDK。如果你想更深入地理解协议也可以使用任何能启动HTTP服务器、处理SSE的库如FastAPI、Flask来自行实现但那会复杂很多。对于初次实践强烈建议使用官方SDK。3.2 实现核心工具逻辑在动手接入协议前我们先抛开MCP想清楚这个文件搜索工具本身该怎么实现。这是一个纯粹的Python功能。# file_searcher.py import os from pathlib import Path from typing import List, Tuple class FileSearcher: 一个简单的本地文件内容搜索器 def __init__(self, root_dir: str .): self.root_dir Path(root_dir).resolve() def search(self, keyword: str, file_extensions: List[str] None) - List[Tuple[str, str, int]]: 在指定目录下递归搜索包含关键词的文本文件。 参数: keyword: 要搜索的关键词 file_extensions: 限制搜索的文件后缀如 [.txt, .py, .md]。为None则搜索所有文件。 返回: 列表每个元素为 (文件路径, 匹配行内容, 行号) if not keyword: return [] results [] # 默认搜索常见文本文件 if file_extensions is None: file_extensions [.txt, .py, .md, .json, .yaml, .yml, .csv, .html, .js, .ts] # 遍历目录 for ext in file_extensions: for file_path in self.root_dir.rglob(f*{ext}): if not file_path.is_file(): continue try: # 以文本模式读取避免二进制文件 with open(file_path, r, encodingutf-8, errorsignore) as f: for line_num, line in enumerate(f, 1): if keyword.lower() in line.lower(): # 返回相对路径更清晰 rel_path file_path.relative_to(self.root_dir) results.append((str(rel_path), line.strip(), line_num)) except (UnicodeDecodeError, PermissionError, OSError): # 跳过无法读取的文件 continue return results这个FileSearcher类就是我们的“工具”内核。它不包含任何MCP相关的代码只负责纯粹的搜索业务逻辑。这种分离非常重要保证了工具核心的独立性和可测试性。3.3 将其“包装”成MCP Server现在我们要用mcpSDK 将这个工具“暴露”出去。核心是创建一个Server实例并向其注册工具。# server.py import asyncio from typing import Any from mcp import ClientSession, Server, StdioServerParameters from mcp.types import Tool, TextContent, CallToolResult from file_searcher import FileSearcher # 初始化我们的工具实例 searcher FileSearcher() async def handle_search_files(arguments: dict[str, Any]) - CallToolResult: 处理 search_files 工具的调用。 这个函数将被MCP SDK在收到客户端调用请求时自动触发。 # 1. 从客户端请求中解析参数 keyword arguments.get(keyword, ) # 参数可以是可选或带默认值的 extensions arguments.get(file_extensions, [.txt, .py, .md]) # 2. 调用核心工具逻辑 print(f[Server] 正在搜索关键词: {keyword}, 文件类型: {extensions}) search_results searcher.search(keyword, extensions) # 3. 将结果格式化为MCP协议要求的格式 if not search_results: content TextContent(typetext, textf未找到包含关键词 {keyword} 的文件。) else: # 将结果组织成易读的文本 result_lines [f找到 {len(search_results)} 处匹配] for file_path, line_content, line_num in search_results[:10]: # 限制前10条避免过长 result_lines.append(f- {file_path} 第{line_num}行: {line_content}) if len(search_results) 10: result_lines.append(f... 以及另外 {len(search_results) - 10} 处匹配。) content TextContent(typetext, text\n.join(result_lines)) # 4. 返回结果 return CallToolResult(content[content]) async def main(): # 创建MCP Server实例 server Server() # 定义我们要暴露的工具 search_tool Tool( namesearch_files, description在项目目录中搜索包含特定关键词的文本文件。, inputSchema{ type: object, properties: { keyword: { type: string, description: 需要搜索的关键词 }, file_extensions: { type: array, items: {type: string}, description: 限制搜索的文件后缀列表例如 [.py, .md]。默认为常见文本文件后缀。, default: [.txt, .py, .md] } }, required: [keyword] # keyword是必填参数 } ) # 将工具注册到Server server.tools.add_tool(search_tool, handle_search_files) # 配置Server通过标准输入输出(stdio)通信 # 这是MCP Server最常见的运行方式由Client如Claude Desktop启动和管理。 server_params StdioServerParameters( commandpython, # 解释器 args[server.py] # 脚本路径这里就是自身 ) # 也可以选择用HTTP Server模式独立运行监听端口 # 但为了与主流Client兼容我们先使用stdio模式。 print([Server] MCP 文件搜索服务器已初始化等待连接...) # 运行Server开始监听请求 async with server.run_stdio(server_params) as session: # 这里会阻塞直到Client断开连接 await session.wait_for_disconnect() if __name__ __main__: asyncio.run(main())这段代码是MCP Server的核心。我们做了以下几件事定义工具Tool使用Tool类详细描述了search_files工具包括它的名字、描述以及最重要的inputSchema。这个Schema遵循JSON Schema标准它告诉AI模型这个工具需要什么参数、参数是什么类型、有何描述。这是AI能正确调用工具的关键。绑定处理函数将handle_search_files函数与search_files工具绑定。当Client调用该工具时这个函数就会被执行。启动Server通过run_stdio方法以标准输入输出流的方式启动服务器。这是MCP的典型部署方式由客户端进程启动并管理Server进程的生命周期。现在一个最简单的MCP Server就完成了。你可以运行python server.py但它会等待一个MCP Client来连接它目前我们还看不到效果。接下来我们需要一个Client来测试。4. 连接与测试打造一个简易MCP Client为了验证我们的Server是否工作正常我们需要一个Client。我们可以写一个简单的脚本模拟AI应用如Claude Desktop的行为来连接并调用我们的Server。4.1 实现一个测试Client# test_client.py import asyncio import json from mcp import ClientSession, StdioServerParameters import subprocess import sys async def test_mcp_server(): # 1. 配置如何启动Server进程stdio模式 server_params StdioServerParameters( commandsys.executable, # 使用当前Python解释器 args[server.py] ) # 2. 启动Server进程并建立会话 print([Client] 正在启动并连接MCP Server...) async with ClientSession(server_params) as session: # 3. 初始化连接握手 await session.initialize() print([Client] 连接初始化成功。) # 4. 列出Server提供的所有工具 tools await session.list_tools() print(f[Client] 服务器提供了 {len(tools.tools)} 个工具:) for tool in tools.tools: print(f - {tool.name}: {tool.description}) # 5. 调用 search_files 工具 print(\n[Client] 正在调用 search_files 工具...) try: result await session.call_tool( tool_namesearch_files, arguments{keyword: MCP, file_extensions: [.py, .md]} ) # 6. 处理并打印结果 if result.content: for content_item in result.content: if content_item.type text: print(f[工具返回结果]:\n{content_item.text}) else: print(f[工具返回了非文本内容]: {content_item}) else: print([工具调用完成但无内容返回。]) except Exception as e: print(f[Client] 工具调用失败: {e}) print(\n[Client] 测试完成断开连接。) if __name__ __main__: asyncio.run(test_mcp_server())运行这个测试客户端python test_client.py。你会看到类似以下的输出[Client] 正在启动并连接MCP Server... [Server] MCP 文件搜索服务器已初始化等待连接... [Client] 连接初始化成功。 [Client] 服务器提供了 1 个工具: - search_files: 在项目目录中搜索包含特定关键词的文本文件。 [Server] 正在搜索关键词: MCP, 文件类型: [.py, .md] [Client] 正在调用 search_files 工具... [工具返回结果]: 找到 3 处匹配 - server.py 第12行: from mcp import ClientSession, Server, StdioServerParameters - server.py 第13行: from mcp.types import Tool, TextContent, CallToolResult - README.md 第1行: # MCP 文件搜索服务器示例 [Client] 测试完成断开连接。成功了我们的Client成功启动了Server获取了工具列表并调用了search_files工具Server也正确地执行了搜索并返回了结果。整个通信过程对开发者是透明的我们只需要关注工具的实现和调用。4.2 深入协议通信看看背后发生了什么为了更深入理解我们可以给Server和Client加上详细的日志看看它们之间到底传递了哪些JSON-RPC消息。修改Server和Client的代码在关键节点打印收发信息。在Server的handle_search_files函数开头加一句print(f“[Server] 收到调用参数: {arguments}”)。 在Client的call_tool前后打印arguments和result的原始内容。你会看到通信的本质是这样的Client - Server (初始化):{jsonrpc: 2.0, id: 1, method: initialize, params: {...}}Server - Client (响应):{jsonrpc: 2.0, id: 1, result: {...}}Client - Server (列出工具):{jsonrpc: 2.0, id: 2, method: tools/list}Server - Client (响应):{jsonrpc: 2.0, id: 2, result: {tools: [{...}]}}Client - Server (调用工具):{jsonrpc: 2.0, id: 3, method: tools/call, params: {name: search_files, arguments: {keyword: MCP}}}Server - Client (返回结果):{jsonrpc: 2.0, id: 3, result: {content: [{type: text, text: 找到 ...}]}}这就是MCP协议的骨架。mcpSDK帮我们封装了所有这些消息的序列化、反序列化和发送接收。5. 进阶与集成让AI真正使用你的工具通过了自制Client的测试说明我们的MCP Server在协议层面是健康的。但这还不够我们的终极目标是让像Claude、GPT-4这样的AI模型能使用它。这就需要将我们的Server集成到真正的MCP Client环境中。5.1 集成到Claude DesktopClaude Desktop是官方支持MCP的客户端之一配置起来相对简单。找到Claude的MCP配置文件。它的位置通常如下macOS:~/Library/Application Support/Claude/claude_desktop_config.jsonWindows:%APPDATA%\Claude\claude_desktop_config.jsonLinux:~/.config/Claude/claude_desktop_config.json编辑这个JSON文件。如果文件不存在就创建它。我们需要在其中添加一个mcpServers配置项指向我们的Python脚本。{ mcpServers: { file-search: { command: /absolute/path/to/your/venv/bin/python, args: [/absolute/path/to/your/project/server.py], env: { PYTHONPATH: /absolute/path/to/your/project } } } }关键提示command必须是你虚拟环境中Python解释器的绝对路径。使用which python(macOS/Linux) 或where python(Windows在激活的虚拟环境中) 来获取。args中的脚本路径也必须是绝对路径。env中的PYTHONPATH确保脚本能找到你项目中的其他模块如file_searcher.py。配置完成后需要完全重启Claude Desktop应用。验证集成。重启Claude后新建一个对话。如果你在输入框里输入“/”应该能看到一个可用的工具列表其中包含我们定义的search_files。你可以直接告诉Claude“请用search_files工具帮我找一下项目中提到‘MCP’的地方。” Claude会理解你的指令自动调用工具并将结果呈现在对话中。5.2 处理复杂场景与错误我们的第一个工具很简单但现实中的工具可能更复杂。MCP协议也考虑到了这些情况。1. 流式响应Streaming对于耗时长或需要持续输出的工具如执行一个长时间运行的脚本、监控日志MCP支持流式返回结果。在Server的处理函数中你可以返回一个异步生成器async generator分多次发送TextContent或ImageContent。Client和AI就能看到逐步输出的过程体验更好。2. 工具调用错误工具执行可能会出错如参数无效、资源不存在、网络超时。MCP要求Server通过JSON-RPC错误响应来通知Client。在handle_search_files函数中我们应该用try...except捕获异常并抛出mcp.MCPError或返回包含错误信息的CallToolResult。from mcp import MCPError async def handle_search_files(arguments: dict[str, Any]) - CallToolResult: try: keyword arguments.get(keyword, ) if not keyword or len(keyword.strip()) 0: # 抛出标准错误AI会收到清晰的错误信息 raise MCPError(code-32602, message参数 keyword 不能为空。) # ... 正常处理逻辑 ... except MCPError: raise # 重新抛出MCP错误 except Exception as e: # 捕获其他未预期错误避免Server崩溃 raise MCPError(code-32000, messagef工具执行内部错误: {str(e)})3. 提供资源Resources除了工具主动调用MCP Server还可以声明“资源”Resources。资源是Server能提供的静态或动态数据URIAI模型可以直接读取它们的内容而无需调用工具。例如一个数据库Server可以提供一个sqlite:///mydb.db的资源AI可以直接“看到”数据库模式。这通过resources/list和resources/read方法实现。5.3 调试技巧与常见问题在集成过程中你肯定会遇到问题。以下是一些实用的调试方法查看Claude Desktop日志这是最直接的。在Claude Desktop中通常可以通过Help-View Logs或Debug菜单找到日志文件。搜索“MCP”或你的Server名字能看到连接、初始化、调用失败的详细信息。独立测试Server使用我们之前写的test_client.py进行测试确保基础功能在纯净环境下是好的。检查路径和权限这是最常见的问题。确保配置文件中所有路径都是绝对路径并且Claude Desktop进程有权限执行该Python解释器和脚本。验证JSON配置配置文件必须是有效的JSON且结构正确。一个多余的逗号都可能导致整个配置被忽略。Server进程自查可以在Server脚本开头加入print(“Server started with PID:”, os.getpid())然后在活动监视器或任务管理器中查看该进程是否被正确启动。我踩过的一个坑是在macOS上Claude Desktop默认以沙盒模式运行对文件系统的访问受限。如果你的工具需要访问特定目录如用户文档或下载文件夹可能会因权限问题失败。这时需要调整工具逻辑或通过用户明确授权的方式来处理。6. 举一反三还能接入什么文件搜索只是一个起点。理解了MCP的核心后你可以将几乎任何能力封装成Server。以下是一些更有想象力的方向内部API网关将公司内部的各种查询、审批、数据检索API统一封装成一个MCP Server。市场部的同事可以直接问AI“上个季度华东区的销售数据如何” AI通过MCP调用内部BI系统的API拿到数据并生成总结。开发工具链创建一个“开发助手”Server提供诸如run_tests运行单元测试、check_dependencies检查依赖更新、git_operation执行简单的git命令、lint_code代码检查等工具。程序员在IDE里就可以用自然语言让AI助手执行这些重复性任务。硬件与IoT为智能家居设备、实验室仪器编写MCP Server。研究员可以对AI说“把培养箱的温度调到37度”AI通过MCP协议将指令下发到具体的设备驱动。复杂工作流触发器将Zapier、n8n或自定义的复杂工作流封装成一个工具trigger_workflow。AI可以根据对话上下文判断并触发相应的自动化流程。设计一个“好用”的MCP工具关键在于工具描述inputSchema的清晰度。你要像给一个完全不懂技术的实习生写说明书一样描述每个参数是干什么的、期望的格式是什么、有哪些可选值。例如一个“发送邮件”的工具它的recipient参数描述应该是“收件人的电子邮件地址多个地址用分号隔开”而不仅仅是“收件人”。7. 协议之外的思考MCP的边界与未来通过这次从零实践我们看到了MCP的强大与简洁。但它也不是银弹有它的适用边界。优势标准化统一了AI与工具交互的“语言”避免了重复造轮子。安全性Server独立运行权限可控工具暴露范围明确。组合性多个Server可以同时被一个AI使用能力可以像乐高一样拼接。语言无关Server可以用任何语言编写Python、Go、Rust、Node.js只要遵循协议即可。当前限制与考量性能开销每个工具调用都涉及进程间通信IPC和可能的网络开销对于超低延迟的场景需要优化。状态管理MCP Server理论上应该是无状态的。如果工具需要维护会话状态如多轮对话需要Client在调用时传递上下文或在Server端实现某种会话管理这增加了复杂性。工具编排逻辑在AI目前是否调用工具、按什么顺序调用完全由AI模型决定。这对于复杂、有严格顺序依赖的任务链来说可能不够可靠。未来可能需要更上层的“编排层”来管理。生态早期虽然发展很快但成熟的、生产可用的Server和Client生态还在建设中可能会遇到兼容性问题。在我看来MCP最大的价值在于它定义了一个清晰的“人机接口”。过去我们为人类设计GUI或CLI现在我们开始为AI设计“模型接口”Model Interface。这要求我们转变思维从“如何让用户点击”变成“如何让AI理解并正确使用”。这不仅仅是技术实现更是一种新的交互设计范式。亲手实现一遍协议哪怕是最简单的版本这种理解是读十篇文档都无法替代的。它让你看清了魔法背后的齿轮是如何咬合的。下次当你再看到某个炫酷的AI应用能操作各种软件时你大概能猜到背后很可能就运行着几个安静的MCP Server正在按照一套清晰的协议默默地为AI提供着“手”和“眼睛”。
返回列表