1. 从协议视角重新理解MCP Server如果你已经跟着基础教程跑通了几个现成的MCP Server比如文件读写或者简单的天气查询可能会觉得MCPModel Context Protocol不过就是定义几个工具Tools和资源Resources然后通过一个标准化的JSON-RPC接口暴露给AI助手如Claude Desktop。这个理解没错但停留在应用层。当你开始思考“我能不能自己写一个Server”或者“为什么我的Server连不上Client”时就必须穿透这层抽象直接面对它的通信本质——一个基于标准输入输出stdio的JSON-RPC 2.0协议。为什么是stdio这是MCP设计上非常聪明且务实的一点。它避免了复杂的网络端口管理、认证和防火墙问题。Server和Client通常是AI助手的前端进程就像两个命令行程序通过管道连接起来。Server从stdin读取请求处理后将响应写入stdout错误信息则写入stderr。这种模式使得集成变得极其轻量任何能跑命令行程序的环境都能运行MCP无论是本地开发机、容器还是远程服务器。但这也带来了调试的挑战。你没法像调试一个HTTP服务那样直接在浏览器里访问localhost:8080看看是否存活。当你的Server启动失败、消息解析出错或者返回了不符合预期的数据时黑盒般的stdio管道让人无从下手。这时抓包或者说抓“流”就成了透视协议内部状态的“X光机”。我们不是抓网络包而是拦截并解析在父子进程间流动的原始JSON-RPC消息。这能让你清晰地看到Client到底发送了什么method和params你的Server又回传了怎样的result或error协议约定的字段一个都不能少格式稍有偏差通信就会静默失败。所以手写一个MCP Server绝不仅仅是实现几个函数那么简单。它是一个从“协议使用者”到“协议实现者”的思维转变。你需要像协议的制定者一样思考精确地构造每一条消息严谨地处理每一个状态。而抓包工具就是你实现这个转变过程中最可靠的调试伙伴和验证器。接下来我们就从零开始构建一个能实际运行的Server并用抓包工具亲眼见证协议的每一个字节。2. 手写一个极简但完整的MCP Server我们选择用Python来实现因为它语法简洁标准库强大非常适合演示协议细节。这个Server将实现一个最简单的功能一个名为get_server_time的工具调用后返回服务器的当前时间戳和格式化时间。麻雀虽小五脏俱全它会完整走通MCP Server必须实现的几个核心生命周期。2.1 项目初始化与依赖分析首先创建一个新的项目目录。我们刻意不使用官方的mcpSDK比如mcp[cli]目的是为了彻底理解底层协议。你的项目只需要Python标准库。mkdir my_mcp_server cd my_mcp_server touch server.py在动手写代码前我们必须明确MCP Server在启动和运行过程中与Client交换的关键消息序列。这个过程是严格定义好的初始化Initialization: Client启动Server进程后发送的第一条消息必须是initialize请求。Server必须回复initialize_result其中包含其声明的capabilities支持哪些MCP特性如工具、资源等。就绪通知Ready Notification: Server在成功初始化后必须主动向Client发送一条notifications/initialized通知告知Client自己已准备就绪。工具列表Listing Tools: Client在收到就绪通知后会发送tools/list请求。Server必须回复tools/list_result列出所有可用的工具及其输入参数模式JSON Schema。工具调用Tool Call: 用户通过AI助手请求某个功能时Client会发送tools/call请求。Server执行对应逻辑后返回tools/call_result。关闭Shutdown: 当Client需要退出时会发送notifications/shutdown通知。Server收到后应进行清理工作然后等待Client发送exit通知后终止进程。我们的代码将围绕处理这些特定的JSON-RPC消息展开。2.2 核心消息处理循环的实现打开server.py我们从最核心的消息循环开始写起。这个循环负责从标准输入持续读取数据、解析JSON-RPC消息、路由到对应的处理函数并将结果写回标准输出。#!/usr/bin/env python3 import sys import json import time import traceback from datetime import datetime import logging # 设置日志方便调试。注意协议错误应使用stderr。 logging.basicConfig(levellogging.DEBUG, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class SimpleMCPServer: def __init__(self): self.initialized False self.shutdown_requested False def write_message(self, message): 将JSON-RPC消息写入stdout并遵循协议要求添加Content-Length头。 json_str json.dumps(message, ensure_asciiFalse) content fContent-Length: {len(json_str)}\r\n\r\n{json_str} sys.stdout.write(content) sys.stdout.flush() logger.debug(fSent: {json_str[:200]}...) # 日志只记录前200字符 def run(self): 主运行循环从stdin读取消息。 buffer while not self.shutdown_requested: # 读取标准输入 line sys.stdin.readline() if not line: # EOF通常意味着父进程Client关闭了管道 break buffer line # 检查是否收到了完整的HTTP头样式的分隔符\r\n\r\n if \r\n\r\n in buffer: headers_part, body_part buffer.split(\r\n\r\n, 1) headers {} for header_line in headers_part.split(\r\n): if : in header_line: key, value header_line.split(: , 1) headers[key.lower()] value content_length int(headers.get(content-length, 0)) # 检查body部分是否已经接收了足够长度的数据 if len(body_part) content_length: # 提取一条完整消息 message_body body_part[:content_length] # 剩余数据放回buffer用于下一条消息 buffer body_part[content_length:] try: message json.loads(message_body) self.handle_message(message) except json.JSONDecodeError as e: logger.error(fFailed to decode JSON: {e}\nBody: {message_body}) # 发送一个JSON-RPC错误响应 error_response { jsonrpc: 2.0, id: None, # 无法得知id error: { code: -32700, message: Parse error } } self.write_message(error_response) # 如果body还没收全继续循环读取 def handle_message(self, message): 根据JSON-RPC消息的method字段进行路由分发。 method message.get(method) msg_id message.get(id) # 请求才有id通知没有 logger.info(fReceived method: {method}, id: {msg_id}) if method initialize: self.handle_initialize(message, msg_id) elif method tools/list: self.handle_tools_list(message, msg_id) elif method tools/call: self.handle_tools_call(message, msg_id) elif method notifications/shutdown: self.handle_shutdown(message) elif method exit: self.shutdown_requested True else: logger.warning(fUnknown method: {method}) # 对于未知的请求有id需要回复错误 if msg_id is not None: error_response { jsonrpc: 2.0, id: msg_id, error: { code: -32601, message: fMethod not found: {method} } } self.write_message(error_response) # 具体的消息处理器将在下面实现...关键点解析消息边界协议MCP没有使用简单的换行符分隔JSON而是采用了类似HTTP的Content-Length头。这是为了支持消息体内包含任意内容包括换行符。我们的循环必须正确解析这个头部并读取指定长度的Body。请求与通知JSON-RPC 2.0区分请求Request有id字段需要回复和通知Notification无id字段无需回复。initialize、tools/list、tools/call是请求notifications/shutdown和exit是通知。错误处理对无法解析的JSONParse error和未知的方法Method not found必须按照JSON-RPC规范返回错误对象这是实现一个健壮Server的基础。2.3 实现核心消息处理器现在我们来填充上面用到的几个核心消息处理器。def handle_initialize(self, message, msg_id): 处理initialize请求返回Server的能力声明。 params message.get(params, {}) client_info params.get(clientInfo, {}) logger.info(fClient initialized: {client_info}) # 构建响应结果 result { protocolVersion: 2024-11-05, # 使用当前稳定的协议版本 capabilities: { tools: {} # 我们声明支持Tools功能 # 未来还可以声明resources: {}, prompts: {} 等 }, serverInfo: { name: Simple Time Server, version: 0.1.0 } } response { jsonrpc: 2.0, id: msg_id, result: result } self.write_message(response) # 发送initialized通知 initialized_notification { jsonrpc: 2.0, method: notifications/initialized, params: {} # 此通知无参数 } self.write_message(initialized_notification) self.initialized True logger.info(Server initialized and notified client.) def handle_tools_list(self, message, msg_id): 处理tools/list请求返回所有可用工具的定义。 if not self.initialized: # 理论上Client应在收到initialized通知后才发此请求。 # 但为健壮性考虑这里也检查状态。 error_response { jsonrpc: 2.0, id: msg_id, error: { code: -32002, message: Server not initialized } } self.write_message(error_response) return # 定义我们的唯一工具get_server_time tools [ { name: get_server_time, description: 获取服务器的当前时间返回时间戳和格式化字符串。, inputSchema: { type: object, properties: { format: { type: string, description: 可选的时间格式字符串如%Y-%m-%d %H:%M:%S。默认为ISO格式。, default: iso } } } } ] result {tools: tools} response { jsonrpc: 2.0, id: msg_id, result: result } self.write_message(response) def handle_tools_call(self, message, msg_id): 处理tools/call请求执行具体的工具并返回结果。 params message.get(params, {}) tool_name params.get(name) arguments params.get(arguments, {}) if tool_name get_server_time: # 执行工具逻辑 fmt arguments.get(format, iso) now time.time() if fmt iso: formatted datetime.fromtimestamp(now).isoformat() else: try: formatted datetime.fromtimestamp(now).strftime(fmt) except Exception as e: formatted fFormat error: {e} result_content { content: [ { type: text, text: fServer Time (UTC): {formatted}\nTimestamp: {now} } ] } response { jsonrpc: 2.0, id: msg_id, result: result_content } else: # 工具不存在 response { jsonrpc: 2.0, id: msg_id, error: { code: -32602, message: fTool not found: {tool_name} } } self.write_message(response) def handle_shutdown(self, message): 处理shutdown通知。 logger.info(Received shutdown notification.) # 这里可以执行一些清理操作如关闭数据库连接等。 # 无需回复因为是通知。 # Client之后会发送exit通知。关键点解析能力声明Capabilities在initialize_result中capabilities字段是Server的“功能菜单”。我们只声明了tools: {}表示支持工具功能。即使是一个空对象{}也必须有这个键这是协议约定。如果你想支持资源Resources就需要在这里声明resources: {}。工具定义Tool Definitiontools/list_result返回的列表里每个工具都必须有name、description和inputSchema。inputSchema是一个JSON Schema对象用于描述输入参数的结构。这允许Client如Claude在调用前就理解参数格式甚至生成调用界面。工具调用结果Tool Call Resulttools/call_result的result字段结构是固定的必须包含一个content数组。数组中的每个元素目前通常是{type: text, text: ...}。这是MCP协议规定的AI助手可呈现的内容格式。严格的顺序initialized通知必须在initialize_result之后立即发送。很多自研Server的第一个坑就是漏发或错发这个通知导致Client一直等待超时后连接失败。2.4 启动脚本与权限设置最后添加启动代码并确保文件有可执行权限。if __name__ __main__: server SimpleMCPServer() try: server.run() except Exception as e: logger.error(fServer crashed: {e}) traceback.print_exc() sys.exit(1)给脚本加上执行权限chmod x server.py。现在一个纯手工打造的、符合MCP协议的Server就完成了。你可以尝试用最原始的方式测试它echo ... | python3 server.py但这很麻烦。更有效的方式是结合Claude Desktop和抓包工具来验证这正是下一章的内容。注意在生产环境中强烈建议使用官方SDK如modelcontextprotocol/sdkfor JavaScript/TypeScript,mcpfor Python它们帮你处理了所有这些底层协议细节、错误处理和生命周期管理。但通过这次手写你获得了对协议最深刻的理解这是解决一切诡异问题的终极武器。3. 使用抓包工具透视MCP协议通信现在我们有了一个自研的Server但它真的能工作吗当它和Claude Desktop连接时到底在“说”些什么光看日志不够直观我们需要直接窥视stdin/stdout上流动的原始数据。在Windows上我们可以使用强大的Fiddler Everywhere或Wireshark配置稍复杂而在macOS/Linux上script命令或socat是更轻量的选择。这里我将展示一种跨平台的、基于Python的“中间人”抓包方法它原理清晰且能直接输出格式化JSON非常适合学习和调试。3.1 构建一个简单的Stdio流量转发与记录器我们写一个简单的Python脚本作为“中间人”Man-in-the-Middle。它位于Claude DesktopClient和我们的SimpleMCPServer之间双向转发数据同时将经过的每一条消息打印到控制台或文件。#!/usr/bin/env python3 # mitm_logger.py import sys import json import subprocess import threading import time from queue import Queue, Empty def log_message(direction, data): 格式化并记录一条消息。 try: # 尝试解析为JSON并美化输出 message json.loads(data) pretty_json json.dumps(message, indent2, ensure_asciiFalse) print(f\n{*60}) print(f[{direction}] JSON-RPC Message:) print(pretty_json) print(f{*60}\n) except json.JSONDecodeError: # 如果不是完整的JSON可能是分片或者头部信息直接打印原始数据 print(f\n[{direction}] Raw Data (non-JSON or partial):) print(repr(data[:500])) # 只打印前500字符 def read_and_forward(stream_from, stream_to, queue, direction): 从一个流读取数据转发到另一个流同时将数据放入队列供记录。 try: while True: # 读取一块数据这里简化处理实际应像Server一样解析Content-Length # 为了演示我们每次读4KB chunk stream_from.read(4096) if not chunk: break # EOF # 转发 stream_to.write(chunk) stream_to.flush() # 记录 queue.put((direction, chunk)) except Exception as e: print(fError in {direction} reader: {e}, filesys.stderr) finally: queue.put(None) # 发送结束信号 def main(): if len(sys.argv) 2: print(fUsage: {sys.argv[0]} command [args...], filesys.stderr) sys.exit(1) # 启动子进程我们的MCP Server server_proc subprocess.Popen( sys.argv[1:], # 从命令行参数获取要启动的Server命令 stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, # 将Server的stderr重定向到管道我们单独处理 textFalse, # 使用二进制模式避免编码问题 bufsize0 ) # 队列用于在主线程中记录消息避免多线程打印混乱 log_queue Queue() # 启动两个线程分别处理Client-Server和Server-Client的数据流 # 注意这里我们模拟Client即我们手动输入。实际应与Claude Desktop集成。 # 为了演示我们创建一个虚拟的“Client”线程来发送初始化请求。 def dummy_client(): # 构建一个标准的initialize请求 init_request { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2024-11-05, clientInfo: { name: MCP-Debug-Client, version: 1.0.0 } } } request_json json.dumps(init_request) message fContent-Length: {len(request_json)}\r\n\r\n{request_json} server_proc.stdin.write(message.encode()) server_proc.stdin.flush() log_queue.put((C-S, message.encode())) time.sleep(0.5) # 等待Server回复 # 接着发送tools/list请求 list_request { jsonrpc: 2.0, id: 2, method: tools/list, params: {} } request_json json.dumps(list_request) message fContent-Length: {len(request_json)}\r\n\r\n{request_json} server_proc.stdin.write(message.encode()) server_proc.stdin.flush() log_queue.put((C-S, message.encode())) client_thread threading.Thread(targetdummy_client) client_thread.start() # 启动从Server stdout到我们stdout的转发线程S-C stdout_thread threading.Thread( targetread_and_forward, args(server_proc.stdout, sys.stdout.buffer, log_queue, S-C) ) stdout_thread.daemon True stdout_thread.start() # 单独处理Server的stderr直接打印到我们的stderr def read_stderr(): while True: chunk server_proc.stderr.read(4096) if not chunk: break sys.stderr.buffer.write(b[Server STDERR] chunk) sys.stderr.flush() stderr_thread threading.Thread(targetread_stderr) stderr_thread.daemon True stderr_thread.start() # 主线程从队列中取出数据并记录 end_count 0 try: while end_count 2: # 等待两个转发线程结束 try: item log_queue.get(timeout0.1) if item is None: end_count 1 else: direction, data item log_message(direction, data) except Empty: continue except KeyboardInterrupt: print(\nInterrupted by user.) finally: server_proc.terminate() server_proc.wait() if __name__ __main__: main()这个脚本是一个简化版的调试工具。它启动了我们的Server并模拟了一个最简单的Client发送两条请求。运行它python3 mitm_logger.py python3 server.py。你将在控制台看到类似这样的输出 [C-S] JSON-RPC Message: { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2024-11-05, clientInfo: { name: MCP-Debug-Client, version: 1.0.0 } } } [S-C] JSON-RPC Message: { jsonrpc: 2.0, id: 1, result: { protocolVersion: 2024-11-05, capabilities: { tools: {} }, serverInfo: { name: Simple Time Server, version: 0.1.0 } } } [S-C] JSON-RPC Message: { jsonrpc: 2.0, method: notifications/initialized, params: {} } 太棒了你亲眼看到了协议的三次握手Client的initialize请求Server的initialize_result响应以及紧随其后的notifications/initialized通知。这就是协议规范在字节层面的具体体现。3.2 与真实Claude Desktop集成并抓包上面的调试器是自娱自乐。要让我们的Server被真正的Claude Desktop调用需要将其配置为Claude的MCP Server。这里以macOS上的Claude Desktop为例其他平台原理类似找到Claude Desktop的配置文件通常位于~/Library/Application Support/Claude/claude_desktop_config.json。编辑配置文件在mcpServers部分添加我们的Server。{ mcpServers: { simple-time-server: { command: python3, args: [/ABSOLUTE/PATH/TO/YOUR/my_mcp_server/server.py] } } }重启Claude Desktop。现在真正的挑战来了。当Claude Desktop启动时它会自动运行我们配置的Server命令并通过stdio与之通信。我们如何抓取这个“进程间”的流量直接修改Claude的配置指向我们的mitm_logger.py是一个办法但更通用的方法是使用系统级的调试工具。在macOS/Linux上可以使用script命令或socatscript命令可以记录一个终端会话的所有输入输出。你可以写一个包装脚本先启动script记录再在script启动的shell中运行Claude Desktop。但这会记录所有终端活动需要过滤。socat更精确它可以创建一对虚拟的PTY伪终端让Claude和我们的Server通过这对PTY通信而socat自己则在中间双向拷贝数据并打印。命令类似socat -x -v PTY,link/tmp/mcp-client,raw,echo0 PTY,link/tmp/mcp-server,raw,echo0。然后将Claude配置中的command改为socat创建的PTY路径之一将我们的Server连接到另一个PTY。-v参数会将所有传输的十六进制和ASCII数据打印到stderr。在Windows上可以使用Fiddler Everywhere或Wireshark对于stdio流这些网络抓包工具默认抓不到。但MCP也支持stdio和sseServer-Sent Events等传输方式。如果是sse基于HTTP那么Fiddler和Wireshark就能直接抓包。不过Claude Desktop目前主要使用stdio与本地Server通信。一个可行的办法是编写一个本地的“代理Server”。这个代理作为一个HTTP/sse Server运行Claude配置连接这个代理。代理再将请求通过stdio转发给我们真正的Python Server同时记录所有流量。这需要更多代码但提供了最大的灵活性和可观察性。实操心得在开发初期最快速的验证方式其实就是我们写的mitm_logger.py。先确保你的Server能正确响应最基本的initialize和tools/list请求。一旦这两步通了集成到Claude Desktop的成功率就大大提升。集成失败时首先查看Claude Desktop的日志位置因平台而异通常会有“Server exited with code 1”或“Failed to parse message”之类的错误这些线索能帮你快速定位是启动问题还是协议格式问题。4. 协议拆解从抓包结果分析关键帧与常见陷阱通过抓包我们获得了协议通信的第一手资料。现在让我们像一个协议分析师一样仔细审视这些“关键帧”并总结出开发中最容易踩中的陷阱。4.1 关键帧深度解析让我们基于抓包结果逐一拆解每个核心消息帧的必须字段和可选字段。1. 初始化请求帧 (Client - Server){ jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2024-11-05, clientInfo: { name: Claude Desktop, version: 1.5.0 }, // 可选客户端可能声明自己支持的能力但Server应忽略 capabilities: {} } }必须字段jsonrpc,id,method,params。protocolVersion这是致命关键。必须与Server实现的版本一致。如果你手写的Server填错了版本比如用了旧的草案日期Claude Desktop可能会直接拒绝通信。目前稳定版本是2024-11-05。clientInfo仅供参考用于日志记录。2. 初始化响应帧 (Server - Client){ jsonrpc: 2.0, id: 1, // 必须与请求的id对应 result: { protocolVersion: 2024-11-05, // 必须与请求中的一致 capabilities: { tools: {}, resources: {}, // 可选如果支持资源则需声明 prompts: {} // 可选如果支持提示词则需声明 }, serverInfo: { name: My Server, version: 0.1.0 } } }capabilities对象即使某个功能你暂时没有实现比如resources也不能省略这个键。你必须返回一个对象其内部可以是空的{}但键必须存在。例如只支持工具就必须是capabilities: {tools: {}}。如果完全返回capabilities: {}Client会认为你什么都不支持。serverInfo可选但强烈建议提供有助于在Client界面识别Server。3. 工具列表响应帧 (Server - Client){ jsonrpc: 2.0, id: 2, result: { tools: [ { name: get_weather, description: 获取指定城市的天气。, inputSchema: { type: object, properties: { city: { type: string, description: 城市名称如北京。 }, unit: { type: string, enum: [celsius, fahrenheit], description: 温度单位。, default: celsius } }, required: [city] // 明确哪些参数是必需的 } } ] } }inputSchema的重要性这不仅仅是文档。AI助手如Claude会利用这个JSON Schema来理解如何调用你的工具。description字段要清晰required数组要准确。如果required里写了city但调用时没传Client端可能会直接报错而不会发请求到Server。enum和default善用这些字段可以极大提升用户体验。enum限定了输入范围default提供了默认值。4. 工具调用结果帧 (Server - Client){ jsonrpc: 2.0, id: 123, result: { content: [ { type: text, text: 当前温度是22度天气晴朗。, mimeType: text/plain // 可选 }, { type: image, data: base64EncodedString..., mimeType: image/png } ] } }content数组这是返回数据的核心。目前主要支持text和image类型。text是必须支持的。你可以返回多条content它们会被组合呈现给用户。错误响应如果工具执行出错应返回标准的JSON-RPC错误帧而不是在result里放错误信息。{ jsonrpc: 2.0, id: 123, error: { code: -32000, message: Failed to fetch weather data: Network error., data: {city: Beijing} // 可选提供错误上下文 } }4.2 开发中最常见的五个陷阱与排查清单结合手写Server和抓包分析的经验我总结了五个最容易导致集成失败的坑点并给出排查思路。陷阱一Content-Length头格式错误或计算不准现象Client日志报“Parse error”或“Invalid frame”Server可能收不到任何请求或收到乱码。根因没有严格按照Content-Length: 数字\r\n\r\n消息体的格式发送。数字必须是消息体字节数对于UTF-8字符串一个中文字符可能是3个字节。或者在读取时没有正确解析这个头部导致消息边界错乱。排查用抓包工具看原始字节流。检查\r\n\r\n是否完整检查Content-Length的值是否等于后面JSON字符串的实际字节长度len(json_str.encode(‘utf-8’))。陷阱二initialized通知漏发或顺序错误现象Client在发送initialize后一直等待超时后断开连接日志显示“Server did not send initialized notification”。根因Server在回复initialize_result后没有立即发送notifications/initialized方法通知。或者错误地将其作为initialize_result的一部分。排查抓包确认在initialize_result响应帧之后是否紧跟着一个method为notifications/initialized、且没有id字段的通知帧。陷阱三capabilities声明不完整或错误现象Server能初始化但Client里看不到任何工具或资源。根因initialize_result中的capabilities字段缺少对应的键。例如你实现了工具但返回的是capabilities: {}或capabilities: {resources: {}}。排查抓包查看initialize_result确认capabilities.tools是否存在即使为空对象{}。如果你实现了资源则capabilities.resources也必须存在。陷阱四工具调用结果格式不符合协议现象工具调用后AI助手没有显示结果或显示乱码。根因tools/call_result的result字段结构错误。最常见的是直接返回了字符串或自定义对象而不是{content: [{type: text, text: ...}]}结构。排查抓包查看tools/call_result帧确保result.content是一个数组且里面的对象有type和text字段。陷阱五JSON-RPCid不匹配或缺失现象某些请求没有回应或者Client报告“收到未知的响应”。根因对于请求有id的响应必须携带完全相同的id值。如果Server在处理请求时弄丢了id或者响应了一个与任何未完成请求都不匹配的idClient会无法处理该响应。排查在Server代码中确保将请求中的id原封不动地复制到响应对象中。对于通知如initialized则绝不能有id字段。当你遇到连接或调用问题时按照这份清单逐一核对抓包数据几乎能解决90%的协议层问题。剩下的10%可能就是环境配置或权限问题了。 手写MCP Server：从JSON-RPC协议到调试实战
1. 从协议视角重新理解MCP Server如果你已经跟着基础教程跑通了几个现成的MCP Server比如文件读写或者简单的天气查询可能会觉得MCPModel Context Protocol不过就是定义几个工具Tools和资源Resources然后通过一个标准化的JSON-RPC接口暴露给AI助手如Claude Desktop。这个理解没错但停留在应用层。当你开始思考“我能不能自己写一个Server”或者“为什么我的Server连不上Client”时就必须穿透这层抽象直接面对它的通信本质——一个基于标准输入输出stdio的JSON-RPC 2.0协议。为什么是stdio这是MCP设计上非常聪明且务实的一点。它避免了复杂的网络端口管理、认证和防火墙问题。Server和Client通常是AI助手的前端进程就像两个命令行程序通过管道连接起来。Server从stdin读取请求处理后将响应写入stdout错误信息则写入stderr。这种模式使得集成变得极其轻量任何能跑命令行程序的环境都能运行MCP无论是本地开发机、容器还是远程服务器。但这也带来了调试的挑战。你没法像调试一个HTTP服务那样直接在浏览器里访问localhost:8080看看是否存活。当你的Server启动失败、消息解析出错或者返回了不符合预期的数据时黑盒般的stdio管道让人无从下手。这时抓包或者说抓“流”就成了透视协议内部状态的“X光机”。我们不是抓网络包而是拦截并解析在父子进程间流动的原始JSON-RPC消息。这能让你清晰地看到Client到底发送了什么method和params你的Server又回传了怎样的result或error协议约定的字段一个都不能少格式稍有偏差通信就会静默失败。所以手写一个MCP Server绝不仅仅是实现几个函数那么简单。它是一个从“协议使用者”到“协议实现者”的思维转变。你需要像协议的制定者一样思考精确地构造每一条消息严谨地处理每一个状态。而抓包工具就是你实现这个转变过程中最可靠的调试伙伴和验证器。接下来我们就从零开始构建一个能实际运行的Server并用抓包工具亲眼见证协议的每一个字节。2. 手写一个极简但完整的MCP Server我们选择用Python来实现因为它语法简洁标准库强大非常适合演示协议细节。这个Server将实现一个最简单的功能一个名为get_server_time的工具调用后返回服务器的当前时间戳和格式化时间。麻雀虽小五脏俱全它会完整走通MCP Server必须实现的几个核心生命周期。2.1 项目初始化与依赖分析首先创建一个新的项目目录。我们刻意不使用官方的mcpSDK比如mcp[cli]目的是为了彻底理解底层协议。你的项目只需要Python标准库。mkdir my_mcp_server cd my_mcp_server touch server.py在动手写代码前我们必须明确MCP Server在启动和运行过程中与Client交换的关键消息序列。这个过程是严格定义好的初始化Initialization: Client启动Server进程后发送的第一条消息必须是initialize请求。Server必须回复initialize_result其中包含其声明的capabilities支持哪些MCP特性如工具、资源等。就绪通知Ready Notification: Server在成功初始化后必须主动向Client发送一条notifications/initialized通知告知Client自己已准备就绪。工具列表Listing Tools: Client在收到就绪通知后会发送tools/list请求。Server必须回复tools/list_result列出所有可用的工具及其输入参数模式JSON Schema。工具调用Tool Call: 用户通过AI助手请求某个功能时Client会发送tools/call请求。Server执行对应逻辑后返回tools/call_result。关闭Shutdown: 当Client需要退出时会发送notifications/shutdown通知。Server收到后应进行清理工作然后等待Client发送exit通知后终止进程。我们的代码将围绕处理这些特定的JSON-RPC消息展开。2.2 核心消息处理循环的实现打开server.py我们从最核心的消息循环开始写起。这个循环负责从标准输入持续读取数据、解析JSON-RPC消息、路由到对应的处理函数并将结果写回标准输出。#!/usr/bin/env python3 import sys import json import time import traceback from datetime import datetime import logging # 设置日志方便调试。注意协议错误应使用stderr。 logging.basicConfig(levellogging.DEBUG, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class SimpleMCPServer: def __init__(self): self.initialized False self.shutdown_requested False def write_message(self, message): 将JSON-RPC消息写入stdout并遵循协议要求添加Content-Length头。 json_str json.dumps(message, ensure_asciiFalse) content fContent-Length: {len(json_str)}\r\n\r\n{json_str} sys.stdout.write(content) sys.stdout.flush() logger.debug(fSent: {json_str[:200]}...) # 日志只记录前200字符 def run(self): 主运行循环从stdin读取消息。 buffer while not self.shutdown_requested: # 读取标准输入 line sys.stdin.readline() if not line: # EOF通常意味着父进程Client关闭了管道 break buffer line # 检查是否收到了完整的HTTP头样式的分隔符\r\n\r\n if \r\n\r\n in buffer: headers_part, body_part buffer.split(\r\n\r\n, 1) headers {} for header_line in headers_part.split(\r\n): if : in header_line: key, value header_line.split(: , 1) headers[key.lower()] value content_length int(headers.get(content-length, 0)) # 检查body部分是否已经接收了足够长度的数据 if len(body_part) content_length: # 提取一条完整消息 message_body body_part[:content_length] # 剩余数据放回buffer用于下一条消息 buffer body_part[content_length:] try: message json.loads(message_body) self.handle_message(message) except json.JSONDecodeError as e: logger.error(fFailed to decode JSON: {e}\nBody: {message_body}) # 发送一个JSON-RPC错误响应 error_response { jsonrpc: 2.0, id: None, # 无法得知id error: { code: -32700, message: Parse error } } self.write_message(error_response) # 如果body还没收全继续循环读取 def handle_message(self, message): 根据JSON-RPC消息的method字段进行路由分发。 method message.get(method) msg_id message.get(id) # 请求才有id通知没有 logger.info(fReceived method: {method}, id: {msg_id}) if method initialize: self.handle_initialize(message, msg_id) elif method tools/list: self.handle_tools_list(message, msg_id) elif method tools/call: self.handle_tools_call(message, msg_id) elif method notifications/shutdown: self.handle_shutdown(message) elif method exit: self.shutdown_requested True else: logger.warning(fUnknown method: {method}) # 对于未知的请求有id需要回复错误 if msg_id is not None: error_response { jsonrpc: 2.0, id: msg_id, error: { code: -32601, message: fMethod not found: {method} } } self.write_message(error_response) # 具体的消息处理器将在下面实现...关键点解析消息边界协议MCP没有使用简单的换行符分隔JSON而是采用了类似HTTP的Content-Length头。这是为了支持消息体内包含任意内容包括换行符。我们的循环必须正确解析这个头部并读取指定长度的Body。请求与通知JSON-RPC 2.0区分请求Request有id字段需要回复和通知Notification无id字段无需回复。initialize、tools/list、tools/call是请求notifications/shutdown和exit是通知。错误处理对无法解析的JSONParse error和未知的方法Method not found必须按照JSON-RPC规范返回错误对象这是实现一个健壮Server的基础。2.3 实现核心消息处理器现在我们来填充上面用到的几个核心消息处理器。def handle_initialize(self, message, msg_id): 处理initialize请求返回Server的能力声明。 params message.get(params, {}) client_info params.get(clientInfo, {}) logger.info(fClient initialized: {client_info}) # 构建响应结果 result { protocolVersion: 2024-11-05, # 使用当前稳定的协议版本 capabilities: { tools: {} # 我们声明支持Tools功能 # 未来还可以声明resources: {}, prompts: {} 等 }, serverInfo: { name: Simple Time Server, version: 0.1.0 } } response { jsonrpc: 2.0, id: msg_id, result: result } self.write_message(response) # 发送initialized通知 initialized_notification { jsonrpc: 2.0, method: notifications/initialized, params: {} # 此通知无参数 } self.write_message(initialized_notification) self.initialized True logger.info(Server initialized and notified client.) def handle_tools_list(self, message, msg_id): 处理tools/list请求返回所有可用工具的定义。 if not self.initialized: # 理论上Client应在收到initialized通知后才发此请求。 # 但为健壮性考虑这里也检查状态。 error_response { jsonrpc: 2.0, id: msg_id, error: { code: -32002, message: Server not initialized } } self.write_message(error_response) return # 定义我们的唯一工具get_server_time tools [ { name: get_server_time, description: 获取服务器的当前时间返回时间戳和格式化字符串。, inputSchema: { type: object, properties: { format: { type: string, description: 可选的时间格式字符串如%Y-%m-%d %H:%M:%S。默认为ISO格式。, default: iso } } } } ] result {tools: tools} response { jsonrpc: 2.0, id: msg_id, result: result } self.write_message(response) def handle_tools_call(self, message, msg_id): 处理tools/call请求执行具体的工具并返回结果。 params message.get(params, {}) tool_name params.get(name) arguments params.get(arguments, {}) if tool_name get_server_time: # 执行工具逻辑 fmt arguments.get(format, iso) now time.time() if fmt iso: formatted datetime.fromtimestamp(now).isoformat() else: try: formatted datetime.fromtimestamp(now).strftime(fmt) except Exception as e: formatted fFormat error: {e} result_content { content: [ { type: text, text: fServer Time (UTC): {formatted}\nTimestamp: {now} } ] } response { jsonrpc: 2.0, id: msg_id, result: result_content } else: # 工具不存在 response { jsonrpc: 2.0, id: msg_id, error: { code: -32602, message: fTool not found: {tool_name} } } self.write_message(response) def handle_shutdown(self, message): 处理shutdown通知。 logger.info(Received shutdown notification.) # 这里可以执行一些清理操作如关闭数据库连接等。 # 无需回复因为是通知。 # Client之后会发送exit通知。关键点解析能力声明Capabilities在initialize_result中capabilities字段是Server的“功能菜单”。我们只声明了tools: {}表示支持工具功能。即使是一个空对象{}也必须有这个键这是协议约定。如果你想支持资源Resources就需要在这里声明resources: {}。工具定义Tool Definitiontools/list_result返回的列表里每个工具都必须有name、description和inputSchema。inputSchema是一个JSON Schema对象用于描述输入参数的结构。这允许Client如Claude在调用前就理解参数格式甚至生成调用界面。工具调用结果Tool Call Resulttools/call_result的result字段结构是固定的必须包含一个content数组。数组中的每个元素目前通常是{type: text, text: ...}。这是MCP协议规定的AI助手可呈现的内容格式。严格的顺序initialized通知必须在initialize_result之后立即发送。很多自研Server的第一个坑就是漏发或错发这个通知导致Client一直等待超时后连接失败。2.4 启动脚本与权限设置最后添加启动代码并确保文件有可执行权限。if __name__ __main__: server SimpleMCPServer() try: server.run() except Exception as e: logger.error(fServer crashed: {e}) traceback.print_exc() sys.exit(1)给脚本加上执行权限chmod x server.py。现在一个纯手工打造的、符合MCP协议的Server就完成了。你可以尝试用最原始的方式测试它echo ... | python3 server.py但这很麻烦。更有效的方式是结合Claude Desktop和抓包工具来验证这正是下一章的内容。注意在生产环境中强烈建议使用官方SDK如modelcontextprotocol/sdkfor JavaScript/TypeScript,mcpfor Python它们帮你处理了所有这些底层协议细节、错误处理和生命周期管理。但通过这次手写你获得了对协议最深刻的理解这是解决一切诡异问题的终极武器。3. 使用抓包工具透视MCP协议通信现在我们有了一个自研的Server但它真的能工作吗当它和Claude Desktop连接时到底在“说”些什么光看日志不够直观我们需要直接窥视stdin/stdout上流动的原始数据。在Windows上我们可以使用强大的Fiddler Everywhere或Wireshark配置稍复杂而在macOS/Linux上script命令或socat是更轻量的选择。这里我将展示一种跨平台的、基于Python的“中间人”抓包方法它原理清晰且能直接输出格式化JSON非常适合学习和调试。3.1 构建一个简单的Stdio流量转发与记录器我们写一个简单的Python脚本作为“中间人”Man-in-the-Middle。它位于Claude DesktopClient和我们的SimpleMCPServer之间双向转发数据同时将经过的每一条消息打印到控制台或文件。#!/usr/bin/env python3 # mitm_logger.py import sys import json import subprocess import threading import time from queue import Queue, Empty def log_message(direction, data): 格式化并记录一条消息。 try: # 尝试解析为JSON并美化输出 message json.loads(data) pretty_json json.dumps(message, indent2, ensure_asciiFalse) print(f\n{*60}) print(f[{direction}] JSON-RPC Message:) print(pretty_json) print(f{*60}\n) except json.JSONDecodeError: # 如果不是完整的JSON可能是分片或者头部信息直接打印原始数据 print(f\n[{direction}] Raw Data (non-JSON or partial):) print(repr(data[:500])) # 只打印前500字符 def read_and_forward(stream_from, stream_to, queue, direction): 从一个流读取数据转发到另一个流同时将数据放入队列供记录。 try: while True: # 读取一块数据这里简化处理实际应像Server一样解析Content-Length # 为了演示我们每次读4KB chunk stream_from.read(4096) if not chunk: break # EOF # 转发 stream_to.write(chunk) stream_to.flush() # 记录 queue.put((direction, chunk)) except Exception as e: print(fError in {direction} reader: {e}, filesys.stderr) finally: queue.put(None) # 发送结束信号 def main(): if len(sys.argv) 2: print(fUsage: {sys.argv[0]} command [args...], filesys.stderr) sys.exit(1) # 启动子进程我们的MCP Server server_proc subprocess.Popen( sys.argv[1:], # 从命令行参数获取要启动的Server命令 stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, # 将Server的stderr重定向到管道我们单独处理 textFalse, # 使用二进制模式避免编码问题 bufsize0 ) # 队列用于在主线程中记录消息避免多线程打印混乱 log_queue Queue() # 启动两个线程分别处理Client-Server和Server-Client的数据流 # 注意这里我们模拟Client即我们手动输入。实际应与Claude Desktop集成。 # 为了演示我们创建一个虚拟的“Client”线程来发送初始化请求。 def dummy_client(): # 构建一个标准的initialize请求 init_request { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2024-11-05, clientInfo: { name: MCP-Debug-Client, version: 1.0.0 } } } request_json json.dumps(init_request) message fContent-Length: {len(request_json)}\r\n\r\n{request_json} server_proc.stdin.write(message.encode()) server_proc.stdin.flush() log_queue.put((C-S, message.encode())) time.sleep(0.5) # 等待Server回复 # 接着发送tools/list请求 list_request { jsonrpc: 2.0, id: 2, method: tools/list, params: {} } request_json json.dumps(list_request) message fContent-Length: {len(request_json)}\r\n\r\n{request_json} server_proc.stdin.write(message.encode()) server_proc.stdin.flush() log_queue.put((C-S, message.encode())) client_thread threading.Thread(targetdummy_client) client_thread.start() # 启动从Server stdout到我们stdout的转发线程S-C stdout_thread threading.Thread( targetread_and_forward, args(server_proc.stdout, sys.stdout.buffer, log_queue, S-C) ) stdout_thread.daemon True stdout_thread.start() # 单独处理Server的stderr直接打印到我们的stderr def read_stderr(): while True: chunk server_proc.stderr.read(4096) if not chunk: break sys.stderr.buffer.write(b[Server STDERR] chunk) sys.stderr.flush() stderr_thread threading.Thread(targetread_stderr) stderr_thread.daemon True stderr_thread.start() # 主线程从队列中取出数据并记录 end_count 0 try: while end_count 2: # 等待两个转发线程结束 try: item log_queue.get(timeout0.1) if item is None: end_count 1 else: direction, data item log_message(direction, data) except Empty: continue except KeyboardInterrupt: print(\nInterrupted by user.) finally: server_proc.terminate() server_proc.wait() if __name__ __main__: main()这个脚本是一个简化版的调试工具。它启动了我们的Server并模拟了一个最简单的Client发送两条请求。运行它python3 mitm_logger.py python3 server.py。你将在控制台看到类似这样的输出 [C-S] JSON-RPC Message: { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2024-11-05, clientInfo: { name: MCP-Debug-Client, version: 1.0.0 } } } [S-C] JSON-RPC Message: { jsonrpc: 2.0, id: 1, result: { protocolVersion: 2024-11-05, capabilities: { tools: {} }, serverInfo: { name: Simple Time Server, version: 0.1.0 } } } [S-C] JSON-RPC Message: { jsonrpc: 2.0, method: notifications/initialized, params: {} } 太棒了你亲眼看到了协议的三次握手Client的initialize请求Server的initialize_result响应以及紧随其后的notifications/initialized通知。这就是协议规范在字节层面的具体体现。3.2 与真实Claude Desktop集成并抓包上面的调试器是自娱自乐。要让我们的Server被真正的Claude Desktop调用需要将其配置为Claude的MCP Server。这里以macOS上的Claude Desktop为例其他平台原理类似找到Claude Desktop的配置文件通常位于~/Library/Application Support/Claude/claude_desktop_config.json。编辑配置文件在mcpServers部分添加我们的Server。{ mcpServers: { simple-time-server: { command: python3, args: [/ABSOLUTE/PATH/TO/YOUR/my_mcp_server/server.py] } } }重启Claude Desktop。现在真正的挑战来了。当Claude Desktop启动时它会自动运行我们配置的Server命令并通过stdio与之通信。我们如何抓取这个“进程间”的流量直接修改Claude的配置指向我们的mitm_logger.py是一个办法但更通用的方法是使用系统级的调试工具。在macOS/Linux上可以使用script命令或socatscript命令可以记录一个终端会话的所有输入输出。你可以写一个包装脚本先启动script记录再在script启动的shell中运行Claude Desktop。但这会记录所有终端活动需要过滤。socat更精确它可以创建一对虚拟的PTY伪终端让Claude和我们的Server通过这对PTY通信而socat自己则在中间双向拷贝数据并打印。命令类似socat -x -v PTY,link/tmp/mcp-client,raw,echo0 PTY,link/tmp/mcp-server,raw,echo0。然后将Claude配置中的command改为socat创建的PTY路径之一将我们的Server连接到另一个PTY。-v参数会将所有传输的十六进制和ASCII数据打印到stderr。在Windows上可以使用Fiddler Everywhere或Wireshark对于stdio流这些网络抓包工具默认抓不到。但MCP也支持stdio和sseServer-Sent Events等传输方式。如果是sse基于HTTP那么Fiddler和Wireshark就能直接抓包。不过Claude Desktop目前主要使用stdio与本地Server通信。一个可行的办法是编写一个本地的“代理Server”。这个代理作为一个HTTP/sse Server运行Claude配置连接这个代理。代理再将请求通过stdio转发给我们真正的Python Server同时记录所有流量。这需要更多代码但提供了最大的灵活性和可观察性。实操心得在开发初期最快速的验证方式其实就是我们写的mitm_logger.py。先确保你的Server能正确响应最基本的initialize和tools/list请求。一旦这两步通了集成到Claude Desktop的成功率就大大提升。集成失败时首先查看Claude Desktop的日志位置因平台而异通常会有“Server exited with code 1”或“Failed to parse message”之类的错误这些线索能帮你快速定位是启动问题还是协议格式问题。4. 协议拆解从抓包结果分析关键帧与常见陷阱通过抓包我们获得了协议通信的第一手资料。现在让我们像一个协议分析师一样仔细审视这些“关键帧”并总结出开发中最容易踩中的陷阱。4.1 关键帧深度解析让我们基于抓包结果逐一拆解每个核心消息帧的必须字段和可选字段。1. 初始化请求帧 (Client - Server){ jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2024-11-05, clientInfo: { name: Claude Desktop, version: 1.5.0 }, // 可选客户端可能声明自己支持的能力但Server应忽略 capabilities: {} } }必须字段jsonrpc,id,method,params。protocolVersion这是致命关键。必须与Server实现的版本一致。如果你手写的Server填错了版本比如用了旧的草案日期Claude Desktop可能会直接拒绝通信。目前稳定版本是2024-11-05。clientInfo仅供参考用于日志记录。2. 初始化响应帧 (Server - Client){ jsonrpc: 2.0, id: 1, // 必须与请求的id对应 result: { protocolVersion: 2024-11-05, // 必须与请求中的一致 capabilities: { tools: {}, resources: {}, // 可选如果支持资源则需声明 prompts: {} // 可选如果支持提示词则需声明 }, serverInfo: { name: My Server, version: 0.1.0 } } }capabilities对象即使某个功能你暂时没有实现比如resources也不能省略这个键。你必须返回一个对象其内部可以是空的{}但键必须存在。例如只支持工具就必须是capabilities: {tools: {}}。如果完全返回capabilities: {}Client会认为你什么都不支持。serverInfo可选但强烈建议提供有助于在Client界面识别Server。3. 工具列表响应帧 (Server - Client){ jsonrpc: 2.0, id: 2, result: { tools: [ { name: get_weather, description: 获取指定城市的天气。, inputSchema: { type: object, properties: { city: { type: string, description: 城市名称如北京。 }, unit: { type: string, enum: [celsius, fahrenheit], description: 温度单位。, default: celsius } }, required: [city] // 明确哪些参数是必需的 } } ] } }inputSchema的重要性这不仅仅是文档。AI助手如Claude会利用这个JSON Schema来理解如何调用你的工具。description字段要清晰required数组要准确。如果required里写了city但调用时没传Client端可能会直接报错而不会发请求到Server。enum和default善用这些字段可以极大提升用户体验。enum限定了输入范围default提供了默认值。4. 工具调用结果帧 (Server - Client){ jsonrpc: 2.0, id: 123, result: { content: [ { type: text, text: 当前温度是22度天气晴朗。, mimeType: text/plain // 可选 }, { type: image, data: base64EncodedString..., mimeType: image/png } ] } }content数组这是返回数据的核心。目前主要支持text和image类型。text是必须支持的。你可以返回多条content它们会被组合呈现给用户。错误响应如果工具执行出错应返回标准的JSON-RPC错误帧而不是在result里放错误信息。{ jsonrpc: 2.0, id: 123, error: { code: -32000, message: Failed to fetch weather data: Network error., data: {city: Beijing} // 可选提供错误上下文 } }4.2 开发中最常见的五个陷阱与排查清单结合手写Server和抓包分析的经验我总结了五个最容易导致集成失败的坑点并给出排查思路。陷阱一Content-Length头格式错误或计算不准现象Client日志报“Parse error”或“Invalid frame”Server可能收不到任何请求或收到乱码。根因没有严格按照Content-Length: 数字\r\n\r\n消息体的格式发送。数字必须是消息体字节数对于UTF-8字符串一个中文字符可能是3个字节。或者在读取时没有正确解析这个头部导致消息边界错乱。排查用抓包工具看原始字节流。检查\r\n\r\n是否完整检查Content-Length的值是否等于后面JSON字符串的实际字节长度len(json_str.encode(‘utf-8’))。陷阱二initialized通知漏发或顺序错误现象Client在发送initialize后一直等待超时后断开连接日志显示“Server did not send initialized notification”。根因Server在回复initialize_result后没有立即发送notifications/initialized方法通知。或者错误地将其作为initialize_result的一部分。排查抓包确认在initialize_result响应帧之后是否紧跟着一个method为notifications/initialized、且没有id字段的通知帧。陷阱三capabilities声明不完整或错误现象Server能初始化但Client里看不到任何工具或资源。根因initialize_result中的capabilities字段缺少对应的键。例如你实现了工具但返回的是capabilities: {}或capabilities: {resources: {}}。排查抓包查看initialize_result确认capabilities.tools是否存在即使为空对象{}。如果你实现了资源则capabilities.resources也必须存在。陷阱四工具调用结果格式不符合协议现象工具调用后AI助手没有显示结果或显示乱码。根因tools/call_result的result字段结构错误。最常见的是直接返回了字符串或自定义对象而不是{content: [{type: text, text: ...}]}结构。排查抓包查看tools/call_result帧确保result.content是一个数组且里面的对象有type和text字段。陷阱五JSON-RPCid不匹配或缺失现象某些请求没有回应或者Client报告“收到未知的响应”。根因对于请求有id的响应必须携带完全相同的id值。如果Server在处理请求时弄丢了id或者响应了一个与任何未完成请求都不匹配的idClient会无法处理该响应。排查在Server代码中确保将请求中的id原封不动地复制到响应对象中。对于通知如initialized则绝不能有id字段。当你遇到连接或调用问题时按照这份清单逐一核对抓包数据几乎能解决90%的协议层问题。剩下的10%可能就是环境配置或权限问题了。