☰
【MHS协议】第四章:ESP32 接入 MHS 协议实战:从零构建一个 MHS 兼容设备
2026/10/23 10:16:56 网站建设 项目流程
一、写在前面: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® LX7,240 MHz)

开发板

ESP32-S3-DevKitC-1(或任意 ESP32-S3 开发板)

板载外设

1 个板载 LED(GPIO2)、1 个按键(BOOT,GPIO0)

可选外设

DHT11 温湿度传感器(GPIO4)

为什么选 ESP32-S3?ESP32-S3 支持 Wi-Fi 和 BLE,具备足够的计算能力和内存来运行 MHS 驱动逻辑和 MCP 通信栈。乐鑫的 ESP32 系列已被列为 MHS 的潜在测试平台之一。

MHS 协议版本

项目

说明

协议名称

Model Hardware Standard (MHS)

当前版本

Research Preview v1.0(2026年8月27日发布)

协议状态

研究预览版,尚未开源

核心原语

​​read​​​、​​write​​

控制通道

MCP、CLI、代码文件(API)


三、实战场景:AI 通过 MHS 读取 ESP32 的环境数据

我们要实现的场景是:

用户通过 Claude(或其他 AI 智能体)向 ESP32 发送自然语言指令,ESP32 返回当前温湿度数据,并可控地点亮/熄灭板载 LED。

这模拟了 MHS 协议的核心流程:

  1. 设备发现:ESP32 通过 MHS 驱动向网络宣告自己的存在
  2. 设备描述:ESP32 提供参考文件(Reference File),描述自己能做什么
  3. AI 控制:AI 通过 ​​read​​ 读取传感器数据,通过 ​​write​​ 控制 LED
  4. 安全边界:驱动层阻止超出安全范围的指令

四、项目架构
┌─────────────────────────────────────────────────────────────┐ │ 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() { StaticJsonDocument<1024> 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) { StaticJsonDocument<256> 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) { StaticJsonDocument<256> 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?param=xxx - 执行 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; } StaticJsonDocument<128> 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"].as<String>(); String value = doc["value"].as<String>(); String result = handleWrite(param, value); server.send(200, "application/json", result); } // GET /mhs/discover - 设备发现 void handleDiscover() { StaticJsonDocument<256> 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 = "<html><body>" "<h1>ESP32-S3 MHS Driver</h1>" "<p>Device: " + String(DEVICE_NAME) + "</p>" "<p>MHS Version: " + String(MHS_VERSION) + "</p>" "<p>Endpoints:</p>" "<ul>" "<li><a href='/mhs/discover'>GET /mhs/discover</a> - Device discovery</li>" "<li><a href='/mhs/reference'>GET /mhs/reference</a> - Reference file</li>" "<li>GET /mhs/read?param=temperature - Read temperature</li>" "<li>GET /mhs/read?param=humidity - Read humidity</li>" "<li>POST /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 = f"http://{ESP32_IP}" # ---------- 初始化 MCP Server ---------- server = Server("mhs-esp32-server") # ---------- 工具定义 ---------- @server.list_tools() async def handle_list_tools() -> list[types.Tool]: """向 Claude 声明可用的工具""" return [ types.Tool( name="read_sensor", description="Read 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( name="control_led", description="Control the onboard LED of ESP32", inputSchema={ "type": "object", "properties": { "state": { "type": "string", "enum": ["on", "off"], "description": "LED state" } }, "required": ["state"] } ), types.Tool( name="get_device_info", description="Get 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(timeout=10.0) as client: # 工具1: 读取传感器 if name == "read_sensor": param = arguments.get("parameter") if not param: return [types.TextContent(type="text", text="Error: Missing 'parameter'")] try: response = await client.get( f"{ESP32_BASE_URL}/mhs/read", params={"param": param} ) result = response.json() return [types.TextContent( type="text", text=f"Sensor reading: {json.dumps(result, indent=2)}" )] except Exception as e: return [types.TextContent(type="text", text=f"Error: {str(e)}")] # 工具2: 控制 LED elif name == "control_led": state = arguments.get("state") if state not in ["on", "off"]: return [types.TextContent(type="text", text="Error: 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( type="text", text=f"⚠️ Safety limit enforced: {result.get('message')}" )] return [types.TextContent( type="text", text=f"LED turned {state}. Response: {json.dumps(result, indent=2)}" )] except Exception as e: return [types.TextContent(type="text", text=f"Error: {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( type="text", text=f"Device Info:\n{json.dumps(info, indent=2)}\n\n" f"Reference File:\n{json.dumps(ref, indent=2)}" )] except Exception as e: return [types.TextContent(type="text", text=f"Error: {str(e)}")] else: return [types.TextContent(type="text", text=f"Unknown 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 = f"http://{ESP32_IP}" with httpx.Client(timeout=10.0) as client: # 1. 设备发现 print("=" * 50) print("1. Device Discovery") resp = client.get(f"{base}/mhs/discover") print(json.dumps(resp.json(), indent=2)) # 2. 获取参考文件 print("\n" + "=" * 50) print("2. Reference File") resp = client.get(f"{base}/mhs/reference") print(json.dumps(resp.json(), indent=2)) # 3. 读取温度 print("\n" + "=" * 50) print("3. Read Temperature") resp = client.get(f"{base}/mhs/read", params={"param": "temperature"}) print(json.dumps(resp.json(), indent=2)) # 4. 读取湿度 print("\n" + "=" * 50) print("4. Read Humidity") resp = client.get(f"{base}/mhs/read", params={"param": "humidity"}) print(json.dumps(resp.json(), indent=2)) # 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(), indent=2)) # 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(), indent=2)) # 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 {i+1}: {result.get('status', 'unknown')}") if result.get("status") == "rejected": print(f" Reason: {result.get('message')}") if __name__ == "__main__": test_mhs()

六、运行步骤
步骤 1:烧录 ESP32 固件
  1. 安装 ​​Arduino IDE​​ 或 ​​PlatformIO​​
  2. 安装 ESP32 开发板支持包(Arduino IDE 中搜索 ​​esp32​​ 安装)
  3. 将上面的 Arduino 代码复制到新工程,修改 WiFi 名称和密码
  4. 选择开发板:​​ESP32S3 Dev Module​​
  5. 编译并烧录到 ESP32-S3
  6. 打开串口监视器(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 ESP32"
  • "Turn on the LED"
  • "Get device information"

七、运行效果预览
7.1 ESP32 串口输出
Connecting to WiFi..... WiFi connected! ESP32 IP address: 192.168.1.100 MHS HTTP server started
7.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 可以“读懂”设备的能力和限制
  • 安全护栏:驱动层强制执行安全边界

下一步,你可以:

  1. 扩展更多传感器和执行器
  2. 实现多设备协同(多个 ESP32 通过 MESH 组网)
  3. 等待 MHS 正式开源后,迁移到官方 SDK

重要提醒:本文实现的是基于公开设计理念的模拟实现,并非 Anthropic 官方 MHS SDK。正式版 MHS 开源后,API 和规范可能会有调整。建议关注 ​​Anthropic 官方公告​​ 获取最新进展。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询