十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

【MHS协议】第四章:ESP32 接入 MHS 协议实战:从零构建一个 MHS 兼容设备

【MHS协议】第四章:ESP32 接入 MHS 协议实战:从零构建一个 MHS 兼容设备 一、写在前面MHS 的现状与本文定位在开始之前有必要先说明一个现实情况Anthropic 于 2026 年 8 月 27 日发布了 MHS 研究预览版Research Preview目前尚未开源。这意味着 MHS 的完整规范、官方 SDK 和驱动程序库目前仅对受邀的研究合作伙伴开放。普通开发者暂时无法直接下载“官方 MHS 驱动”来安装到 ESP32 上。但MHS 的核心设计思想是公开的并且明确表示可与任何具有可编程接口的设备配合使用且与具体 AI 模型解耦。基于官方公布的技术架构我们完全可以在 ESP32 上模拟实现 MHS 的核心机制——标准化驱动、​​read​​/​​write​​ 原语、参考文件生成和安全边界——从而搭建一个可运行的 MHS 兼容原型。本文的目标是在现有条件下让你真正“跑起来”一个模拟 MHS 协议的 ESP32 设备并能通过 MCP 与 AI 智能体交互。二、硬件版本与协议版本硬件版本项目规格主控芯片ESP32-S3双核 Xtensa® LX7240 MHz开发板ESP32-S3-DevKitC-1或任意 ESP32-S3 开发板板载外设1 个板载 LEDGPIO2、1 个按键BOOTGPIO0可选外设DHT11 温湿度传感器GPIO4为什么选 ESP32-S3ESP32-S3 支持 Wi-Fi 和 BLE具备足够的计算能力和内存来运行 MHS 驱动逻辑和 MCP 通信栈。乐鑫的 ESP32 系列已被列为 MHS 的潜在测试平台之一。MHS 协议版本项目说明协议名称Model Hardware Standard (MHS)当前版本Research Preview v1.02026年8月27日发布协议状态研究预览版尚未开源核心原语​​read​​​、​​write​​控制通道MCP、CLI、代码文件API三、实战场景AI 通过 MHS 读取 ESP32 的环境数据我们要实现的场景是用户通过 Claude或其他 AI 智能体向 ESP32 发送自然语言指令ESP32 返回当前温湿度数据并可控地点亮/熄灭板载 LED。这模拟了 MHS 协议的核心流程设备发现ESP32 通过 MHS 驱动向网络宣告自己的存在设备描述ESP32 提供参考文件Reference File描述自己能做什么AI 控制AI 通过 ​​read​​ 读取传感器数据通过 ​​write​​ 控制 LED安全边界驱动层阻止超出安全范围的指令四、项目架构┌─────────────────────────────────────────────────────────────┐ │ AI 智能体 (Claude) │ │ (MCP Client) │ └─────────────────────┬───────────────────────────────────────┘ │ MCP 协议 (SSE / HTTP) ┌─────────────────────▼───────────────────────────────────────┐ │ MCP Server (Python) │ │ - 接收 AI 的自然语言指令 │ │ - 转换为 MHS read/write 命令 │ │ - 通过 HTTP/WebSocket 与 ESP32 通信 │ └─────────────────────┬───────────────────────────────────────┘ │ HTTP REST API ┌─────────────────────▼───────────────────────────────────────┐ │ ESP32-S3 (MHS 模拟驱动) │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ MHS Driver Layer │ │ │ │ - read(温度) → DHT11 传感器 │ │ │ │ - read(湿度) → DHT11 传感器 │ │ │ │ - write(LED, ON/OFF) → GPIO2 │ │ │ │ - 安全边界: LED 无法超频闪烁 │ │ │ └─────────────────────────────────────────────────────┘ │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ Reference File (JSON) │ │ │ │ - 设备名称: ESP32-S3 Environmental Monitor │ │ │ │ - 可读参数: temperature, humidity │ │ │ │ - 可写参数: led_state │ │ │ │ - 安全限制: led_max_freq 10Hz │ │ │ └─────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────┘五、完整代码实现5.1 ESP32 端Arduino 代码这是运行在 ESP32-S3 上的 MHS 模拟驱动固件。它实现了HTTP 服务器接收 ​​read​​ 和 ​​write​​ 命令参考文件Reference File的自动生成与提供安全边界的检查与强制执行// // 文件名: esp32_mhs_driver.ino // 硬件: ESP32-S3-DevKitC-1 // 协议: MHS Research Preview v1.0 (模拟实现) // 功能: 模拟 MHS 标准化驱动提供 read/write 原语 // #include WiFi.h #include WebServer.h #include ArduinoJson.h #include DHT.h // ---------- 配置 ---------- const char* ssid YOUR_WIFI_SSID; const char* password YOUR_WIFI_PASSWORD; // MHS 设备信息对应 Reference File const char* DEVICE_NAME ESP32-S3 Environmental Monitor; const char* DEVICE_ID mhs-esp32-s3-001; const char* MHS_VERSION research-preview-v1.0; // 引脚定义 #define LED_PIN 2 #define DHT_PIN 4 #define DHT_TYPE DHT11 // ---------- 全局对象 ---------- WebServer server(80); DHT dht(DHT_PIN, DHT_TYPE); // 设备状态 bool ledState false; unsigned long lastLedToggle 0; const int LED_MAX_FREQ_HZ 10; // 安全边界最大闪烁频率 10Hz const unsigned long MIN_LED_INTERVAL_MS 1000 / LED_MAX_FREQ_HZ; // ---------- 生成 MHS 参考文件 (Reference File) ---------- String generateReferenceFile() { StaticJsonDocument1024 doc; // 设备元信息 doc[mhs_version] MHS_VERSION; doc[device_id] DEVICE_ID; doc[device_name] DEVICE_NAME; doc[device_type] environmental_sensor; // 可读参数 (read) JsonArray readable doc.createNestedArray(readable); JsonObject temp readable.createNestedObject(); temp[name] temperature; temp[unit] celsius; temp[description] Current ambient temperature; temp[range_min] 0; temp[range_max] 50; JsonObject hum readable.createNestedObject(); hum[name] humidity; hum[unit] percent; hum[description] Current relative humidity; hum[range_min] 20; hum[range_max] 90; // 可写参数 (write) JsonArray writable doc.createNestedArray(writable); JsonObject led writable.createNestedObject(); led[name] led_state; led[type] boolean; led[description] Control onboard LED (on/off); led[allowed_values] [true, false]; // 安全边界 (Safety Limits) JsonObject safety doc.createNestedObject(safety_limits); safety[led_max_frequency_hz] LED_MAX_FREQ_HZ; safety[description] LED cannot be toggled faster than 10Hz to prevent hardware stress; String output; serializeJson(doc, output); return output; } // ---------- MHS read 处理 ---------- String handleRead(const String param) { StaticJsonDocument256 response; response[status] success; response[parameter] param; if (param temperature) { float t dht.readTemperature(); if (isnan(t)) { response[status] error; response[message] Failed to read temperature sensor; } else { response[value] t; response[unit] celsius; } } else if (param humidity) { float h dht.readHumidity(); if (isnan(h)) { response[status] error; response[message] Failed to read humidity sensor; } else { response[value] h; response[unit] percent; } } else { response[status] error; response[message] Unknown readable parameter: param; } String output; serializeJson(response, output); return output; } // ---------- MHS write 处理 (含安全边界检查) ---------- String handleWrite(const String param, const String value) { StaticJsonDocument256 response; response[parameter] param; if (param led_state) { // 安全边界检查防止超频闪烁 unsigned long now millis(); if (now - lastLedToggle MIN_LED_INTERVAL_MS) { response[status] rejected; response[message] Safety limit: LED toggling too fast (max 10Hz); response[safety_limit_hz] LED_MAX_FREQ_HZ; String output; serializeJson(response, output); return output; } if (value true || value 1 || value on) { ledState true; digitalWrite(LED_PIN, HIGH); response[status] success; response[applied_value] true; } else if (value false || value 0 || value off) { ledState false; digitalWrite(LED_PIN, LOW); response[status] success; response[applied_value] false; } else { response[status] error; response[message] Invalid value for led_state. Use true/false; } lastLedToggle now; } else { response[status] error; response[message] Unknown writable parameter: param; } String output; serializeJson(response, output); return output; } // ---------- HTTP 路由 ---------- // GET /mhs/reference - 获取参考文件 void handleReference() { server.send(200, application/json, generateReferenceFile()); } // GET /mhs/read?paramxxx - 执行 read 操作 void handleReadRoute() { if (!server.hasArg(param)) { server.send(400, application/json, {\error\:\Missing param parameter\}); return; } String param server.arg(param); String result handleRead(param); server.send(200, application/json, result); } // POST /mhs/write - 执行 write 操作 // Body: {param:led_state,value:true} void handleWriteRoute() { if (!server.hasArg(plain)) { server.send(400, application/json, {\error\:\Missing request body\}); return; } StaticJsonDocument128 doc; DeserializationError error deserializeJson(doc, server.arg(plain)); if (error) { server.send(400, application/json, {\error\:\Invalid JSON\}); return; } if (!doc.containsKey(param) || !doc.containsKey(value)) { server.send(400, application/json, {\error\:\Missing param or value field\}); return; } String param doc[param].asString(); String value doc[value].asString(); String result handleWrite(param, value); server.send(200, application/json, result); } // GET /mhs/discover - 设备发现 void handleDiscover() { StaticJsonDocument256 doc; doc[device_id] DEVICE_ID; doc[device_name] DEVICE_NAME; doc[mhs_version] MHS_VERSION; doc[endpoints] /mhs/reference, /mhs/read, /mhs/write; doc[status] ready; String output; serializeJson(doc, output); server.send(200, application/json, output); } // ---------- 根路径 ---------- void handleRoot() { String html htmlbody h1ESP32-S3 MHS Driver/h1 pDevice: String(DEVICE_NAME) /p pMHS Version: String(MHS_VERSION) /p pEndpoints:/p ul lia href/mhs/discoverGET /mhs/discover/a - Device discovery/li lia href/mhs/referenceGET /mhs/reference/a - Reference file/li liGET /mhs/read?paramtemperature - Read temperature/li liGET /mhs/read?paramhumidity - Read humidity/li liPOST /mhs/write - Write led_state/li /ul /body/html; server.send(200, text/html, html); } // ---------- setup ---------- void setup() { Serial.begin(115200); // 初始化 GPIO pinMode(LED_PIN, OUTPUT); digitalWrite(LED_PIN, LOW); // 初始化 DHT dht.begin(); // 连接 WiFi WiFi.begin(ssid, password); Serial.print(Connecting to WiFi); while (WiFi.status() ! WL_CONNECTED) { delay(500); Serial.print(.); } Serial.println(\nWiFi connected!); Serial.print(ESP32 IP address: ); Serial.println(WiFi.localIP()); // 注册 HTTP 路由 server.on(/, handleRoot); server.on(/mhs/discover, handleDiscover); server.on(/mhs/reference, handleReference); server.on(/mhs/read, handleReadRoute); server.on(/mhs/write, HTTP_POST, handleWriteRoute); server.begin(); Serial.println(MHS HTTP server started); } // ---------- loop ---------- void loop() { server.handleClient(); delay(10); }5.2 MCP Server 端Python 代码MCP Server 作为 AI 智能体Claude和 ESP32 之间的桥梁将自然语言指令转换为 MHS 的 ​​read​​/​​write​​ 命令。# # 文件名: mhs_mcp_server.py # 功能: MCP Server连接 Claude 与 ESP32 MHS 驱动 # 依赖: pip install mcp httpx # import asyncio import json import httpx from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.types as types # ---------- 配置 ---------- ESP32_IP 192.168.1.100 # 替换为你的 ESP32 IP 地址 ESP32_BASE_URL fhttp://{ESP32_IP} # ---------- 初始化 MCP Server ---------- server Server(mhs-esp32-server) # ---------- 工具定义 ---------- server.list_tools() async def handle_list_tools() - list[types.Tool]: 向 Claude 声明可用的工具 return [ types.Tool( nameread_sensor, descriptionRead environmental data from the ESP32 sensor, inputSchema{ type: object, properties: { parameter: { type: string, enum: [temperature, humidity], description: The sensor parameter to read } }, required: [parameter] } ), types.Tool( namecontrol_led, descriptionControl the onboard LED of ESP32, inputSchema{ type: object, properties: { state: { type: string, enum: [on, off], description: LED state } }, required: [state] } ), types.Tool( nameget_device_info, descriptionGet device information and reference file, inputSchema{ type: object, properties: {} } ) ] # ---------- 工具执行 ---------- server.call_tool() async def handle_call_tool( name: str, arguments: dict | None ) - list[types.TextContent | types.ImageContent | types.EmbeddedResource]: async with httpx.AsyncClient(timeout10.0) as client: # 工具1: 读取传感器 if name read_sensor: param arguments.get(parameter) if not param: return [types.TextContent(typetext, textError: Missing parameter)] try: response await client.get( f{ESP32_BASE_URL}/mhs/read, params{param: param} ) result response.json() return [types.TextContent( typetext, textfSensor reading: {json.dumps(result, indent2)} )] except Exception as e: return [types.TextContent(typetext, textfError: {str(e)})] # 工具2: 控制 LED elif name control_led: state arguments.get(state) if state not in [on, off]: return [types.TextContent(typetext, textError: State must be on or off)] value true if state on else false try: response await client.post( f{ESP32_BASE_URL}/mhs/write, json{param: led_state, value: value} ) result response.json() if result.get(status) rejected: return [types.TextContent( typetext, textf⚠️ Safety limit enforced: {result.get(message)} )] return [types.TextContent( typetext, textfLED turned {state}. Response: {json.dumps(result, indent2)} )] except Exception as e: return [types.TextContent(typetext, textfError: {str(e)})] # 工具3: 获取设备信息 elif name get_device_info: try: response await client.get(f{ESP32_BASE_URL}/mhs/discover) info response.json() # 同时获取参考文件 ref_response await client.get(f{ESP32_BASE_URL}/mhs/reference) ref ref_response.json() return [types.TextContent( typetext, textfDevice Info:\n{json.dumps(info, indent2)}\n\n fReference File:\n{json.dumps(ref, indent2)} )] except Exception as e: return [types.TextContent(typetext, textfError: {str(e)})] else: return [types.TextContent(typetext, textfUnknown tool: {name})] # ---------- 启动 Server ---------- async def main(): async with server.run_stdio(): await asyncio.Future() if __name__ __main__: asyncio.run(main())5.3 测试脚本验证 MHS 接口在连接 Claude 之前先用这个脚本验证 ESP32 的 MHS 接口是否正常工作# # 文件名: test_mhs_esp32.py # 功能: 测试 ESP32 MHS 接口 # import httpx import json ESP32_IP 192.168.1.100 # 替换为你的 ESP32 IP def test_mhs(): base fhttp://{ESP32_IP} with httpx.Client(timeout10.0) as client: # 1. 设备发现 print( * 50) print(1. Device Discovery) resp client.get(f{base}/mhs/discover) print(json.dumps(resp.json(), indent2)) # 2. 获取参考文件 print(\n * 50) print(2. Reference File) resp client.get(f{base}/mhs/reference) print(json.dumps(resp.json(), indent2)) # 3. 读取温度 print(\n * 50) print(3. Read Temperature) resp client.get(f{base}/mhs/read, params{param: temperature}) print(json.dumps(resp.json(), indent2)) # 4. 读取湿度 print(\n * 50) print(4. Read Humidity) resp client.get(f{base}/mhs/read, params{param: humidity}) print(json.dumps(resp.json(), indent2)) # 5. 控制 LED - 开 print(\n * 50) print(5. Write LED: ON) resp client.post(f{base}/mhs/write, json{param: led_state, value: true}) print(json.dumps(resp.json(), indent2)) # 6. 控制 LED - 关 print(\n * 50) print(6. Write LED: OFF) resp client.post(f{base}/mhs/write, json{param: led_state, value: false}) print(json.dumps(resp.json(), indent2)) # 7. 测试安全边界快速连续切换 LED print(\n * 50) print(7. Safety Limit Test (rapid toggling)) for i in range(3): resp client.post(f{base}/mhs/write, json{param: led_state, value: true}) result resp.json() print(f Toggle {i1}: {result.get(status, unknown)}) if result.get(status) rejected: print(f Reason: {result.get(message)}) if __name__ __main__: test_mhs()六、运行步骤步骤 1烧录 ESP32 固件安装 ​​Arduino IDE​​ 或 ​​PlatformIO​​安装 ESP32 开发板支持包Arduino IDE 中搜索 ​​esp32​​ 安装将上面的 Arduino 代码复制到新工程修改 WiFi 名称和密码选择开发板​​ESP32S3 Dev Module​​编译并烧录到 ESP32-S3打开串口监视器115200 baud记录 ESP32 的 IP 地址步骤 2安装 MCP Server 依赖pip install mcp httpx步骤 3配置并启动 MCP Server修改 Python 代码中的 ​​ESP32_IP​​ 为实际 IP 地址然后启动python mhs_mcp_server.py步骤 4在 Claude Desktop 中配置 MCP在 Claude Desktop 的配置文件中添加{ mcpServers: { mhs-esp32: { command: python, args: [/path/to/mhs_mcp_server.py] } } }重启 Claude Desktop 后就可以用自然语言控制 ESP32 了Read the current temperature from the ESP32Turn on the LEDGet device information七、运行效果预览7.1 ESP32 串口输出Connecting to WiFi..... WiFi connected! ESP32 IP address: 192.168.1.100 MHS HTTP server started7.2 测试脚本输出节选1. Device Discovery { device_id: mhs-esp32-s3-001, device_name: ESP32-S3 Environmental Monitor, mhs_version: research-preview-v1.0, endpoints: /mhs/reference, /mhs/read, /mhs/write, status: ready } 3. Read Temperature { status: success, parameter: temperature, value: 25.3, unit: celsius } 7. Safety Limit Test (rapid toggling) Toggle 1: success Toggle 2: rejected Reason: Safety limit: LED toggling too fast (max 10Hz)八、总结通过这个实战项目我们完成了目标实现状态确定硬件版本✅ ESP32-S3确定 MHS 协议版本✅ Research Preview v1.0实现 MHS 标准化驱动✅ ​​read​​​/​​write​​ 原语生成参考文件✅ JSON 格式 Reference File安全边界检查✅ LED 最大频率限制设备自动发现✅ ​​/mhs/discover​​ 端点MCP 集成✅ Python MCP Server可运行的小场景✅ 温湿度读取 LED 控制虽然 MHS 尚未正式开源但这个实现已经覆盖了官方公布的核心设计理念标准化驱动ESP32 固件作为 MHS 驱动统一了硬件接口read/write 原语所有操作都通过这两个基本命令完成参考文件AI 可以“读懂”设备的能力和限制安全护栏驱动层强制执行安全边界下一步你可以扩展更多传感器和执行器实现多设备协同多个 ESP32 通过 MESH 组网等待 MHS 正式开源后迁移到官方 SDK重要提醒本文实现的是基于公开设计理念的模拟实现并非 Anthropic 官方 MHS SDK。正式版 MHS 开源后API 和规范可能会有调整。建议关注 ​​Anthropic 官方公告​​ 获取最新进展。
返回列表