1. 项目概述：从“小鸿AI WS63”到MCP Server的桥梁

最近在折腾一个挺有意思的项目，核心是把一个叫“小鸿AI WS63”的本地AI工具，通过WebSocket协议，接入到一个标准的MCP（Model Context Protocol） Server里。听起来有点绕，简单来说，就是让这个本地AI模型能像ChatGPT插件一样，被外部的AI应用（比如Claude Desktop、Cursor等）安全、标准化地调用。整个过程的核心，就是设计并实现一套稳定、高效的WebSocket通信协议。这可不是简单的“开个WebSocket连上就行”，里面涉及到握手认证、消息格式定义、错误处理、状态同步等一系列细节，任何一个环节没处理好，都可能让整个通信链路变得脆弱不堪。我自己在实现过程中踩了不少坑，也总结出一些能让连接更稳定、开发更顺畅的经验。如果你也在做类似AI工具集成或者需要设计一个健壮的WebSocket服务，这篇从实战中摸爬滚打出来的协议详解，应该能给你省下不少调试时间。

2. 协议整体架构与设计思路拆解

2.1 为什么是WebSocket而非HTTP？

在决定通信协议时，我们首先排除了传统的HTTP轮询。对于AI工具调用这种可能涉及长时间运行、需要服务端主动推送状态（如生成进度、流式输出）的场景，HTTP轮询不仅实时性差，还会带来巨大的无效请求开销。虽然HTTP/2的Server-Send Event (SSE) 是一个选项，但它本质上是单向的（服务端到客户端）。

WebSocket协议则完美契合了我们的需求。它在单个TCP连接上提供全双工通信，连接建立后，客户端和服务端可以随时相互发送数据帧，开销极小。这对于“小鸿AI WS63”这类模型非常重要，因为一次推理过程，服务端可能需要持续地向客户端发送token流，而客户端也可能需要中途发送“停止生成”的指令。这种低延迟、双向、持续的消息交换能力，是WebSocket的天然优势。

2.2 MCP Server的角色与约束

MCP Server在这里扮演了一个“协议转换器”或“适配器”的角色。它的核心职责是：

1. 协议标准化：对外暴露标准的MCP协议接口（通常基于JSON-RPC over WebSocket），让任何兼容MCP的客户端都能以统一的方式发现和调用工具。
2. 会话与状态管理：管理客户端连接、会话状态，并处理客户端的并发请求。
3. 适配“小鸿AI WS63”：将MCP协议格式的请求（如tools/call），翻译成“小鸿AI WS63”能理解的内部指令，通过我们设计的私有WebSocket协议发送过去；同时，将WS63返回的结果或流，重新包装成MCP协议格式的消息返回给客户端。

这意味着我们设计的WebSocket协议需要有两层：一层是MCP Client到MCP Server之间的标准MCP协议，另一层是MCP Server到“小鸿AI WS63”后端服务之间的私有协议。本文重点详解后者，即MCP Server与WS63后端之间的私有WebSocket通信协议。

2.3 核心设计目标

我们的私有协议设计围绕以下几个核心目标展开：

• 简单高效：消息结构尽可能扁平，减少不必要的嵌套和序列化/反序列化开销。
• 状态明确：每个请求都有唯一的标识，响应和错误必须能准确关联到原请求。
• 支持流式响应：必须能够处理模型逐词生成（Token Streaming）的输出模式。
• 健壮性：包含心跳机制保活，定义清晰的错误码和重连逻辑。
• 可扩展性：消息类型易于增加，以支持未来可能新增的指令，如模型切换、参数动态调整等。

3. 通信协议消息格式详解

协议采用JSON作为消息载体，因为它人类可读、易于调试，且几乎所有编程语言都有成熟的库支持。每条消息都是一个独立的JSON对象。

3.1 基础消息结构

所有消息都遵循一个基础结构，包含type和data两个顶级字段，部分消息会包含id。

