議并構建MCP Server實戰(zhàn))
1. 項目背景與核心價值為什么需要關注這個協(xié)議最近在折騰一個智能家居項目想把一個叫“小鴻AI WS63”的智能設備我猜它可能是個帶AI功能的傳感器或者控制器的數(shù)據(jù)實時地接入到我自己的服務器應用里。設備廠商給的文檔很簡陋只說支持WebSocket連接但具體怎么連、數(shù)據(jù)格式是啥、心跳怎么維持一概沒提。這讓我想起了之前集成各種IoT設備時踩過的坑協(xié)議不透明對接全靠猜最后要么通信不穩(wěn)定要么數(shù)據(jù)解析出錯。于是我決定把這次逆向和對接“小鴻AI WS63”與自建MCP Server模型上下文協(xié)議服務器的整個過程以及最終梳理出的WebSocket通信協(xié)議細節(jié)完整地記錄下來。MCP Server是當前AI應用開發(fā)中的一個熱門概念它本質上是一個標準化的接口服務器用于為大型語言模型如GPT、Claude提供工具調用和上下文數(shù)據(jù)。讓設備數(shù)據(jù)通過WebSocket流入MCP Server就能讓AI模型實時感知到物理世界的變化從而實現(xiàn)更智能的自動化決策。這個協(xié)議詳解的價值在于它不僅僅是一份技術文檔。對于開發(fā)者而言它是一份可以直接“抄作業(yè)”的對接指南能幫你省去大量抓包、猜格式、試錯的時間。對于架構師它展示了如何為一個私有協(xié)議設備構建穩(wěn)定、可擴展的實時數(shù)據(jù)通道。整個過程涉及網絡抓包、協(xié)議逆向、數(shù)據(jù)編解碼、連接保活策略等一整套實戰(zhàn)技能無論你是做物聯(lián)網、實時通信還是AI應用集成都能從中找到共鳴和參考。2. 逆向工程起點從零捕獲并解析原始通信流當面對一個沒有文檔的通信協(xié)議時第一步永遠是抓包。我的目標設備“小鴻AI WS63”提供了一個Wi-Fi配置模式使其能連接到我的本地網絡。這樣我就能在同一個局域網內用我的開發(fā)機進行流量監(jiān)聽。2.1 工具選型與網絡環(huán)境搭建我選擇了Wireshark作為主要的抓包工具因為它對網絡協(xié)議的解析能力最強。為了能抓到設備與服務器假設是廠商云端之間的通信我需要讓設備的流量經過我的電腦。有兩種常見方案設置代理在電腦上運行一個HTTP/WebSocket代理如mitmproxy并將設備的網關設置為我的電腦IP。但很多嵌入式設備不支持配置代理此路不通。ARP欺騙/網關鏡像這是更通用的方法。我使用了arpspoof工具需配合iptables讓我電腦成為設備和路由器之間的“中間人”。具體命令如下# 啟用IP轉發(fā) echo 1 /proc/sys/net/ipv4/ip_forward # 對設備進行ARP欺騙讓它認為我的電腦是網關 arpspoof -i eth0 -t 設備IP 網關IP # 對網關進行ARP欺騙讓它認為我的電腦是設備 arpspoof -i eth0 -t 網關IP 設備IP # 將經過我電腦的WebSocket流量通常端口443或自定義端口重定向到本機的一個端口方便Wireshark抓取 iptables -t nat -A PREROUTING -p tcp --dport 443 -j REDIRECT --to-port 8443然后在Wireshark中監(jiān)聽eth0接口并設置過濾條件tcp.port 8443。這樣設備與真實服務器之間的TLS加密流量就被我“劫持”并解密前提是我在電腦上安裝了設備的CA證書對于非加密WebSocket則更簡單。2.2 首次連接與協(xié)議特征識別啟動抓包后給設備上電。在Wireshark中我很快看到了設備發(fā)起的TCP連接。通過跟蹤TCP流Follow - TCP Stream原始數(shù)據(jù)呈現(xiàn)出來。關鍵特征出現(xiàn)了一個標準的HTTP Upgrade請求GET /ws/v1/data HTTP/1.1 Host: device-cloud.example.com:8883 Upgrade: websocket Connection: Upgrade Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ Sec-WebSocket-Version: 13這證實了它使用WebSocket并且路徑是/ws/v1/data。服務器回復101 Switching Protocols握手成功。隨后是二進制數(shù)據(jù)流。WebSocket幀的Payload部分是二進制的無法直接閱讀。這是逆向的核心難點。2.3 二進制載荷解析從亂碼到結構Wireshark可以解析WebSocket幀但payload需要自己分析。我將一段時間內的二進制payload全部導出保存為raw.bin文件。接下來就是“猜”結構。根據(jù)經驗這類IoT設備數(shù)據(jù)幀通常包含幀頭Header固定的字節(jié)序列如0xAA 0x55用于標識幀開始。長度字段Length指示后續(xù)數(shù)據(jù)部分的長度。命令字/類型Cmd/Type標識這條消息是傳感器數(shù)據(jù)、心跳、配置請求還是響應。數(shù)據(jù)載荷Payload具體的業(yè)務數(shù)據(jù)。校驗和ChecksumCRC8或CRC16用于驗證數(shù)據(jù)完整性。我用十六進制編輯器打開raw.bin并寫了一個簡單的Python腳本進行模式搜索import binascii with open(raw.bin, rb) as f: data f.read() hex_str binascii.hexlify(data).decode(utf-8) # 尋找可能的兩字節(jié)幀頭如 AA55 for i in range(0, len(hex_str)-4, 2): if hex_str[i:i4] aa55: print(fPossible header at byte offset {i//2}: {hex_str[i:i20]})通過對比多個數(shù)據(jù)包我發(fā)現(xiàn)了一個規(guī)律每隔大約30秒就會有一個非常短例如8字節(jié)的數(shù)據(jù)包交互。這極有可能是心跳包。鎖定這些短包對比它們的hex值我假設了最簡單的結構[0xAA, 0x55, 0x01, 0x00, 0xXX, 0xXX]其中0x01可能是心跳命令最后兩字節(jié)是校驗和。通過計算常見的CRC8/CRC16算法與最后兩字節(jié)的匹配我驗證了校驗算法是CRC16-CCITT初始值0xFFFF。注意逆向工程中心跳包和錯誤響應包往往是突破口因為它們結構相對固定且重復出現(xiàn)。先搞定它們再攻克復雜的數(shù)據(jù)包。3. “小鴻AI WS63” WebSocket協(xié)議幀格式全解構經過對數(shù)十個數(shù)據(jù)包的比對、分類和驗證我最終還原出了“小鴻AI WS63”設備端使用的WebSocket二進制幀格式。這不是官方標準而是基于實際通信逆向得出的事實標準。3.1 通用幀結構Big-Endian所有上行設備-服務器和下行服務器-設備的數(shù)據(jù)幀都遵循以下結構字節(jié)偏移字段名長度字節(jié)說明0-1幀頭Header2固定為0xAA55標識一幀的開始。2協(xié)議版本Version1當前協(xié)議版本觀察到的值為0x01。3命令字Command1定義幀的類型是協(xié)議解析的核心。4-5序列號Seq2請求/響應對的標識用于匹配響應。通常由發(fā)起方設置響應方回顯。6-7數(shù)據(jù)載荷長度Length2不包含幀頭、版本、命令、序列號、長度、校驗和這前10個字節(jié)。即后續(xù)Payload的實際字節(jié)數(shù)。8-(8N-1)數(shù)據(jù)載荷PayloadN可變長度由Length字段定義。內容格式根據(jù)Command不同而不同。(8N)-(8N1)校驗和Checksum2從幀頭0xAA55到Payload最后一個字節(jié)的所有數(shù)據(jù)進行CRC16-CCITT計算的結果初始值0xFFFF。關鍵點解析字節(jié)序所有多字節(jié)字段幀頭、序列號、長度、校驗和均采用大端序Big-Endian即網絡字節(jié)序。這在解析時至關重要例如長度字段0x00 0x10表示十進制16而不是4096。長度計算Length len(Payload)。整個幀的總長度是10 Length字節(jié)。校驗范圍校驗和的計算不包含它自身。這是CRC校驗的常規(guī)做法。3.2 核心命令字Command枚舉與含義通過對交互流程的歸類我識別出以下關鍵命令命令值Hex方向名稱描述0x01設備 - 服務器心跳請求Heartbeat設備定期發(fā)送用于保活。Payload通常為空Length0。0x81服務器 - 設備心跳響應Heartbeat ACK服務器對心跳的確認。Payload通常為空。0x02設備 - 服務器傳感器數(shù)據(jù)上報Sensor Data設備上報其采集的數(shù)據(jù)如溫度、濕度、AI識別結果。Payload結構復雜見下文。0x82服務器 - 設備數(shù)據(jù)上報確認Data ACK服務器確認收到數(shù)據(jù)。Payload可包含服務器時間戳用于同步。0x03服務器 - 設備配置下發(fā)Config Update服務器向設備發(fā)送新的配置參數(shù)。0x83設備 - 服務器配置響應Config Response設備對配置下發(fā)的響應成功/失敗及原因。0x04設備 - 服務器事件上報Event Report上報非周期性的AI事件如檢測到特定物體、異常報警。0x84服務器 - 設備事件響應Event ACK服務器確認收到事件。3.3 關鍵Payload結構詳解以傳感器數(shù)據(jù)0x02為例這是最復雜的部分。設備上報的數(shù)據(jù)可能包含多種傳感器信息。其Payload結構是一個TLVType-Length-Value的嵌套結構。外層結構設備級字節(jié)偏移字段長度說明0-5設備ID6設備的唯一標識符通常是MAC地址或燒錄的ID。6-9時間戳4設備端的Unix時間戳秒級。10-11傳感器數(shù)量N2本次上報包含的傳感器數(shù)據(jù)塊個數(shù)。12-...傳感器數(shù)據(jù)塊列表可變包含N個傳感器數(shù)據(jù)塊每個塊是一個TLV結構。內層結構傳感器數(shù)據(jù)塊 - TLV 每個傳感器數(shù)據(jù)塊由三部分組成Type (1字節(jié))傳感器類型。例如0x01溫度0x02濕度0x10AI識別結果JSON字符串0x11電池電壓。Length (2字節(jié))后續(xù)Value字段的字節(jié)長度。Value (可變長度)傳感器讀數(shù)的具體值。其格式由Type決定0x01溫度2字節(jié)有符號整數(shù)單位0.1°C。例如0x00 0x96 150 15.0°C。0x02濕度1字節(jié)無符號整數(shù)單位1%。例如0x45 69%。0x10AI結果UTF-8編碼的JSON字符串。例如{object: person, confidence: 0.87, bbox: [10,20,100,200]}。0x11電壓2字節(jié)無符號整數(shù)單位mV。例如0x0B 0xB8 3000 3.000V。一個完整的數(shù)據(jù)上報幀解析示例十六進制AA 55 01 02 00 01 00 1A // 幀頭 |版本|命令|序列號 |長度(26) 00 00 00 00 00 01 5F 90 7B 2C 00 02 // 設備ID(00:00:00:00:00:01) |時間戳(0x5F907B2C) |傳感器數(shù)量(2) 01 00 02 00 96 02 00 01 45 11 00 02 0B B8 // 傳感器塊1: Type0x01(溫度), Len2, Value0x0096(15.0°C) // 傳感器塊2: Type0x02(濕度), Len1, Value0x45(69%) // 傳感器塊3: Type0x11(電壓), Len2, Value0x0BB8(3.000V) A1 7B // 校驗和 (CRC16 of data from AA55 to ...B8)4. 構建自有的MCP Server與協(xié)議適配層了解了設備協(xié)議下一步就是構建我們自己的MCP Server來接收和處理這些數(shù)據(jù)。我們的目標不是模擬原廠服務器而是實現(xiàn)一個協(xié)議適配層將設備的私有協(xié)議轉換為MCP標準格式供AI模型使用。4.1 MCP Server核心概念與選型MCPModel Context Protocol的核心思想是為AI模型提供一個統(tǒng)一的“工具箱”和“數(shù)據(jù)源”接口。一個MCP Server可以聲明一系列Tools函數(shù)和Resources數(shù)據(jù)客戶端如Claude Desktop、Cursor可以發(fā)現(xiàn)并調用它們。我選擇使用TypeScript/Node.js和modelcontextprotocol/sdk官方SDK來構建Server。原因如下生態(tài)成熟Node.js的WebSocket庫如ws非常強大適合處理高并發(fā)連接。開發(fā)效率TypeScript的強類型有助于定義復雜的協(xié)議數(shù)據(jù)結構減少錯誤。SDK支持官方SDK封裝了MCP的底層通信JSON-RPC over STDIO/SSE讓我們專注于業(yè)務邏輯。4.2 項目結構與核心模塊設計project/ ├── package.json ├── tsconfig.json ├── src/ │ ├── index.ts # 主入口啟動MCP Server和WebSocket Server │ ├── protocol/ # 協(xié)議解析層 │ │ ├── decoder.ts # 二進制幀解碼器 │ │ ├── encoder.ts # 二進制幀編碼器 │ │ └── types.ts # 協(xié)議相關的類型定義命令、傳感器類型等 │ ├── device-manager.ts # 設備連接管理、狀態(tài)維護 │ ├── mcp-handlers.ts # MCP Tools和Resources的實現(xiàn) │ └── ws-server.ts # 專用于小鴻設備的WebSocket服務器 └── ...4.3 WebSocket服務器實現(xiàn)與協(xié)議解碼在ws-server.ts中我們使用ws庫創(chuàng)建一個WebSocket服務器監(jiān)聽特定端口如8888。// ws-server.ts import WebSocket, { WebSocketServer } from ws; import { decodeFrame, isHeartbeat, parseSensorData } from ./protocol/decoder; import { encodeHeartbeatAck } from ./protocol/encoder; import { DeviceManager } from ./device-manager; export function createDeviceWebSocketServer(port: number, deviceManager: DeviceManager) { const wss new WebSocketServer({ port }); wss.on(connection, (ws: WebSocket, request) { const clientIp request.socket.remoteAddress; console.log(新的設備連接來自: ${clientIp}); let deviceId: string | null null; ws.on(message, (data: Buffer) { try { // 1. 解碼二進制幀 const frame decodeFrame(data); console.log(收到命令: 0x${frame.command.toString(16).padStart(2, 0)}, 序列號: ${frame.seq}); // 2. 處理心跳 if (isHeartbeat(frame)) { const ackFrame encodeHeartbeatAck(frame.seq); ws.send(ackFrame); console.log(已發(fā)送心跳響應給設備 ${deviceId}); return; } // 3. 處理數(shù)據(jù)上報 (0x02) if (frame.command 0x02) { const sensorReport parseSensorData(frame.payload); deviceId sensorReport.deviceId; // 從數(shù)據(jù)中提取設備ID // 將數(shù)據(jù)交給設備管理器處理 deviceManager.updateDeviceData(deviceId, { ...sensorReport, lastSeen: Date.now(), wsConnection: ws }); // 發(fā)送確認幀 (可選根據(jù)協(xié)議需要) // const ackFrame encodeDataAck(frame.seq, Date.now()); // ws.send(ackFrame); } // 4. 處理其他命令... } catch (error) { console.error(解析設備數(shù)據(jù)幀失敗:, error); // 可以考慮發(fā)送一個錯誤響應幀或者直接關閉連接 ws.close(1002, Protocol error); } }); ws.on(close, () { console.log(設備連接關閉: ${deviceId || clientIp}); if (deviceId) { deviceManager.removeDevice(deviceId); } }); ws.on(error, (error) { console.error(WebSocket錯誤: ${error}); }); }); console.log(小鴻設備WebSocket服務器已啟動在端口 ${port}); return wss; }decoder.ts是核心實現(xiàn)了幀的驗證和解碼// protocol/decoder.ts import crc from crc; import { SensorDataReport, DeviceFrame } from ./types; export function decodeFrame(buffer: Buffer): DeviceFrame { // 1. 基礎長度檢查 if (buffer.length 10) { throw new Error(幀長度過短); } // 2. 檢查幀頭 if (buffer.readUInt16BE(0) ! 0xaa55) { throw new Error(無效的幀頭); } // 3. 提取字段 const version buffer.readUInt8(2); const command buffer.readUInt8(3); const seq buffer.readUInt16BE(4); const length buffer.readUInt16BE(6); // 4. 校驗長度字段是否與實際Buffer長度一致 if (buffer.length ! 10 length) { throw new Error(長度字段不匹配。預期: ${10 length}, 實際: ${buffer.length}); } // 5. 計算并校驗CRC const expectedChecksum buffer.readUInt16BE(8 length); const dataToCheck buffer.slice(0, 8 length); // 從幀頭到Payload結束 const actualChecksum crc.crc16ccitt(dataToCheck, 0xffff); if (expectedChecksum ! actualChecksum) { throw new Error(CRC校驗失敗。預期: 0x${expectedChecksum.toString(16)}, 實際: 0x${actualChecksum.toString(16)}); } // 6. 提取Payload const payload buffer.slice(8, 8 length); return { version, command, seq, length, payload }; } export function parseSensorData(payload: Buffer): SensorDataReport { // 解析3.3節(jié)描述的TLV結構 const deviceId payload.slice(0, 6).toString(hex); // 轉為MAC格式 const timestamp payload.readUInt32BE(6); const sensorCount payload.readUInt16BE(10); const sensors []; let offset 12; for (let i 0; i sensorCount; i) { const type payload.readUInt8(offset); const len payload.readUInt16BE(offset 1); const valueBuffer payload.slice(offset 3, offset 3 len); // 根據(jù)type解析valueBuffer let value: any; switch (type) { case 0x01: // 溫度 value valueBuffer.readInt16BE(0) / 10.0; break; case 0x02: // 濕度 value valueBuffer.readUInt8(0); break; case 0x10: // AI結果 (JSON字符串) value JSON.parse(valueBuffer.toString(utf8)); break; // ... 其他類型 default: value valueBuffer; } sensors.push({ type, value }); offset 3 len; } return { deviceId, timestamp, sensors }; }5. 將設備數(shù)據(jù)暴露為MCP資源與工具設備數(shù)據(jù)接入后我們需要通過MCP協(xié)議將其暴露出去。這主要通過實現(xiàn)Resources和Tools來完成。5.1 定義MCP資源Resources資源代表可讀的數(shù)據(jù)源。我們可以將每個設備的最新狀態(tài)定義為一個資源。// mcp-handlers.ts import { Server } from modelcontextprotocol/sdk/server/index.js; import { DeviceManager } from ./device-manager; export function setupMcpResources(server: Server, deviceManager: DeviceManager) { // 聲明一個資源列表例如所有設備 server.setRequestHandler(ListResourcesRequestSchema, async () { const devices deviceManager.getAllDevices(); const resources: Resource[] devices.map(device ({ uri: device://${device.id}/state, name: 設備 ${device.id} 的實時狀態(tài), description: 包含傳感器讀數(shù)、在線狀態(tài)等信息, mimeType: application/json })); // 還可以聲明一個匯總資源 resources.push({ uri: device://summary, name: 所有設備狀態(tài)匯總, description: 所有已連接設備的快照, mimeType: application/json }); return { resources }; }); // 處理資源讀取請求 server.setRequestHandler(ReadResourceRequestSchema, async (request) { const url new URL(request.params.uri); if (url.pathname /summary) { const devices deviceManager.getAllDevices(); return { contents: [{ uri: request.params.uri, mimeType: application/json, text: JSON.stringify({ timestamp: Date.now(), onlineCount: devices.length, devices: devices.map(d ({ id: d.id, lastSeen: d.lastSeen, data: d.latestData // 最新傳感器數(shù)據(jù) })) }, null, 2) }] }; } // 匹配 device://{deviceId}/state const match url.pathname.match(/^\/([^\/])\/state$/); if (match) { const deviceId match[1]; const device deviceManager.getDevice(deviceId); if (!device) { throw new Error(設備 ${deviceId} 未找到); } return { contents: [{ uri: request.params.uri, mimeType: application/json, text: JSON.stringify(device.latestData, null, 2) }] }; } throw new Error(未知資源: ${request.params.uri}); }); }5.2 定義MCP工具Tools工具代表可執(zhí)行的函數(shù)。我們可以提供一些控制或查詢工具。// mcp-handlers.ts export function setupMcpTools(server: Server, deviceManager: DeviceManager) { server.setRequestHandler(ListToolsRequestSchema, async () { return { tools: [ { name: get_device_history, description: 獲取指定設備在過去一段時間內的歷史傳感器數(shù)據(jù), inputSchema: { type: object, properties: { deviceId: { type: string, description: 設備ID (如MAC地址) }, durationMinutes: { type: number, description: 查詢最近多少分鐘的數(shù)據(jù), default: 60 } }, required: [deviceId] } }, { name: send_device_command, description: 向指定設備發(fā)送控制命令如下發(fā)配置, inputSchema: { type: object, properties: { deviceId: { type: string }, command: { type: string, enum: [reboot, get_config, set_report_interval], description: 要執(zhí)行的命令 }, params: { type: object, description: 命令參數(shù) } }, required: [deviceId, command] } } ] }; }); server.setRequestHandler(CallToolRequestSchema, async (request) { const { name, arguments: args } request.params; if (name get_device_history) { const { deviceId, durationMinutes 60 } args as any; // 從設備管理器或數(shù)據(jù)庫中查詢歷史數(shù)據(jù) const history deviceManager.getDeviceHistory(deviceId, durationMinutes); return { content: [{ type: text, text: 設備 ${deviceId} 最近${durationMinutes}分鐘的歷史數(shù)據(jù)\n${JSON.stringify(history, null, 2)} }] }; } if (name send_device_command) { const { deviceId, command, params } args as any; const device deviceManager.getDevice(deviceId); if (!device || !device.wsConnection) { throw new Error(設備 ${deviceId} 未連接); } // 根據(jù)命令編碼對應的下行幀 const commandFrame encodeControlCommand(deviceId, command, params); device.wsConnection.send(commandFrame); return { content: [{ type: text, text: 已向設備 ${deviceId} 發(fā)送命令: ${command} }] }; } throw new Error(未知工具: ${name}); }); }5.3 主程序入口與集成最后在index.ts中將所有部分串聯(lián)起來// index.ts import { Server } from modelcontextprotocol/sdk/server/index.js; import { StdioServerTransport } from modelcontextprotocol/sdk/server/stdio.js; import { DeviceManager } from ./device-manager; import { createDeviceWebSocketServer } from ./ws-server; import { setupMcpResources, setupMcpTools } from ./mcp-handlers; async function main() { // 1. 初始化設備管理器用于在內存中維護設備狀態(tài) const deviceManager new DeviceManager(); // 2. 啟動面向小鴻設備的WebSocket服務器 const wsServer createDeviceWebSocketServer(8888, deviceManager); // 3. 創(chuàng)建并啟動MCP Server const server new Server( { name: xiaohong-ai-mcp-server, version: 1.0.0, }, { capabilities: { resources: {}, // 啟用資源功能 tools: {}, // 啟用工具功能 }, } ); // 4. 設置MCP請求處理器 setupMcpResources(server, deviceManager); setupMcpTools(server, deviceManager); // 5. 使用Stdio傳輸層啟動這是最通用的方式被Claude Desktop等客戶端支持 const transport new StdioServerTransport(); await server.connect(transport); console.error(小鴻AI MCP Server 已通過Stdio啟動); } main().catch((error) { console.error(服務器啟動失敗:, error); process.exit(1); });6. 實戰(zhàn)部署、測試與排錯指南理論完成代碼就緒接下來就是真刀真槍的測試和部署。6.1 模擬設備測試在真實設備接入前編寫一個模擬客戶端進行全鏈路測試至關重要。# simulator.py - 模擬小鴻AI WS63設備 import asyncio import websockets import struct import crc16 import json import time async def simulate_device(): uri ws://localhost:8888/ws/v1/data # 連接到我們自建的服務器 async with websockets.connect(uri) as websocket: device_id bytes([0x00, 0x11, 0x22, 0x33, 0x44, 0x55]) seq 1 while True: # 1. 發(fā)送心跳 heartbeat_frame build_frame(0x01, seq, b) await websocket.send(heartbeat_frame) print(f發(fā)送心跳, seq{seq}) seq 1 # 等待心跳響應可選 # try: # response await asyncio.wait_for(websocket.recv(), timeout2.0) # print(f收到響應: {response.hex()}) # except asyncio.TimeoutError: # print(心跳響應超時) # 2. 每隔一段時間發(fā)送傳感器數(shù)據(jù) await asyncio.sleep(30) # 模擬30秒上報間隔 sensor_data build_sensor_payload(device_id, [ (0x01, struct.pack(h, 245)), # 溫度24.5°C (0x02, bytes([65])), # 濕度65% (0x10, json.dumps({object: cat, confidence: 0.92}).encode(utf-8)) ]) data_frame build_frame(0x02, seq, sensor_data) await websocket.send(data_frame) print(f發(fā)送傳感器數(shù)據(jù), seq{seq}) seq 1 await asyncio.sleep(30) def build_frame(command, seq, payload): 構建協(xié)議幀 header b\xaa\x55 version b\x01 cmd bytes([command]) seq_bytes seq.to_bytes(2, big) length len(payload).to_bytes(2, big) # 計算CRC從header到payload data_for_crc header version cmd seq_bytes length payload checksum crc16.crc16xmodem(data_for_crc, 0xffff) checksum_bytes checksum.to_bytes(2, big) return data_for_crc checksum_bytes def build_sensor_payload(device_id, sensor_list): 構建傳感器數(shù)據(jù)Payload timestamp int(time.time()).to_bytes(4, big) count len(sensor_list).to_bytes(2, big) payload device_id timestamp count for sensor_type, value_bytes in sensor_list: payload bytes([sensor_type]) len(value_bytes).to_bytes(2, big) value_bytes return payload asyncio.run(simulate_device())運行模擬器觀察MCP Server的日志確認連接建立、數(shù)據(jù)解析、資源更新都正常。6.2 連接真實設備將“小鴻AI WS63”設備配置到與MCP Server同一局域網并修改其服務器地址為MCP Server的IP和端口8888。這通常需要通過設備廠商的配網APP或本地配置頁面完成。設備連接后在服務器日志中應看到新的連接和源源不斷的數(shù)據(jù)上報。6.3 使用MCP客戶端測試啟動一個支持MCP的客戶端如Claude Desktop。在其設置中添加自定義MCP Server配置為我們的服務器例如通過Stdio調用我們的Node.js腳本。連接成功后你就可以在Claude的對話中直接使用我們定義的工具和資源了。例如你可以問Claude“/get_device_historydeviceId00:11:22:33:44:55 durationMinutes10”它會調用我們的工具并返回格式化后的數(shù)據(jù)。或者你可以讓它“讀取一下所有設備的當前狀態(tài)”它會通過device://summary資源獲取信息。6.4 常見問題與排錯連接失敗檢查防火墻是否開放了8888端口。檢查設備端配置的服務器地址和端口是否正確。在服務器端使用netstat -an | grep 8888查看端口監(jiān)聽狀態(tài)。數(shù)據(jù)解析錯誤首先檢查日志中的CRC錯誤。如果CRC頻繁失敗可能是字節(jié)序假設錯誤大端/小端或者幀頭判斷有誤。用Wireshark抓取MCP Server收到的原始數(shù)據(jù)與設備直連原廠服務器的數(shù)據(jù)進行比對。MCP客戶端無法發(fā)現(xiàn)工具/資源確保MCP Server通過Stdio正確啟動并且客戶端配置的command能正確啟動你的腳本。檢查服務器日志是否有初始化錯誤。MCP協(xié)議要求Server在啟動后立即發(fā)送initialize請求確認你的SDK處理正確。連接不穩(wěn)定頻繁斷開檢查心跳機制。確保你的服務器能正確響應0x01心跳請求并回復0x81。檢查設備的心跳間隔如果服務器在超時時間內未收到任何數(shù)據(jù)心跳或業(yè)務數(shù)據(jù)應主動斷開連接并清理資源。性能問題當連接數(shù)百個設備時需要考慮優(yōu)化。ws庫本身性能很好瓶頸可能在業(yè)務邏輯。可以將設備狀態(tài)更新改為異步非阻塞操作考慮使用Redis等內存數(shù)據(jù)庫存儲設備狀態(tài)而非全部放在Node.js內存中。整個對接過程從抓包逆向到最終實現(xiàn)一個功能完整的MCP Server最耗時的部分往往是協(xié)議細節(jié)的確認和邊界情況的處理。這份詳解希望能為你提供一個清晰的路線圖和可復用的代碼框架讓你在對接類似私有協(xié)議物聯(lián)網設備時能少走彎路快速構建起連接物理世界與AI模型的可靠橋梁。 逆向解析小鴻AI WS63 WebSocket協(xié)議并構建MCP Server實戰(zhàn)
1. 項目背景與核心價值為什么需要關注這個協(xié)議最近在折騰一個智能家居項目想把一個叫“小鴻AI WS63”的智能設備我猜它可能是個帶AI功能的傳感器或者控制器的數(shù)據(jù)實時地接入到我自己的服務器應用里。設備廠商給的文檔很簡陋只說支持WebSocket連接但具體怎么連、數(shù)據(jù)格式是啥、心跳怎么維持一概沒提。這讓我想起了之前集成各種IoT設備時踩過的坑協(xié)議不透明對接全靠猜最后要么通信不穩(wěn)定要么數(shù)據(jù)解析出錯。于是我決定把這次逆向和對接“小鴻AI WS63”與自建MCP Server模型上下文協(xié)議服務器的整個過程以及最終梳理出的WebSocket通信協(xié)議細節(jié)完整地記錄下來。MCP Server是當前AI應用開發(fā)中的一個熱門概念它本質上是一個標準化的接口服務器用于為大型語言模型如GPT、Claude提供工具調用和上下文數(shù)據(jù)。讓設備數(shù)據(jù)通過WebSocket流入MCP Server就能讓AI模型實時感知到物理世界的變化從而實現(xiàn)更智能的自動化決策。這個協(xié)議詳解的價值在于它不僅僅是一份技術文檔。對于開發(fā)者而言它是一份可以直接“抄作業(yè)”的對接指南能幫你省去大量抓包、猜格式、試錯的時間。對于架構師它展示了如何為一個私有協(xié)議設備構建穩(wěn)定、可擴展的實時數(shù)據(jù)通道。整個過程涉及網絡抓包、協(xié)議逆向、數(shù)據(jù)編解碼、連接保活策略等一整套實戰(zhàn)技能無論你是做物聯(lián)網、實時通信還是AI應用集成都能從中找到共鳴和參考。2. 逆向工程起點從零捕獲并解析原始通信流當面對一個沒有文檔的通信協(xié)議時第一步永遠是抓包。我的目標設備“小鴻AI WS63”提供了一個Wi-Fi配置模式使其能連接到我的本地網絡。這樣我就能在同一個局域網內用我的開發(fā)機進行流量監(jiān)聽。2.1 工具選型與網絡環(huán)境搭建我選擇了Wireshark作為主要的抓包工具因為它對網絡協(xié)議的解析能力最強。為了能抓到設備與服務器假設是廠商云端之間的通信我需要讓設備的流量經過我的電腦。有兩種常見方案設置代理在電腦上運行一個HTTP/WebSocket代理如mitmproxy并將設備的網關設置為我的電腦IP。但很多嵌入式設備不支持配置代理此路不通。ARP欺騙/網關鏡像這是更通用的方法。我使用了arpspoof工具需配合iptables讓我電腦成為設備和路由器之間的“中間人”。具體命令如下# 啟用IP轉發(fā) echo 1 /proc/sys/net/ipv4/ip_forward # 對設備進行ARP欺騙讓它認為我的電腦是網關 arpspoof -i eth0 -t 設備IP 網關IP # 對網關進行ARP欺騙讓它認為我的電腦是設備 arpspoof -i eth0 -t 網關IP 設備IP # 將經過我電腦的WebSocket流量通常端口443或自定義端口重定向到本機的一個端口方便Wireshark抓取 iptables -t nat -A PREROUTING -p tcp --dport 443 -j REDIRECT --to-port 8443然后在Wireshark中監(jiān)聽eth0接口并設置過濾條件tcp.port 8443。這樣設備與真實服務器之間的TLS加密流量就被我“劫持”并解密前提是我在電腦上安裝了設備的CA證書對于非加密WebSocket則更簡單。2.2 首次連接與協(xié)議特征識別啟動抓包后給設備上電。在Wireshark中我很快看到了設備發(fā)起的TCP連接。通過跟蹤TCP流Follow - TCP Stream原始數(shù)據(jù)呈現(xiàn)出來。關鍵特征出現(xiàn)了一個標準的HTTP Upgrade請求GET /ws/v1/data HTTP/1.1 Host: device-cloud.example.com:8883 Upgrade: websocket Connection: Upgrade Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ Sec-WebSocket-Version: 13這證實了它使用WebSocket并且路徑是/ws/v1/data。服務器回復101 Switching Protocols握手成功。隨后是二進制數(shù)據(jù)流。WebSocket幀的Payload部分是二進制的無法直接閱讀。這是逆向的核心難點。2.3 二進制載荷解析從亂碼到結構Wireshark可以解析WebSocket幀但payload需要自己分析。我將一段時間內的二進制payload全部導出保存為raw.bin文件。接下來就是“猜”結構。根據(jù)經驗這類IoT設備數(shù)據(jù)幀通常包含幀頭Header固定的字節(jié)序列如0xAA 0x55用于標識幀開始。長度字段Length指示后續(xù)數(shù)據(jù)部分的長度。命令字/類型Cmd/Type標識這條消息是傳感器數(shù)據(jù)、心跳、配置請求還是響應。數(shù)據(jù)載荷Payload具體的業(yè)務數(shù)據(jù)。校驗和ChecksumCRC8或CRC16用于驗證數(shù)據(jù)完整性。我用十六進制編輯器打開raw.bin并寫了一個簡單的Python腳本進行模式搜索import binascii with open(raw.bin, rb) as f: data f.read() hex_str binascii.hexlify(data).decode(utf-8) # 尋找可能的兩字節(jié)幀頭如 AA55 for i in range(0, len(hex_str)-4, 2): if hex_str[i:i4] aa55: print(fPossible header at byte offset {i//2}: {hex_str[i:i20]})通過對比多個數(shù)據(jù)包我發(fā)現(xiàn)了一個規(guī)律每隔大約30秒就會有一個非常短例如8字節(jié)的數(shù)據(jù)包交互。這極有可能是心跳包。鎖定這些短包對比它們的hex值我假設了最簡單的結構[0xAA, 0x55, 0x01, 0x00, 0xXX, 0xXX]其中0x01可能是心跳命令最后兩字節(jié)是校驗和。通過計算常見的CRC8/CRC16算法與最后兩字節(jié)的匹配我驗證了校驗算法是CRC16-CCITT初始值0xFFFF。注意逆向工程中心跳包和錯誤響應包往往是突破口因為它們結構相對固定且重復出現(xiàn)。先搞定它們再攻克復雜的數(shù)據(jù)包。3. “小鴻AI WS63” WebSocket協(xié)議幀格式全解構經過對數(shù)十個數(shù)據(jù)包的比對、分類和驗證我最終還原出了“小鴻AI WS63”設備端使用的WebSocket二進制幀格式。這不是官方標準而是基于實際通信逆向得出的事實標準。3.1 通用幀結構Big-Endian所有上行設備-服務器和下行服務器-設備的數(shù)據(jù)幀都遵循以下結構字節(jié)偏移字段名長度字節(jié)說明0-1幀頭Header2固定為0xAA55標識一幀的開始。2協(xié)議版本Version1當前協(xié)議版本觀察到的值為0x01。3命令字Command1定義幀的類型是協(xié)議解析的核心。4-5序列號Seq2請求/響應對的標識用于匹配響應。通常由發(fā)起方設置響應方回顯。6-7數(shù)據(jù)載荷長度Length2不包含幀頭、版本、命令、序列號、長度、校驗和這前10個字節(jié)。即后續(xù)Payload的實際字節(jié)數(shù)。8-(8N-1)數(shù)據(jù)載荷PayloadN可變長度由Length字段定義。內容格式根據(jù)Command不同而不同。(8N)-(8N1)校驗和Checksum2從幀頭0xAA55到Payload最后一個字節(jié)的所有數(shù)據(jù)進行CRC16-CCITT計算的結果初始值0xFFFF。關鍵點解析字節(jié)序所有多字節(jié)字段幀頭、序列號、長度、校驗和均采用大端序Big-Endian即網絡字節(jié)序。這在解析時至關重要例如長度字段0x00 0x10表示十進制16而不是4096。長度計算Length len(Payload)。整個幀的總長度是10 Length字節(jié)。校驗范圍校驗和的計算不包含它自身。這是CRC校驗的常規(guī)做法。3.2 核心命令字Command枚舉與含義通過對交互流程的歸類我識別出以下關鍵命令命令值Hex方向名稱描述0x01設備 - 服務器心跳請求Heartbeat設備定期發(fā)送用于保活。Payload通常為空Length0。0x81服務器 - 設備心跳響應Heartbeat ACK服務器對心跳的確認。Payload通常為空。0x02設備 - 服務器傳感器數(shù)據(jù)上報Sensor Data設備上報其采集的數(shù)據(jù)如溫度、濕度、AI識別結果。Payload結構復雜見下文。0x82服務器 - 設備數(shù)據(jù)上報確認Data ACK服務器確認收到數(shù)據(jù)。Payload可包含服務器時間戳用于同步。0x03服務器 - 設備配置下發(fā)Config Update服務器向設備發(fā)送新的配置參數(shù)。0x83設備 - 服務器配置響應Config Response設備對配置下發(fā)的響應成功/失敗及原因。0x04設備 - 服務器事件上報Event Report上報非周期性的AI事件如檢測到特定物體、異常報警。0x84服務器 - 設備事件響應Event ACK服務器確認收到事件。3.3 關鍵Payload結構詳解以傳感器數(shù)據(jù)0x02為例這是最復雜的部分。設備上報的數(shù)據(jù)可能包含多種傳感器信息。其Payload結構是一個TLVType-Length-Value的嵌套結構。外層結構設備級字節(jié)偏移字段長度說明0-5設備ID6設備的唯一標識符通常是MAC地址或燒錄的ID。6-9時間戳4設備端的Unix時間戳秒級。10-11傳感器數(shù)量N2本次上報包含的傳感器數(shù)據(jù)塊個數(shù)。12-...傳感器數(shù)據(jù)塊列表可變包含N個傳感器數(shù)據(jù)塊每個塊是一個TLV結構。內層結構傳感器數(shù)據(jù)塊 - TLV 每個傳感器數(shù)據(jù)塊由三部分組成Type (1字節(jié))傳感器類型。例如0x01溫度0x02濕度0x10AI識別結果JSON字符串0x11電池電壓。Length (2字節(jié))后續(xù)Value字段的字節(jié)長度。Value (可變長度)傳感器讀數(shù)的具體值。其格式由Type決定0x01溫度2字節(jié)有符號整數(shù)單位0.1°C。例如0x00 0x96 150 15.0°C。0x02濕度1字節(jié)無符號整數(shù)單位1%。例如0x45 69%。0x10AI結果UTF-8編碼的JSON字符串。例如{object: person, confidence: 0.87, bbox: [10,20,100,200]}。0x11電壓2字節(jié)無符號整數(shù)單位mV。例如0x0B 0xB8 3000 3.000V。一個完整的數(shù)據(jù)上報幀解析示例十六進制AA 55 01 02 00 01 00 1A // 幀頭 |版本|命令|序列號 |長度(26) 00 00 00 00 00 01 5F 90 7B 2C 00 02 // 設備ID(00:00:00:00:00:01) |時間戳(0x5F907B2C) |傳感器數(shù)量(2) 01 00 02 00 96 02 00 01 45 11 00 02 0B B8 // 傳感器塊1: Type0x01(溫度), Len2, Value0x0096(15.0°C) // 傳感器塊2: Type0x02(濕度), Len1, Value0x45(69%) // 傳感器塊3: Type0x11(電壓), Len2, Value0x0BB8(3.000V) A1 7B // 校驗和 (CRC16 of data from AA55 to ...B8)4. 構建自有的MCP Server與協(xié)議適配層了解了設備協(xié)議下一步就是構建我們自己的MCP Server來接收和處理這些數(shù)據(jù)。我們的目標不是模擬原廠服務器而是實現(xiàn)一個協(xié)議適配層將設備的私有協(xié)議轉換為MCP標準格式供AI模型使用。4.1 MCP Server核心概念與選型MCPModel Context Protocol的核心思想是為AI模型提供一個統(tǒng)一的“工具箱”和“數(shù)據(jù)源”接口。一個MCP Server可以聲明一系列Tools函數(shù)和Resources數(shù)據(jù)客戶端如Claude Desktop、Cursor可以發(fā)現(xiàn)并調用它們。我選擇使用TypeScript/Node.js和modelcontextprotocol/sdk官方SDK來構建Server。原因如下生態(tài)成熟Node.js的WebSocket庫如ws非常強大適合處理高并發(fā)連接。開發(fā)效率TypeScript的強類型有助于定義復雜的協(xié)議數(shù)據(jù)結構減少錯誤。SDK支持官方SDK封裝了MCP的底層通信JSON-RPC over STDIO/SSE讓我們專注于業(yè)務邏輯。4.2 項目結構與核心模塊設計project/ ├── package.json ├── tsconfig.json ├── src/ │ ├── index.ts # 主入口啟動MCP Server和WebSocket Server │ ├── protocol/ # 協(xié)議解析層 │ │ ├── decoder.ts # 二進制幀解碼器 │ │ ├── encoder.ts # 二進制幀編碼器 │ │ └── types.ts # 協(xié)議相關的類型定義命令、傳感器類型等 │ ├── device-manager.ts # 設備連接管理、狀態(tài)維護 │ ├── mcp-handlers.ts # MCP Tools和Resources的實現(xiàn) │ └── ws-server.ts # 專用于小鴻設備的WebSocket服務器 └── ...4.3 WebSocket服務器實現(xiàn)與協(xié)議解碼在ws-server.ts中我們使用ws庫創(chuàng)建一個WebSocket服務器監(jiān)聽特定端口如8888。// ws-server.ts import WebSocket, { WebSocketServer } from ws; import { decodeFrame, isHeartbeat, parseSensorData } from ./protocol/decoder; import { encodeHeartbeatAck } from ./protocol/encoder; import { DeviceManager } from ./device-manager; export function createDeviceWebSocketServer(port: number, deviceManager: DeviceManager) { const wss new WebSocketServer({ port }); wss.on(connection, (ws: WebSocket, request) { const clientIp request.socket.remoteAddress; console.log(新的設備連接來自: ${clientIp}); let deviceId: string | null null; ws.on(message, (data: Buffer) { try { // 1. 解碼二進制幀 const frame decodeFrame(data); console.log(收到命令: 0x${frame.command.toString(16).padStart(2, 0)}, 序列號: ${frame.seq}); // 2. 處理心跳 if (isHeartbeat(frame)) { const ackFrame encodeHeartbeatAck(frame.seq); ws.send(ackFrame); console.log(已發(fā)送心跳響應給設備 ${deviceId}); return; } // 3. 處理數(shù)據(jù)上報 (0x02) if (frame.command 0x02) { const sensorReport parseSensorData(frame.payload); deviceId sensorReport.deviceId; // 從數(shù)據(jù)中提取設備ID // 將數(shù)據(jù)交給設備管理器處理 deviceManager.updateDeviceData(deviceId, { ...sensorReport, lastSeen: Date.now(), wsConnection: ws }); // 發(fā)送確認幀 (可選根據(jù)協(xié)議需要) // const ackFrame encodeDataAck(frame.seq, Date.now()); // ws.send(ackFrame); } // 4. 處理其他命令... } catch (error) { console.error(解析設備數(shù)據(jù)幀失敗:, error); // 可以考慮發(fā)送一個錯誤響應幀或者直接關閉連接 ws.close(1002, Protocol error); } }); ws.on(close, () { console.log(設備連接關閉: ${deviceId || clientIp}); if (deviceId) { deviceManager.removeDevice(deviceId); } }); ws.on(error, (error) { console.error(WebSocket錯誤: ${error}); }); }); console.log(小鴻設備WebSocket服務器已啟動在端口 ${port}); return wss; }decoder.ts是核心實現(xiàn)了幀的驗證和解碼// protocol/decoder.ts import crc from crc; import { SensorDataReport, DeviceFrame } from ./types; export function decodeFrame(buffer: Buffer): DeviceFrame { // 1. 基礎長度檢查 if (buffer.length 10) { throw new Error(幀長度過短); } // 2. 檢查幀頭 if (buffer.readUInt16BE(0) ! 0xaa55) { throw new Error(無效的幀頭); } // 3. 提取字段 const version buffer.readUInt8(2); const command buffer.readUInt8(3); const seq buffer.readUInt16BE(4); const length buffer.readUInt16BE(6); // 4. 校驗長度字段是否與實際Buffer長度一致 if (buffer.length ! 10 length) { throw new Error(長度字段不匹配。預期: ${10 length}, 實際: ${buffer.length}); } // 5. 計算并校驗CRC const expectedChecksum buffer.readUInt16BE(8 length); const dataToCheck buffer.slice(0, 8 length); // 從幀頭到Payload結束 const actualChecksum crc.crc16ccitt(dataToCheck, 0xffff); if (expectedChecksum ! actualChecksum) { throw new Error(CRC校驗失敗。預期: 0x${expectedChecksum.toString(16)}, 實際: 0x${actualChecksum.toString(16)}); } // 6. 提取Payload const payload buffer.slice(8, 8 length); return { version, command, seq, length, payload }; } export function parseSensorData(payload: Buffer): SensorDataReport { // 解析3.3節(jié)描述的TLV結構 const deviceId payload.slice(0, 6).toString(hex); // 轉為MAC格式 const timestamp payload.readUInt32BE(6); const sensorCount payload.readUInt16BE(10); const sensors []; let offset 12; for (let i 0; i sensorCount; i) { const type payload.readUInt8(offset); const len payload.readUInt16BE(offset 1); const valueBuffer payload.slice(offset 3, offset 3 len); // 根據(jù)type解析valueBuffer let value: any; switch (type) { case 0x01: // 溫度 value valueBuffer.readInt16BE(0) / 10.0; break; case 0x02: // 濕度 value valueBuffer.readUInt8(0); break; case 0x10: // AI結果 (JSON字符串) value JSON.parse(valueBuffer.toString(utf8)); break; // ... 其他類型 default: value valueBuffer; } sensors.push({ type, value }); offset 3 len; } return { deviceId, timestamp, sensors }; }5. 將設備數(shù)據(jù)暴露為MCP資源與工具設備數(shù)據(jù)接入后我們需要通過MCP協(xié)議將其暴露出去。這主要通過實現(xiàn)Resources和Tools來完成。5.1 定義MCP資源Resources資源代表可讀的數(shù)據(jù)源。我們可以將每個設備的最新狀態(tài)定義為一個資源。// mcp-handlers.ts import { Server } from modelcontextprotocol/sdk/server/index.js; import { DeviceManager } from ./device-manager; export function setupMcpResources(server: Server, deviceManager: DeviceManager) { // 聲明一個資源列表例如所有設備 server.setRequestHandler(ListResourcesRequestSchema, async () { const devices deviceManager.getAllDevices(); const resources: Resource[] devices.map(device ({ uri: device://${device.id}/state, name: 設備 ${device.id} 的實時狀態(tài), description: 包含傳感器讀數(shù)、在線狀態(tài)等信息, mimeType: application/json })); // 還可以聲明一個匯總資源 resources.push({ uri: device://summary, name: 所有設備狀態(tài)匯總, description: 所有已連接設備的快照, mimeType: application/json }); return { resources }; }); // 處理資源讀取請求 server.setRequestHandler(ReadResourceRequestSchema, async (request) { const url new URL(request.params.uri); if (url.pathname /summary) { const devices deviceManager.getAllDevices(); return { contents: [{ uri: request.params.uri, mimeType: application/json, text: JSON.stringify({ timestamp: Date.now(), onlineCount: devices.length, devices: devices.map(d ({ id: d.id, lastSeen: d.lastSeen, data: d.latestData // 最新傳感器數(shù)據(jù) })) }, null, 2) }] }; } // 匹配 device://{deviceId}/state const match url.pathname.match(/^\/([^\/])\/state$/); if (match) { const deviceId match[1]; const device deviceManager.getDevice(deviceId); if (!device) { throw new Error(設備 ${deviceId} 未找到); } return { contents: [{ uri: request.params.uri, mimeType: application/json, text: JSON.stringify(device.latestData, null, 2) }] }; } throw new Error(未知資源: ${request.params.uri}); }); }5.2 定義MCP工具Tools工具代表可執(zhí)行的函數(shù)。我們可以提供一些控制或查詢工具。// mcp-handlers.ts export function setupMcpTools(server: Server, deviceManager: DeviceManager) { server.setRequestHandler(ListToolsRequestSchema, async () { return { tools: [ { name: get_device_history, description: 獲取指定設備在過去一段時間內的歷史傳感器數(shù)據(jù), inputSchema: { type: object, properties: { deviceId: { type: string, description: 設備ID (如MAC地址) }, durationMinutes: { type: number, description: 查詢最近多少分鐘的數(shù)據(jù), default: 60 } }, required: [deviceId] } }, { name: send_device_command, description: 向指定設備發(fā)送控制命令如下發(fā)配置, inputSchema: { type: object, properties: { deviceId: { type: string }, command: { type: string, enum: [reboot, get_config, set_report_interval], description: 要執(zhí)行的命令 }, params: { type: object, description: 命令參數(shù) } }, required: [deviceId, command] } } ] }; }); server.setRequestHandler(CallToolRequestSchema, async (request) { const { name, arguments: args } request.params; if (name get_device_history) { const { deviceId, durationMinutes 60 } args as any; // 從設備管理器或數(shù)據(jù)庫中查詢歷史數(shù)據(jù) const history deviceManager.getDeviceHistory(deviceId, durationMinutes); return { content: [{ type: text, text: 設備 ${deviceId} 最近${durationMinutes}分鐘的歷史數(shù)據(jù)\n${JSON.stringify(history, null, 2)} }] }; } if (name send_device_command) { const { deviceId, command, params } args as any; const device deviceManager.getDevice(deviceId); if (!device || !device.wsConnection) { throw new Error(設備 ${deviceId} 未連接); } // 根據(jù)命令編碼對應的下行幀 const commandFrame encodeControlCommand(deviceId, command, params); device.wsConnection.send(commandFrame); return { content: [{ type: text, text: 已向設備 ${deviceId} 發(fā)送命令: ${command} }] }; } throw new Error(未知工具: ${name}); }); }5.3 主程序入口與集成最后在index.ts中將所有部分串聯(lián)起來// index.ts import { Server } from modelcontextprotocol/sdk/server/index.js; import { StdioServerTransport } from modelcontextprotocol/sdk/server/stdio.js; import { DeviceManager } from ./device-manager; import { createDeviceWebSocketServer } from ./ws-server; import { setupMcpResources, setupMcpTools } from ./mcp-handlers; async function main() { // 1. 初始化設備管理器用于在內存中維護設備狀態(tài) const deviceManager new DeviceManager(); // 2. 啟動面向小鴻設備的WebSocket服務器 const wsServer createDeviceWebSocketServer(8888, deviceManager); // 3. 創(chuàng)建并啟動MCP Server const server new Server( { name: xiaohong-ai-mcp-server, version: 1.0.0, }, { capabilities: { resources: {}, // 啟用資源功能 tools: {}, // 啟用工具功能 }, } ); // 4. 設置MCP請求處理器 setupMcpResources(server, deviceManager); setupMcpTools(server, deviceManager); // 5. 使用Stdio傳輸層啟動這是最通用的方式被Claude Desktop等客戶端支持 const transport new StdioServerTransport(); await server.connect(transport); console.error(小鴻AI MCP Server 已通過Stdio啟動); } main().catch((error) { console.error(服務器啟動失敗:, error); process.exit(1); });6. 實戰(zhàn)部署、測試與排錯指南理論完成代碼就緒接下來就是真刀真槍的測試和部署。6.1 模擬設備測試在真實設備接入前編寫一個模擬客戶端進行全鏈路測試至關重要。# simulator.py - 模擬小鴻AI WS63設備 import asyncio import websockets import struct import crc16 import json import time async def simulate_device(): uri ws://localhost:8888/ws/v1/data # 連接到我們自建的服務器 async with websockets.connect(uri) as websocket: device_id bytes([0x00, 0x11, 0x22, 0x33, 0x44, 0x55]) seq 1 while True: # 1. 發(fā)送心跳 heartbeat_frame build_frame(0x01, seq, b) await websocket.send(heartbeat_frame) print(f發(fā)送心跳, seq{seq}) seq 1 # 等待心跳響應可選 # try: # response await asyncio.wait_for(websocket.recv(), timeout2.0) # print(f收到響應: {response.hex()}) # except asyncio.TimeoutError: # print(心跳響應超時) # 2. 每隔一段時間發(fā)送傳感器數(shù)據(jù) await asyncio.sleep(30) # 模擬30秒上報間隔 sensor_data build_sensor_payload(device_id, [ (0x01, struct.pack(h, 245)), # 溫度24.5°C (0x02, bytes([65])), # 濕度65% (0x10, json.dumps({object: cat, confidence: 0.92}).encode(utf-8)) ]) data_frame build_frame(0x02, seq, sensor_data) await websocket.send(data_frame) print(f發(fā)送傳感器數(shù)據(jù), seq{seq}) seq 1 await asyncio.sleep(30) def build_frame(command, seq, payload): 構建協(xié)議幀 header b\xaa\x55 version b\x01 cmd bytes([command]) seq_bytes seq.to_bytes(2, big) length len(payload).to_bytes(2, big) # 計算CRC從header到payload data_for_crc header version cmd seq_bytes length payload checksum crc16.crc16xmodem(data_for_crc, 0xffff) checksum_bytes checksum.to_bytes(2, big) return data_for_crc checksum_bytes def build_sensor_payload(device_id, sensor_list): 構建傳感器數(shù)據(jù)Payload timestamp int(time.time()).to_bytes(4, big) count len(sensor_list).to_bytes(2, big) payload device_id timestamp count for sensor_type, value_bytes in sensor_list: payload bytes([sensor_type]) len(value_bytes).to_bytes(2, big) value_bytes return payload asyncio.run(simulate_device())運行模擬器觀察MCP Server的日志確認連接建立、數(shù)據(jù)解析、資源更新都正常。6.2 連接真實設備將“小鴻AI WS63”設備配置到與MCP Server同一局域網并修改其服務器地址為MCP Server的IP和端口8888。這通常需要通過設備廠商的配網APP或本地配置頁面完成。設備連接后在服務器日志中應看到新的連接和源源不斷的數(shù)據(jù)上報。6.3 使用MCP客戶端測試啟動一個支持MCP的客戶端如Claude Desktop。在其設置中添加自定義MCP Server配置為我們的服務器例如通過Stdio調用我們的Node.js腳本。連接成功后你就可以在Claude的對話中直接使用我們定義的工具和資源了。例如你可以問Claude“/get_device_historydeviceId00:11:22:33:44:55 durationMinutes10”它會調用我們的工具并返回格式化后的數(shù)據(jù)。或者你可以讓它“讀取一下所有設備的當前狀態(tài)”它會通過device://summary資源獲取信息。6.4 常見問題與排錯連接失敗檢查防火墻是否開放了8888端口。檢查設備端配置的服務器地址和端口是否正確。在服務器端使用netstat -an | grep 8888查看端口監(jiān)聽狀態(tài)。數(shù)據(jù)解析錯誤首先檢查日志中的CRC錯誤。如果CRC頻繁失敗可能是字節(jié)序假設錯誤大端/小端或者幀頭判斷有誤。用Wireshark抓取MCP Server收到的原始數(shù)據(jù)與設備直連原廠服務器的數(shù)據(jù)進行比對。MCP客戶端無法發(fā)現(xiàn)工具/資源確保MCP Server通過Stdio正確啟動并且客戶端配置的command能正確啟動你的腳本。檢查服務器日志是否有初始化錯誤。MCP協(xié)議要求Server在啟動后立即發(fā)送initialize請求確認你的SDK處理正確。連接不穩(wěn)定頻繁斷開檢查心跳機制。確保你的服務器能正確響應0x01心跳請求并回復0x81。檢查設備的心跳間隔如果服務器在超時時間內未收到任何數(shù)據(jù)心跳或業(yè)務數(shù)據(jù)應主動斷開連接并清理資源。性能問題當連接數(shù)百個設備時需要考慮優(yōu)化。ws庫本身性能很好瓶頸可能在業(yè)務邏輯。可以將設備狀態(tài)更新改為異步非阻塞操作考慮使用Redis等內存數(shù)據(jù)庫存儲設備狀態(tài)而非全部放在Node.js內存中。整個對接過程從抓包逆向到最終實現(xiàn)一個功能完整的MCP Server最耗時的部分往往是協(xié)議細節(jié)的確認和邊界情況的處理。這份詳解希望能為你提供一個清晰的路線圖和可復用的代碼框架讓你在對接類似私有協(xié)議物聯(lián)網設備時能少走彎路快速構建起連接物理世界與AI模型的可靠橋梁。