1. 从协议视角重新理解MCP Server
如果你已经跟着基础教程跑通了几个现成的MCP Server,比如文件读写或者简单的天气查询,可能会觉得MCP(Model 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(level=logging.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_ascii=False) content = f"Content-Length: {len(json_str)}\r\n\r\n{json_str}" sys.stdout.write(content) sys.stdout.flush() logger.debug(f"Sent: {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(f"Failed 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(f"Received 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(f"Unknown method: {method}") # 对于未知的请求(有id),需要回复错误 if msg_id is not None: error_response = { "jsonrpc": "2.0", "id": msg_id, "error": { "code": -32601, "message": f"Method 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是通知。 - 错误处理:对无法解析的JSON(Parse 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(f"Client 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 = f"Format error: {e}" result_content = { "content": [ { "type": "text", "text": f"Server 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": f"Tool 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 Definition):
tools/list_result返回的列表里,每个工具都必须有name、description和inputSchema。inputSchema是一个JSON Schema对象,用于描述输入参数的结构。这允许Client(如Claude)在调用前就理解参数格式,甚至生成调用界面。 - 工具调用结果(Tool Call Result):
tools/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(f"Server 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 Desktop(Client)和我们的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, indent=2, ensure_ascii=False) 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(f"Error in {direction} reader: {e}", file=sys.stderr) finally: queue.put(None) # 发送结束信号 def main(): if len(sys.argv) < 2: print(f"Usage: {sys.argv[0]} <command> [args...]", file=sys.stderr) sys.exit(1) # 启动子进程(我们的MCP Server) server_proc = subprocess.Popen( sys.argv[1:], # 从命令行参数获取要启动的Server命令 stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, # 将Server的stderr重定向到管道,我们单独处理 text=False, # 使用二进制模式,避免编码问题 bufsize=0 ) # 队列用于在主线程中记录消息,避免多线程打印混乱 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 = f"Content-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 = f"Content-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(target=dummy_client) client_thread.start() # 启动从Server stdout到我们stdout的转发线程(S->C) stdout_thread = threading.Thread( target=read_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(target=read_stderr) stderr_thread.daemon = True stderr_thread.start() # 主线程:从队列中取出数据并记录 end_count = 0 try: while end_count < 2: # 等待两个转发线程结束 try: item = log_queue.get(timeout=0.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命令或socat:
script命令可以记录一个终端会话的所有输入输出。你可以写一个包装脚本,先启动script记录,再在script启动的shell中运行Claude Desktop。但这会记录所有终端活动,需要过滤。socat更精确:它可以创建一对虚拟的PTY(伪终端),让Claude和我们的Server通过这对PTY通信,而socat自己则在中间双向拷贝数据并打印。命令类似:socat -x -v PTY,link=/tmp/mcp-client,raw,echo=0 PTY,link=/tmp/mcp-server,raw,echo=0。然后将Claude配置中的command改为socat创建的PTY路径之一,将我们的Server连接到另一个PTY。-v参数会将所有传输的十六进制和ASCII数据打印到stderr。
在Windows上,可以使用Fiddler Everywhere或Wireshark:
- 对于stdio流,这些网络抓包工具默认抓不到。但MCP也支持
stdio和sse(Server-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,或者响应了一个与任何未完成请求都不匹配的id,Client会无法处理该响应。 - 排查:在Server代码中,确保将请求中的
id原封不动地复制到响应对象中。对于通知(如initialized),则绝不能有id字段。
当你遇到连接或调用问题时,按照这份清单逐一核对抓包数据,几乎能解决90%的协议层问题。剩下的10%,可能就是环境配置或权限问题了。