{ "id": "req_123456", // 可选，请求标识符，用于匹配请求与响应 "type": "message_type_string", // 必需，消息类型 "data": {} // 必需，消息主体内容，其结构根据`type`不同而变化 }

3.2 关键消息类型解析

我们定义了以下几种核心消息类型，涵盖了从连接到调用的完整生命周期。

3.2.1 连接初始化与认证 (auth)

在WebSocket连接建立后，MCP Server必须首先发送认证消息。这是防止未授权访问的第一道关卡。

客户端（MCP Server） -> 服务端（WS63）发送：

{ "id": "auth_001", "type": "auth", "data": { "api_key": "your_pre_shared_secret_key_here", // 预共享密钥 "protocol_version": "1.0", "client_info": { "name": "mcp-server-adapter", "version": "0.1.0" } } }

注意：api_key不应硬编码在代码中，最好通过环境变量或配置文件注入。在生产环境中，可以考虑使用更复杂的机制，如JWT，但预共享密钥对于内网或可信环境下的简单对接已经足够。

服务端（WS63）响应：

• 成功:

  { "id": "auth_001", "type": "auth_response", "data": { "status": "success", "message": "Authentication successful", "model_info": { "name": "XiaoHong-WS63", "capabilities": ["text_completion", "streaming"] } } }

• 失败:

  { "id": "auth_001", "type": "error", "data": { "code": 4001, "message": "Invalid API key", "original_type": "auth" } }

  认证失败后，服务端应立即关闭WebSocket连接。

3.2.2 工具调用 (tool_call)

这是最核心的消息类型，对应MCP协议中的tools/call请求。

客户端（MCP Server） -> 服务端（WS63）发送：

{ "id": "call_789012", "type": "tool_call", "data": { "tool_name": "generate_text", // 工具名称，对应WS63的某个功能 "arguments": { "prompt": "请用Python写一个快速排序函数，并添加详细注释。", "max_tokens": 1024, "temperature": 0.7, "stream": true // 明确要求流式输出 } } }

参数设计心得：arguments的设计应尽量与WS63后端的原生API参数对齐，这样可以减少MCP Server内部的转换逻辑。stream参数至关重要，它决定了服务端是返回一个完整的响应，还是返回一系列tool_stream消息。

3.2.3 流式响应 (tool_stream)

当tool_call请求中stream为true时，服务端会返回此类型消息。一条完整的响应可能由多条tool_stream消息组成。

服务端（WS63） -> 客户端（MCP Server）发送：

{ "id": "call_789012", // 与请求ID一致 "type": "tool_stream", "data": { "content": "def", // 流式输出的一个片段 "index": 0 // 可选，片段序号，用于客户端按顺序组装 } }

{ "id": "call_789012", "type": "tool_stream", "data": { "content": " quick_sort", "index": 1 } }

// ... 更多 stream 消息

处理流式数据的技巧：客户端需要维护一个缓冲区，按index顺序（如果提供）或接收顺序拼接content。同时，要及时将每个片段转发给最终的MCP客户端（如Claude），以实现打字机效果。要特别注意网络抖动可能导致的消息乱序，虽然WebSocket能保证顺序，但客户端处理逻辑也要健壮。

3.2.4 调用结束 (tool_result)

流式或非流式调用结束后，服务端都会发送此消息，标志本次调用的终结。

服务端（WS63） -> 客户端（MCP Server）发送：

{ "id": "call_789012", "type": "tool_result", "data": { "status": "completed", // 或 "failed", "cancelled" "final_content": "def quick_sort(arr):\n \"\"\"快速排序主函数...\"\"\"\n # 详细代码...", // 非流式时，这里是完整结果；流式时，这里是所有片段的拼接（可选，方便调试） "usage": { "prompt_tokens": 25, "completion_tokens": 120, "total_tokens": 145 } } }

即使对于流式调用，发送一个包含完整内容和统计信息的tool_result也是一个好习惯，这为客户端提供了最终确认和可能的数据校验点。

3.2.5 心跳保活 (ping/pong)

为了检测连接健康状态，我们实现了简单的心跳机制。客户端定期（如每30秒）发送ping，服务端需立即回复pong。

客户端 -> 服务端：

{ "type": "ping", "data": { "timestamp": 1715000000000 } }

服务端 -> 客户端：

{ "type": "pong", "data": { "timestamp": 1715000000000 // 原样返回接收的时间戳 } }

实操心得：心跳超时时间是关键。客户端如果在规定时间（如45秒）内未收到pong响应，应判定连接已死，执行重连逻辑。同时，很多WebSocket库（如Python的websockets，JavaScript的ws）有内置的ping/pong机制，但使用应用层的心跳可以更好地与你的业务逻辑结合，例如在pong中附带服务端当前负载状态。

3.2.6 错误处理 (error)

任何阶段都可能产生错误。错误消息必须包含足够的信息用于诊断。

服务端 -> 客户端 或 客户端 -> 服务端：

{ "id": "call_789012", // 如果错误与特定请求相关，则包含该ID "type": "error", "data": { "code": 5001, "message": "Model inference timeout", "original_type": "tool_call", // 错误源于哪个类型的消息 "details": { // 可选的详细错误信息 "timeout_seconds": 30 } } }

我们定义了一套错误码区间：

• 4xxx: 客户端错误（如无效参数、认证失败）
• 5xxx: 服务端错误（如模型加载失败、内部超时）
• 6xxx: 连接与协议错误

4. 协议实现与核心环节剖析

4.1 连接生命周期管理

一个健壮的连接管理状态机至关重要。下图描绘了核心状态流转：

[初始] -> [连接建立] -> [发送auth] -> [认证成功] -> [就绪] -> [处理请求/心跳] -> [断开或错误] -> [重连决策] \-> [认证失败] -> [连接关闭] \-> [心跳超时] -> [连接关闭] -/

实现要点：

1. 连接建立后立即认证：不要在连接建立和发送认证消息之间做其他事情。
2. 就绪状态：只有收到成功的auth_response后，才将连接标记为“就绪”，允许发送tool_call请求。
3. 请求队列：在“非就绪”状态接收到的业务请求，应暂存到队列中，待就绪后依次发送。同时，要为每个请求设置超时。
4. 优雅重连：连接断开后，重连逻辑应包含指数退避策略（例如，第一次等待1秒，第二次2秒，第三次4秒...直到最大间隔）。重连后需要重新认证。

4.2 消息序列化与传输

我们选择JSON，但需要注意：

• 编码：确保双方都使用UTF-8编码。
• 大小限制：对于非常大的提示（prompt）或生成结果，考虑是否需要对消息进行分片。虽然WebSocket帧本身可以很大（理论上是9位无符号整数），但过大的单条JSON消息会影响解析性能和内存占用。一个实用的做法是，如果arguments或final_content超过一个阈值（如1MB），则记录警告日志。
• 压缩：对于文本数据，在WebSocket层面开启permessage-deflate扩展可以显著减少带宽使用，尤其是在流式传输大量小消息时。

4.3 流式处理的具体实现

流式处理是协议中最复杂的部分。以下是MCP Server端处理流式响应的伪代码逻辑：

async def handle_tool_call(request_id, tool_name, arguments): # 1. 发送请求给WS63后端 await websocket.send(json.dumps({ "id": request_id, "type": "tool_call", "data": {"tool_name": tool_name, "arguments": arguments} })) # 2. 准备流式响应给MCP客户端 buffer = [] async for message in websocket: msg = json.loads(message) if msg.get("id") != request_id: continue # 忽略其他请求的消息 if msg["type"] == "tool_stream": chunk = msg["data"]["content"] buffer.append(chunk) # 立即将chunk转发给MCP客户端（如通过Server-Sent Events） await forward_chunk_to_mcp_client(request_id, chunk) elif msg["type"] == "tool_result": final_result = msg["data"] # 可以选择用buffer里的内容，也可以用final_result里的final_content full_content = "".join(buffer) # 发送最终结果给MCP客户端 await send_final_result_to_mcp_client(request_id, full_content, final_result["usage"]) break # 处理结束 elif msg["type"] == "error": # 处理错误 handle_error(request_id, msg["data"]) break

关键点：要确保forward_chunk_to_mcp_client是非阻塞的，并且能处理下游客户端断开连接的情况。

5. 常见问题、排查技巧与优化实录

在实际开发和运维中，会遇到各种各样的问题。下面这个表格整理了一些典型问题及其排查思路：

几个独家避坑技巧：

1. 为WebSocket连接添加“标签”：在创建连接时，生成一个简短的唯一ID（如conn_7a3b）并附加到所有日志中。当同时管理多个连接时，这个标签能让你在浩如烟海的日志中快速定位问题连接的所有活动。
2. 实现“静默超时”检测：除了主动心跳，还可以监测业务消息的活跃度。如果连接在长达数分钟内没有任何消息往来（包括心跳），即使TCP连接没断，也可能意味着应用层“卡死”了，应主动断开重连。
3. 错误消息的“可追溯性”：在服务端生成错误时，除了错误码，尽量在details里包含一个内部追踪ID（如trace_id: "svr-abc123"）。这样当客户端报告错误时，你可以用这个ID快速在服务端日志里找到完整的错误上下文和堆栈跟踪。
4. 压力测试与边界值测试：一定要模拟以下场景：突然断开网络、发送畸形的JSON消息、发送超长prompt、快速连续发送大量请求。观察你的客户端和服务端是优雅降级还是直接崩溃。

6. 进阶考量与扩展方向

当基础协议稳定运行后，可以考虑以下增强方向：

1. 会话支持：扩展协议，允许一个连接内关联多个独立的“会话”。每个tool_call可以指定一个session_id，服务端可以维护会话级别的上下文（如对话历史）。这对于实现多轮对话AI应用非常有用。
2. 能力协商：在auth_response中，服务端可以更详细地声明其能力，例如支持的模型列表、每个模型的最大token限制、是否支持图像输入等。客户端可以根据这些信息动态调整界面和请求参数。
3. 二进制数据传输：如果未来需要支持图像、音频等输入输出，可以在协议中定义一种方式，将二进制数据（如图片base64或字节）嵌入到JSON消息中，或者通过WebSocket的二进制帧单独传输，并在JSON消息中引用。
4. 监控与度量：在消息中预留非干扰性的metadata字段，用于传递诊断信息，如客户端SDK版本、请求发起时间戳等。这有助于后期的性能分析和问题诊断。

设计并实现这样一套通信协议，就像在双方之间铺设一条既标准又坚固的数据管道。它一开始可能只是为了满足“连通”的基本需求，但随着业务复杂度的提升，协议本身的健壮性、可扩展性和可观测性，直接决定了整个系统的稳定上限。从“小鸿AI WS63”到MCP Server的这条路，我走通了，希望这份详尽的协议拆解和实战记录，能成为你搭建自己桥梁时的一张可靠蓝图。