☰
esp-adf 实战:esp_service PC 端 MCP 客户端脚本全解析——用 HTTP / SSE / WebSocket / UART 四种传输驱动真实设备
2026/10/14 3:16:54 网站建设 项目流程
  • 音视频
  • 音频处理
  • 嵌入式
  • 物联网
  • 语音

【免费下载链接】esp-adf

Espressif Advanced Development Framework for Multimedia Applications

项目地址:https://gitcode.com/gh_mirrors/es/esp-adf
点击查看免费下载

本指南围绕 esp-adf 开源仓库中 components/esp_service/test_apps/scripts/README.md 定位的四个 PC 端 MCP(Model Context Protocol)客户端脚本展开:test_mcp_http.py、test_mcp_sse.py、test_mcp_ws.py、test_mcp_uart.py。它们面向真实 DUT(Device Under Test)设备,通过 JSON-RPC 2.0 协议与板上esp_serviceMCP 服务器通信,逐一验证tools/list、tools/call及各类异常分支。读完本文,你将掌握如何在 PC 上构建、烧录带 MCP 传输的固件,正确配置 Kconfig 开关,并熟练运行这四类脚本完成对设备的传输层烟测与功能回归。

脚本在整个测试体系中的定位

在 esp_service 组件的测试布局中,scripts/目录只是冰山一角,它的上位说明文档 components/esp_service/test_apps/README.md 给出了完整脉络:test_apps/是框架级测试与仿真平台,专门针对 service manager、MCP JSON-RPC、各传输层以及 PC 端仿真进行压力测试;而配套的 components/esp_service/examples/mock_services/ 只演示六个 mock service 的“用法”,两者通过EXTRA_COMPONENT_DIRS保持同步,示例与测试始终使用同一份服务实现。

test_apps/的整体布局如下:

test_apps/ ├── CMakeLists.txt # IDF on-device project ├── sdkconfig.defaults ├── pytest_esp_service_test_apps.py ├── main/ # IDF on-device runner (Unity + optional transport smoke) │ ├── app_main.c │ ├── ut_esp_service_core.c │ ├── ut_service_manager.c │ ├── ut_*_transport.c # gated by Kconfig (见 CMakeLists.txt) │ ├── ut_transport_lifecycle_test.c │ ├── Kconfig.projbuild │ └── CMakeLists.txt ├── scripts/ # PC-side MCP JSON-RPC clients against a real DUT │ ├── README.md # 本文关联文档 │ ├── test_mcp_http.py # Plain HTTP POST /mcp │ ├── test_mcp_sse.py # HTTP+SSE (搭配 ESP_SVC_TEST_SSE_HOLD_AFTER_SMOKE) │ ├── test_mcp_ws.py # WebSocket (搭配 ESP_SVC_TEST_WS_HOLD_AFTER_SMOKE) │ └── test_mcp_uart.py # UART (搭配 ESP_SVC_TEST_UART_HOLD_AFTER_SMOKE) └── pc/ # PC / host runner (FreeRTOS POSIX port, native CMake)

四份脚本与main/下的ut_http_transport.c、ut_sse_transport.c、ut_ws_transport.c、ut_uart_transport.c一一对应:板上 Unity 测试先在设备内跑完传输层烟测,之后 PC 脚本再对仍在运行的服务器发起真实网络/串口请求,形成“设备内自测 + 设备外联测”的双层验证闭环。集成入口见 main/app_main.c,它会依次执行 Unity 用例与各 Kconfig 门控的传输测试。

前置条件:构建带 MCP 传输的固件

脚本运行的前提是设备上已经拉起 MCP 服务器。构建流程与普通 IDF 项目一致(从test_apps/目录执行):

cd test_apps idf.py set-target esp32s3 idf.py menuconfig # 设置 Wi-Fi,在 "ESP Service test_apps Configuration" 下切换传输开关 idf.py build flash monitor

默认配置见 test_apps/sdkconfig.defaults:四个 MCP 传输(CONFIG_ESP_MCP_TRANSPORT_HTTP/SSE/WS/UART)全部使能,但ESP_SVC_TEST_ENABLE_*_TRANSPORT各设备内烟测默认关闭,因此默认固件只跑 Unity 的[esp_service]核心与 manager 测试;CONFIG_HTTPD_WS_SUPPORT=y为 WebSocket 传输提供底层支持,控制台默认走 UART0(CP2102 / ttyUSB0)。

