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