1. 项目缘起当AI需要“动手”时我们缺了什么最近在折腾各种AI应用尤其是想让它帮我处理点本地文件、查查数据库或者控制一下智能家居设备。我发现一个挺普遍的问题你跟大模型聊得天花乱坠但它一遇到需要“动手”操作现实世界的事情就立刻哑火了。它知道该“调用某个API”但具体怎么调用、参数怎么传、返回结果怎么解析它给不了你一个能直接运行的方案。这感觉就像你有一个无所不知的军师但他没有手你得自己把他的话翻译成具体的动作指令。这就是“工具调用”能力的核心痛点。我们当然可以给大模型写一堆“提示词”教它某个工具的用法但这种方式笨重、不通用而且一旦工具更新提示词也得跟着改。有没有一种更优雅、更标准化的方式让AI能像人类安装驱动一样“即插即用”地获得新工具的能力这就是我接触到MCPModel Context Protocol的契机。它不是某个具体的AI产品而是一个协议。你可以把它理解为AI世界的“USB协议”。USB协议定义了键盘、鼠标、U盘等外设如何与电脑通信而MCP则定义了各种工具比如文件系统、数据库、搜索引擎如何以一种标准化的方式向AI模型“自我介绍”并与之交互。这个项目的目标很明确“从零接入MCP”。我们不依赖任何现成的、封装好的SDK或平台而是直接深入到协议层亲手实现一个MCP Server把任意一个我们自己的工具比如一个简单的本地文件查询工具包装起来变成AI可以理解和调用的“能力”。这个过程会让你彻底明白AI的“工具调用”背后数据到底是怎么流转的协议是如何协商的从而获得真正的掌控感。2. 拆解MCP协议的核心三要素与工作流在动手写代码之前我们必须先搞清楚MCP协议到底规定了什么。抛开复杂的术语MCP的核心可以概括为三件事工具能做什么声明、AI想做什么调用、以及工具做完之后的结果响应。整个交互建立在JSON-RPC 2.0这个轻量级的远程调用协议之上通信通常通过标准输入输出stdio或HTTP进行非常适合与AI模型进程集成。2.1 核心概念资源、工具与提示词模板MCP定义了三种主要的能力类型你可以把它们看作是暴露给AI的三类接口资源Resources代表AI可以“读取”的信息源。比如一个文件、一个数据库表的只读视图、一个网页的URL。AI可以获取read这些资源的内容来丰富自己的上下文。例如你可以把一个/docs/api.md路径声明为一个资源AI需要时就能读取这个文件的内容。工具Tools代表AI可以“执行”的操作。这是最核心的部分。一个工具包含名称、描述、输入参数inputSchema的定义。AI根据描述决定是否调用并按照inputSchema的格式传入参数。比如一个“搜索文件”的工具输入参数可能是{“query”: “字符串”}。提示词模板Prompts预定义的、参数化的提示词片段。AI可以获取get这些模板并填入具体参数快速生成高质量的查询或指令。这有助于标准化AI的提问方式。对于本次“把任意工具变成AI能力”的目标我们重点关注工具Tools的实现。这是让AI从“知道”到“做到”的关键跳板。2.2 交互流程一次完整的工具调用是如何发生的假设我们已经实现了一个MCP Server它向AI客户端比如Claude Desktop、Cursor等支持MCP的AI应用声明了一个叫read_local_file的工具。整个调用流程如下初始化与能力通告AI客户端启动我们的MCP Server。Server发送initialize请求完成握手。随后Server主动发送notify消息或响应list_tools请求告诉客户端“嗨我这里有这些工具可用read_local_file它的功能是…它需要这些参数…”。AI决策与调用用户向AI提问“帮我看看/home/user/project/README.md里写了什么”AI模型根据内部逻辑判断需要调用read_local_file工具并构造一个符合inputSchema的调用请求call_tool发送给Server。请求里包含了工具名和参数{“path”: “/home/user/project/README.md”}。Server执行与返回我们的Server收到call_tool请求解析参数执行真正的本地文件读取逻辑。然后将结果封装成标准格式返回给AI客户端。结果包括content文件内容和可能的isError标志。AI整合与回复AI客户端收到工具执行结果将其作为上下文提供给AI模型。模型据此生成最终回答给用户“这个README文件的内容是…”。整个过程中AI模型不需要知道文件是用Python的open()函数读的还是用系统命令cat读的。它只关心协议约定好的接口。这就是协议的价值解耦与标准化。3. 实战用Python从零构建一个MCP Server理论清楚了我们开始动手。我们将用Python实现一个最简单的MCP Server它提供一个工具get_weather获取天气。虽然这个工具本身只是返回一个模拟数据但实现过程完整涵盖了MCP Server的所有核心环节。注意我们选择Python是因为其生态丰富且易于理解MCP协议本身是语言无关的你可以用Node.js、Go、Rust等任何语言实现。3.1 环境准备与项目结构首先确保你的Python环境在3.8以上。我们不需要任何特殊的MCP官方库因为我们要从协议层实现但会用到pydantic来进行严谨的数据验证和序列化这能省去大量手动校验的麻烦。# 创建项目目录并初始化虚拟环境 mkdir mcp-weather-server cd mcp-weather-server python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate # 安装核心依赖 pip install pydantic项目目录结构如下mcp-weather-server/ ├── server.py # MCP Server主逻辑 ├── protocol.py # MCP协议相关数据模型定义 └── requirements.txt # 依赖列表requirements.txt内容pydantic2.03.2 定义协议数据模型protocol.py这是最关键的一步我们根据MCP协议规范定义所有在通信中会用到的JSON-RPC请求和响应模型。使用pydantic能确保出入参的格式绝对正确。# protocol.py from typing import Any, Dict, List, Optional, Union from pydantic import BaseModel, Field # -------------------- 基础JSON-RPC模型 -------------------- class JSONRPCRequest(BaseModel): JSON-RPC 2.0 请求基础模型 jsonrpc: str 2.0 id: Optional[Union[int, str]] None method: str params: Optional[Union[Dict[str, Any], List[Any]]] None class JSONRPCResponse(BaseModel): JSON-RPC 2.0 响应基础模型 jsonrpc: str 2.0 id: Optional[Union[int, str]] None result: Optional[Any] None error: Optional[Dict[str, Any]] None # -------------------- MCP 特定模型 -------------------- class ToolInputSchema(BaseModel): 工具输入参数的模式定义对应JSON Schema type: str object properties: Dict[str, Dict[str, Any]] required: Optional[List[str]] None class Tool(BaseModel): 一个工具的定义 name: str description: str inputSchema: ToolInputSchema class ListToolsResult(BaseModel): 响应 list_tools 方法的结果 tools: List[Tool] class CallToolArgument(BaseModel): 调用工具时传入的参数 name: str arguments: Dict[str, Any] # 这个字典必须符合对应工具的 inputSchema class CallToolResult(BaseModel): 调用工具返回的结果内容 content: List[Dict[str, Any]] # 通常至少包含一个 {type: text, text: 结果字符串} # -------------------- MCP 请求/响应包装 -------------------- class MCPRequest(JSONRPCRequest): MCP专用的请求继承自JSONRPCRequest pass class MCPInitializeResult(BaseModel): 初始化请求的响应结果 protocolVersion: str 2024-11-05 capabilities: Dict[str, Any] {} # 声明Server支持的能力如 tools, resources等 serverInfo: Optional[Dict[str, str]] None class MCPCallToolRequest(MCPRequest): 调用工具的请求method 为 tools/call method: str tools/call params: CallToolArgument class MCPListToolsRequest(MCPRequest): 列出所有工具的请求method 为 tools/list method: str tools/list这段代码定义了通信的“语言规则”。Tool模型描述了一个工具的样子CallToolArgument是AI调用时传来的数据格式CallToolResult是我们返回数据的格式。pydantic会在实例化时自动验证数据如果AI传来的参数不符合inputSchema在解析CallToolArgument时就会抛出验证错误我们可以据此返回友好的错误信息而不是让Server崩溃。3.3 实现MCP Server主逻辑server.py现在我们实现Server的主体。它需要做几件事1. 持续监听标准输入stdin2. 解析收到的JSON-RPC请求3. 根据method路由到对应的处理函数4. 执行处理逻辑5. 将结果写成JSON-RPC响应输出到标准输出stdout。# server.py import sys import json import traceback from typing import Dict, Any from protocol import ( MCPRequest, MCPInitializeResult, MCPCallToolRequest, MCPListToolsRequest, ListToolsResult, Tool, ToolInputSchema, CallToolResult, JSONRPCResponse ) class WeatherMCPServer: def __init__(self): # 定义我们对外提供的工具 self.tools [ Tool( nameget_weather, description获取指定城市的当前天气信息。, inputSchemaToolInputSchema( properties{ city: { type: string, description: 城市名称例如北京、Shanghai } }, required[city] ) ) ] # 请求路由映射 self.method_handlers { initialize: self._handle_initialize, tools/list: self._handle_list_tools, tools/call: self._handle_call_tool, # 可以继续添加其他MCP方法如 resources/list, prompts/list 等 } def _handle_initialize(self, request: MCPRequest) - Dict[str, Any]: 处理初始化请求返回协议版本和支持的能力 result MCPInitializeResult( capabilities{ tools: {} # 声明我们支持tools功能 }, serverInfo{name: Simple Weather MCP Server, version: 0.1.0} ) return result.model_dump() def _handle_list_tools(self, request: MCPListToolsRequest) - Dict[str, Any]: 处理列出工具的请求 result ListToolsResult(toolsself.tools) return result.model_dump() def _handle_call_tool(self, request: MCPCallToolRequest) - Dict[str, Any]: 处理调用工具的请求这里是业务逻辑核心 tool_name request.params.name arguments request.params.arguments if tool_name get_weather: city arguments.get(city, 未知城市) # 这里是模拟的业务逻辑真实场景可以调用天气API weather_info f{city}的模拟天气晴25°C微风。 # 构造符合MCP协议的返回内容 result CallToolResult( content[{type: text, text: weather_info}] ) return result.model_dump() else: # 如果收到未定义的工具名返回错误 raise ValueError(f未知的工具: {tool_name}) def _send_response(self, response: JSONRPCResponse): 将响应对象序列化为JSON写入stdout并立即刷新缓冲区 json_str response.model_dump_json(exclude_noneTrue) sys.stdout.write(json_str \n) sys.stdout.flush() def run(self): 主循环从stdin读取请求处理并响应到stdout print(MCP Weather Server 已启动等待请求..., filesys.stderr) for line in sys.stdin: line line.strip() if not line: continue try: # 1. 解析原始请求 raw_request json.loads(line) request_id raw_request.get(id) method raw_request.get(method) # 2. 根据method路由到处理函数 handler self.method_handlers.get(method) if not handler: # 对于不支持的方法返回方法未找到错误 error_response JSONRPCResponse( idrequest_id, error{code: -32601, message: fMethod not found: {method}} ) self._send_response(error_response) continue # 3. 将原始数据解析为对应的Pydantic请求模型 # 这里根据method做简单的分发解析实际可更精细 if method tools/call: parsed_request MCPCallToolRequest(**raw_request) elif method tools/list: parsed_request MCPListToolsRequest(**raw_request) else: parsed_request MCPRequest(**raw_request) # 4. 调用处理函数执行业务逻辑 result_data handler(parsed_request) # 5. 构造成功响应 response JSONRPCResponse(idrequest_id, resultresult_data) except json.JSONDecodeError: # 请求不是合法的JSON response JSONRPCResponse( idNone, error{code: -32700, message: Parse error} ) except Exception as e: # 处理过程中发生其他错误 error_msg traceback.format_exc() print(f处理请求时出错: {error_msg}, filesys.stderr) response JSONRPCResponse( idrequest_id, error{code: -32603, message: fInternal error: {str(e)}} ) # 6. 发送响应 self._send_response(response) if __name__ __main__: server WeatherMCPServer() server.run()这个Server的核心是run方法里的循环。它从sys.stdin逐行读取AI客户端写入的数据每行都是一个JSON-RPC请求。解析后根据method字段找到对应的处理函数_handle_xxx。处理函数返回结果后被封装成JSONRPCResponse再写回sys.stdoutAI客户端从这边读取。这就是基于stdio的进程间通信IPC简单而高效。 在_handle_call_tool中我们实现了具体的工具逻辑。这里只是返回模拟数据但你可以在这里替换成任何真实的代码调用第三方API、查询数据库、执行系统命令、操作本地文件等等。**这就是“把任意工具变成AI能力”的魔法发生地。**3.4 运行与测试手动模拟AI客户端我们的Server已经可以工作了。但怎么测试呢我们可以手动模拟一个AI客户端通过命令行与之交互。启动Server在终端运行python server.py。你会看到启动信息然后程序会挂起等待标准输入。模拟初始化请求打开另一个终端或者用echo命令向Server进程发送数据。我们需要按照MCP协议顺序发送请求。首先发送initialize请求。# 假设server.py的进程ID是12345我们可以用管道测试这里用文件描述符重定向更复杂我们用一个Python测试脚本更直观为了测试方便我们直接写一个简单的Python测试客户端test_client.py# test_client.py import subprocess import json import time # 启动Server进程并建立管道 proc subprocess.Popen( [python, server.py], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue ) def send_request(req): 发送一个请求并打印返回的响应 req_str json.dumps(req) \n print(f发送: {req_str.strip()}) proc.stdin.write(req_str) proc.stdin.flush() # 读取一行响应 response_line proc.stdout.readline() print(f收到: {response_line.strip()}) return json.loads(response_line) # 1. 初始化 init_req { jsonrpc: 2.0, id: 1, method: initialize, params: {} } send_request(init_req) # 2. 列出工具 list_tools_req { jsonrpc: 2.0, id: 2, method: tools/list, params: {} } send_request(list_tools_req) # 3. 调用工具 get_weather call_tool_req { jsonrpc: 2.0, id: 3, method: tools/call, params: { name: get_weather, arguments: { city: 上海 } } } send_request(call_tool_req) # 关闭进程 proc.terminate()运行python test_client.py你应该能看到类似以下的输出发送: {jsonrpc: 2.0, id: 1, method: initialize, params: {}} 收到: {jsonrpc: 2.0, id: 1, result: {protocolVersion: 2024-11-05, capabilities: {tools: {}}, serverInfo: {name: Simple Weather MCP Server, version: 0.1.0}}} 发送: {jsonrpc: 2.0, id: 2, method: tools/list, params: {}} 收到: {jsonrpc: 2.0, id: 2, result: {tools: [{name: get_weather, description: 获取指定城市的当前天气信息。, inputSchema: {type: object, properties: {city: {type: string, description: 城市名称例如北京、Shanghai}}, required: [city]}}]}} 发送: {jsonrpc: 2.0, id: 3, method: tools/call, params: {name: get_weather, arguments: {city: 上海}}} 收到: {jsonrpc: 2.0, id: 3, result: {content: [{type: text, text: 上海的模拟天气晴25°C微风。}]}}恭喜你已经成功完成了一次完整的MCP协议交互。Server正确声明了工具并处理了工具调用。4. 接入真实AI客户端以Claude Desktop为例手动测试通过了但我们的终极目标是让真正的AI模型来调用它。这里以Anthropic推出的Claude Desktop应用为例它内置了对MCP Server的支持配置起来非常方便。4.1 配置Claude Desktop加载自定义MCP ServerClaude Desktop允许通过配置文件添加自定义的MCP Server。配置文件通常位于macOS:~/Library/Application Support/Claude/claude_desktop_config.jsonWindows:%APPDATA%\Claude\claude_desktop_config.json我们需要在这个JSON配置文件中添加一个mcpServers字段来配置我们的Server。// claude_desktop_config.json { mcpServers: { weather: { command: /path/to/your/venv/bin/python, args: [ /full/path/to/your/mcp-weather-server/server.py ] } } }关键配置解释weather这是你给这个Server起的名字会在Claude内部显示。command启动Server的解释器或可执行文件的绝对路径。这里必须使用虚拟环境中的Python以确保依赖可用。args传递给命令的参数列表第一个参数就是我们Server脚本的绝对路径。踩坑点1路径问题这是新手配置时最容易出错的地方。command和args里的路径必须使用绝对路径不能使用~或相对路径。因为Claude Desktop在启动时其工作目录是不确定的。在macOS上你可以打开终端进入虚拟环境然后用which python获取解释器路径用pwd获取项目目录再拼接出server.py的绝对路径。4.2 验证与使用保存配置文件。完全重启Claude Desktop应用不是关闭窗口而是从任务栏/程序坞彻底退出再重新打开。这是必须的配置只在启动时加载。重启后当你新建一个对话时Claude应该已经加载了我们的Weather Server。你可以尝试提问“请使用get_weather工具查询一下北京的天气。”Claude会识别出可用的工具并在回复中可能以一个小工具图标或“使用工具”的按钮形式呈现。点击执行或等待它自动调用你就能看到它返回了我们Server提供的模拟天气信息。这个过程意味着你刚刚扩展了Claude这个AI模型的能力边界。它原本不知道如何“获取天气”但现在通过你写的MCP Server它获得了这个能力。而且这个能力是结构化、可可靠调用的而不是通过模糊的提示词去“猜”。5. 进阶将复杂本地工具接入MCP模拟天气工具只是个开始。MCP的强大之处在于能将任何本地能力暴露给AI。让我们设想一个更复杂的场景一个本地的项目文件搜索工具。它接收一个关键词在指定的项目目录下递归搜索所有文件返回包含该关键词的文件路径和匹配行的预览。5.1 设计工具接口与实现首先在server.py的__init__中增加这个新工具的定义# 在 self.tools 列表中添加新工具 self.tools [ Tool(...), # 原有的天气工具 Tool( namesearch_project_files, description在指定的项目根目录中递归搜索所有文件内容查找包含特定关键词的行。, inputSchemaToolInputSchema( properties{ project_root: { type: string, description: 项目根目录的绝对路径。 }, keyword: { type: string, description: 需要搜索的关键词。 }, file_extensions: { type: array, items: {type: string}, description: 可选指定要搜索的文件扩展名列表如 [.py, .md, .txt]。默认为搜索所有文件。 } }, required[project_root, keyword] ) ) ]然后在_handle_call_tool方法中添加对应的处理逻辑# 在 _handle_call_tool 的 if-elif 链中添加 elif tool_name search_project_files: project_root arguments[project_root] keyword arguments[keyword] file_extensions arguments.get(file_extensions) import os matches [] # 安全警告在实际生产环境中必须对project_root进行严格的路径安全校验防止目录遍历攻击。 # 这里为演示简化处理。 for root, dirs, files in os.walk(project_root): for file in files: if file_extensions: if not any(file.endswith(ext) for ext in file_extensions): continue file_path os.path.join(root, file) try: with open(file_path, r, encodingutf-8, errorsignore) as f: for line_num, line in enumerate(f, 1): if keyword in line: # 构造一个匹配结果包含文件路径、行号和内容预览 preview line.strip()[:100] # 预览前100个字符 matches.append({ file: file_path, line: line_num, preview: preview }) # 限制返回结果数量避免响应过大 if len(matches) 20: break if len(matches) 20: break except (IOError, OSError, UnicodeDecodeError): # 跳过无法读取的文件如二进制文件、无权限文件 continue if matches: result_text f在目录 {project_root} 中搜索关键词 {keyword} 找到 {len(matches)} 处匹配\n\n for m in matches: result_text f- **文件**: {m[file]} (第{m[line]}行)\n 预览: {m[preview]}\n\n else: result_text f在目录 {project_root} 中未找到包含关键词 {keyword} 的内容。 result CallToolResult( content[{type: text, text: result_text}] ) return result.model_dump()5.2 安全考量与错误处理将本地工具暴露给AI安全是第一要务。上面的示例代码省略了关键的安全检查在实际应用中必须补上路径遍历攻击防护必须验证project_root参数是否在允许的目录范围内例如限制在用户家目录下的某个子目录内。可以使用os.path.abspath解析绝对路径并与一个预设的安全基础路径进行比较。SAFE_BASE_PATH /Users/yourname/Projects # 允许搜索的根目录 requested_path os.path.abspath(project_root) if not requested_path.startswith(SAFE_BASE_PATH): return CallToolResult(content[{type: text, text: 错误请求的目录不在允许的范围内。}]).model_dump()资源消耗限制文件搜索可能耗时耗内存。需要添加超时机制限制遍历的深度或文件数量防止AI意外触发一个对全盘文件的搜索。敏感信息过滤返回的结果中可能意外包含密码、密钥等敏感信息。在返回前应对结果内容进行简单的正则匹配过滤。踩坑点2AI的“不可预测性”AI调用工具的参数可能和你预想的不一样。它可能传一个不存在的路径或者一个空字符串。你的Server必须足够健壮对输入参数进行严格的验证和清理并返回清晰、对人类和AI都友好的错误信息而不是抛出未处理的异常导致Server崩溃。例如可以这样处理try: # 业务逻辑 except Exception as e: error_result CallToolResult( content[{type: text, text: f执行工具时出错{str(e)}。请检查参数是否正确。}] ) return error_result.model_dump()5.3 效果演示与思考配置好这个增强版的Server并重启Claude Desktop后你可以对AI说“请帮我搜索/Users/me/Code/my_python_project目录下所有.py文件中提到pydantic的地方。”AI会理解你的意图调用search_project_files工具并传入project_root、keyword和file_extensions参数。很快它就能将搜索结果整理成清晰的格式返回给你。这一步的质变在于你不再需要手动grep也不需要教AI复杂的shell命令。你通过一个定义良好的接口将本地强大的文件搜索能力“赋予”了AI。AI成为了你的智能操作界面而复杂的、易错的命令行操作被封装在了背后可靠的Server里。6. 协议级实践的深层价值与未来展望通过这个从零构建的过程我们触及了MCP协议级实践的几个核心价值这远不止是“让AI多了一个功能”那么简单。6.1 解耦AI模型与工具实现的分离这是最重要的架构优势。我们的工具逻辑server.py和AI模型Claude是完全独立的进程。这意味着工具可以独立升级和部署比如我们把文件搜索算法从简单字符串匹配升级为正则表达式甚至语义搜索只需要更新Server无需改动AI客户端或模型。可以用任何语言编写工具我们用Python写你也可以用Go写一个高性能的用Rust写一个内存安全的。只要遵守MCP协议这个“契约”AI就能调用。权限和安全边界清晰Server进程以特定的系统权限运行可以严格限制其能访问的资源如网络、文件系统。AI模型本身在一个沙盒里它通过协议来请求操作由Server这个“可信代理”来执行实现了权限的最小化原则。6.2 组合性工具即乐高积木一个MCP Server可以暴露多个工具tools、资源resources和提示词模板prompts。AI客户端可以同时连接多个这样的Server。想象一下一个Server提供数据库查询工具。一个Server提供代码仓库管理Git工具。一个Server提供内部API调用工具。一个Server提供项目文档资源。AI模型可以像搭积木一样根据用户的需求自由组合调用这些来自不同Server的能力完成一个复杂的多步骤任务。例如“基于最新的用户反馈来自数据库为项目Issue #123来自Git写一段回复并参考我们的API文档来自资源。” 这为构建复杂的AI Agent工作流奠定了坚实的基础。6.3 生态与未来标准化的价值MCP由Anthropic提出但正在成为一个开放标准。当越来越多的工具都通过MCP协议暴露接口时就会形成一个生态。任何支持MCP的AI客户端未来可能不止Claude Desktop都能立即获得海量的工具能力无需为每个工具单独开发插件。对于开发者而言为你写的任何命令行工具、本地服务或内部系统包装一个MCP Server就相当于为它接入了整个AI生态。这比针对每个AI平台OpenAI GPTs, Copilot, Cursor等单独开发插件要高效和可持续得多。从我个人的实践来看MCP协议级接入的初期学习曲线比直接用现成SDK要陡峭但它带来的理解深度和灵活性是无可替代的。你不再是一个“插件使用者”而是成为了“能力定义者”。你清楚地知道数据在协议层如何流动如何设计安全可靠的接口如何调试通信问题。这种掌控感是在AI时代构建可靠、复杂应用所必需的底层能力。当你下次再遇到一个希望AI能帮你操作但又没有现成桥接工具的场景时不妨想一想我能不能花一两个小时为它写一个简单的MCP Server这很可能就是最高效、最一劳永逸的解决方案。