Kconfig 传输开关详解

所有测试开关定义在 main/Kconfig.projbuild,归纳如下:

配置项默认值说明
ESP_SVC_TEST_WIFI_SSIDmyssid网络传输(HTTP/SSE/WS)烟测所需的 STA SSID
ESP_SVC_TEST_WIFI_PASSWORDmypasswordWPA/WPA2 密码
ESP_SVC_TEST_ENABLE_LIFECYCLE_STRESSy传输生命周期压力测试(create/start/stop/destroy 反复循环)
ESP_SVC_TEST_LIFECYCLE_STRESS_CYCLES100(范围 1–1000)每个传输的循环次数
ESP_SVC_TEST_ENABLE_HTTP_TRANSPORTn启用板上 HTTP 烟测(依赖ESP_MCP_TRANSPORT_HTTP)
ESP_SVC_TEST_ENABLE_SSE_TRANSPORTn启用板上 SSE 烟测(依赖ESP_MCP_TRANSPORT_SSE)
ESP_SVC_TEST_SSE_HOLD_AFTER_SMOKEn烟测后保持SSE MCP 服务器运行,供 PC 脚本连接
ESP_SVC_TEST_ENABLE_WS_TRANSPORTn启用板上 WS 烟测(依赖ESP_MCP_TRANSPORT_WS)
ESP_SVC_TEST_WS_HOLD_AFTER_SMOKEn烟测后保持 WS 服务器运行
ESP_SVC_TEST_ENABLE_UART_TRANSPORTn启用板上 UART 烟测(依赖ESP_MCP_TRANSPORT_UART)
ESP_SVC_TEST_UART_HOLD_AFTER_SMOKEn烟测后保持 UART 传输运行
ESP_SVC_TEST_UART_TX_PIN43UART TX GPIO(ESP32-S3 U0TXD)
ESP_SVC_TEST_UART_RX_PIN44UART RX GPIO(ESP32-S3 U0RXD)
ESP_SVC_TEST_UART_BAUD_RATE115200UART MCP 波特率

关键前提:要让 PC 脚本能够连接设备,必须同时打开对应传输的Enable与Hold After Smoke两个开关——前者让板上 Unity 烟测跑起来,后者保证烟测结束后服务器不被销毁(见 ut_sse_transport.c 中HOLD_AFTER_SMOKE分支会vTaskDelay保持服务器存活)。此外,HTTP / SSE / WS 三个网络传输一次只应启用其中一个演示,避免端口与资源冲突;Wi-Fi 凭据也需要配置为实际可用的 AP。

运行 PC 脚本:命令、参数与依赖

固件启动、匹配的 “hold after smoke” 开关打开后,即可在工作站上驱动设备。官方要求在test_apps/目录下运行,保证导入与相对路径稳定:

cd test_apps python3 scripts/test_mcp_http.py <device-ip> python3 scripts/test_mcp_sse.py <device-ip> [http-port] python3 scripts/test_mcp_ws.py <device-ip> [port] python3 scripts/test_mcp_uart.py <serial-port> [baud]

实测示例:

python3 scripts/test_mcp_http.py 192.168.1.100 python3 scripts/test_mcp_sse.py 192.168.1.100 8080 python3 scripts/test_mcp_ws.py 192.168.1.100 8081 python3 scripts/test_mcp_uart.py /dev/ttyUSB0 115200

Python 依赖按脚本分装:pip install requests(HTTP、SSE 客户端)、pip install websockets(WS 客户端)、pip install pyserial(UART 客户端)。各脚本在 import 失败时都会给出安装提示并优雅退出(如 test_mcp_sse.py)。

四类传输客户端逐层剖析

1. HTTP:test_mcp_http.py——最直接的同步 JSON-RPC

test_mcp_http.py 用同步requests.post向http://<device-ip>:8080/mcp发送 JSON-RPC 2.0 请求,核心是MCPClient类:

  • _next_id()自增请求 ID,保证每帧可区分;
  • _call(method, params)组包{'jsonrpc': '2.0', 'id': ..., 'method': ...},携带Content-Type: application/json,5 秒超时,失败统一捕获requests.exceptions.RequestException;
  • list_tools()封装tools/list,call_tool(name, arguments)封装tools/call。

脚本共编排 18 个用例,完整覆盖四类 mock service 的工具面:

分组用例
Playerplayer_service_play、player_service_set_volume(volume: 75)、player_service_get_volume、player_service_get_status
LEDled_service_blink(period_ms: 1000)、led_service_get_state、led_service_set_brightness(brightness: 80)
WiFiwifi_service_get_status、wifi_service_get_ssid、wifi_service_get_rssi、wifi_service_reconnect(切到GuestNetwork)并再次查询 SSID 验证生效
OTAota_service_get_version、ota_service_check_update、ota_service_get_progress
异常调用不存在的工具invalid_tool_name、调用非法方法invalid/method

它把每个请求与响应都缩进打印,便于对照 JSON-RPC 帧格式,是最适合初次上手、肉眼核对协议的脚本。

2. SSE:test_mcp_sse.py——MCP 推荐的流式传输

test_mcp_sse.py 实现 MCP 2024-11-05 风格的 HTTP+SSE 会话,通信分两路:

  1. 上行 GET 长连接:GET /mcp/sse建立持久 SSE 流,等待服务器推送必选的endpoint事件,其data中包含带sessionId的 POST 地址(POST /mcp/messages?sessionId=<id>);
  2. 下行 POST 提交:JSON-RPC 请求经requests.post发往该地址,服务器返回HTTP 202 Accepted,真正的响应随后以 SSEmessage事件形式在流中推送,客户端按请求id匹配。

MCPSseSession用threading.Thread跑_reader_loop逐行解析 SSE 帧(event:/data:前缀),用queue.Queue把message事件回传主线程,rpc()轮询队列直到拿到匹配id的响应或超时。内置三个判定用例:

  • tools/list:返回的工具列表中必须含player_service前缀工具;
  • tools/call player_service_get_volume:result.isError必须为假;
  • invalid/method:必须返回 JSON-RPC 错误,且error.code == -32601(Method not found)。

启动时须启用ESP_SVC_TEST_ENABLE_SSE_TRANSPORT与ESP_SVC_TEST_SSE_HOLD_AFTER_SMOKE,保证服务器在板上烟测后继续存活(默认地址http://<ip>:8080/mcp/sse)。这也是 MCP 官方推荐形态,适合验证带会话状态的服务端实现。

3. WebSocket:test_mcp_ws.py——最全面的功能回归套件

test_mcp_ws.py 基于asyncio+websockets库连接ws://<device-ip>:8081/mcp,是全套脚本中用例最丰富的一份,共 10 组功能测试:

测试验证点
Connection建立/断开连接,新连接上立即可用
List Toolstools/list返回非空工具表
Tool Invocation依次调用 player 4 个工具、LED 3 个工具并断言result存在
Persistent Connection同一连接上连续 5 次tools/list全部成功
Error Handling不存在的工具、非法方法、畸形 JSON{invalid json!!!、缺method字段四种异常输入
Rapid Requests单连接 10 次连续player_service_get_volume全部成功并统计平均耗时
Concurrent Clients3 个客户端并行各自发 3 次请求,全部通过
Reconnection断开 0.5s 后重连仍可正常tools/list
JSON-RPC Compliance响应含jsonrpc: "2.0"与id、响应id与请求一致、无id的 notification 不返回响应
Large Payload带 2000 字符 padding 的大参数请求可被处理(服务端max_request_size默认 4096)

它同样依赖ESP_SVC_TEST_ENABLE_WS_TRANSPORT+ESP_SVC_TEST_WS_HOLD_AFTER_SMOKE。注意板上 WS 传输支持并发客户端(MCP_WS_MAX_CLIENTS上限 8,见 esp_service_mcp_trans_ws.c),并发用例正验证了这一点。

4. UART:test_mcp_uart.py——无网络栈的行式 JSON 传输

test_mcp_uart.py 面向无网络环境,通过串口传递 MCP 协议,帧格式约定为每行一条 JSON-RPC(发送以\n结尾,接收逐行读取)。核心要点:

  • 串口参数:serial.Serial默认 8 数据位、无校验、1 停止位,超时 3 秒;波特率默认115200;
  • 序列化使用json.dumps(request, separators=(',', ':'))压缩成单行,避免换行破坏帧边界;
  • 接收端过滤设备日志:只认以{开头的行作为 JSON 响应,其余视为设备日志行跳过;
  • 连接前reset_input_buffer()/reset_output_buffer()并排空残留数据。

8 组用例覆盖连接、工具列表、工具调用(含 LED on/off)、连续多请求、异常输入、10 次快速请求吞吐统计、JSON-RPC 合规(id匹配)、断线重连。运行前需打开ESP_SVC_TEST_ENABLE_UART_TRANSPORT+ESP_SVC_TEST_UART_HOLD_AFTER_SMOKE。

硬件接线(脚本内亦有提示):ESP32-S3 默认走 UART0 的 GPIO43(U0TXD)→ 适配器 RX、GPIO44(U0RXD)← 适配器 TX、GND 共地;也可在 menuconfig 中通过ESP_SVC_TEST_UART_TX_PIN/ESP_SVC_TEST_UART_RX_PIN改到其他引脚。板上烟测则使用UART_NUM_0与 Kconfig 配置的引脚波特率,见 ut_uart_transport.c。

服务端默认配置对照(源码佐证)

脚本里的端口、路径并非魔法数字,全部对应各传输的默认配置宏:

传输默认配置源码位置
HTTPport=8080、uri_path="/mcp"、max_request_size=4096、stack_size=4096、max_uri_handlers=8include/esp_service_mcp_trans_http.h
SSEport=8080、sse_path="/mcp/sse"、messages_path="/mcp/messages"、max_request_size=4096、stack_size=8192include/esp_service_mcp_trans_sse.h
WebSocketport=8081、uri_path="/mcp"、max_request_size=4096、max_clients=4(代码中 0 会被重置为 4)include/esp_service_mcp_trans_ws.h
UARTuart_port=UART_NUM_1、baud_rate=115200、TX/RX 引脚默认不修改include/esp_service_mcp_trans_uart.h

从传输层实现可以印证脚本行为:HTTP 传输在 esp_service_mcp_trans_http.c 中默认 URI 为/mcp,非application/json请求返回 415、超长请求返回 413;SSE 传输用 ESP-IDF ≥ 5.1 的httpd_req_async_handler_begin()挂起 GET 连接以推送事件(esp_service_mcp_trans_sse.c)。所有*_create()都遵循“传入 NULL 配置则套用默认宏”的惯例(如 esp_service_mcp_trans_http.c),应用层可按需覆盖端口与路径。

结果解读与 CI 集成

每个脚本都以可读的 PASS/FAIL 汇总收尾:

  • HTTP 脚本打印全部 18 个请求/响应后输出All tests completed!;
  • SSE 脚本输出[PASS]/[FAIL]清单与Total / Failed统计,任一失败即以非零码退出(sys.exit(0 if failed == 0 else 1));
  • WS / UART 脚本类似地给出分组汇总与总体判定。

在 CI 场景中,pytest_esp_service_test_apps.py 通过dut.expect(r'ESP Service - test_apps')与TEST_APPS RUN COMPLETE断言设备端跑完整轮测试;PC 侧脚本则可作为回归阶段的可选联测环节,验证传输层在真实设备上的互操作性。排查问题时,建议按“菜单配置 → 固件日志(确认烟测与 hold 生效)→ 网络/串口连通性 → 脚本输出”的顺序逐层定位。

相关资源

  • 上位说明(含 PC 端仿真与全部运行方式):components/esp_service/test_apps/README.md
  • 六个 mock service 用法示例:components/esp_service/examples/mock_services/README.md
  • esp_service 组件架构与 MCP 服务器说明:components/esp_service/README.md
  • 与mcp_http_server搭配的 LLM 桥接工具:tools/mcp_llm_bridge/README.md
  • 音视频
  • 音频处理
  • 嵌入式
  • 物联网
  • 语音

【免费下载链接】esp-adf

Espressif Advanced Development Framework for Multimedia Applications

项目地址:https://gitcode.com/gh_mirrors/es/esp-adf
点击查看免费下载

相关推荐

上一篇:FAST 动画体系 AnimateGroup 构造函数详解:批量并行控制多个 AnimateTo/AnimateFrom 动画
下一篇:《投资入门指南》如何看懂10-K年报?美股财报实战阅读指南(附新手清单)

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

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

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

立即咨询