- 音视频
- 音频处理
- 嵌入式
- 物联网
- 语音
【免费下载链接】esp-adf
Espressif Advanced Development Framework for Multimedia Applications
本指南围绕 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_SSID | myssid | 网络传输(HTTP/SSE/WS)烟测所需的 STA SSID |
ESP_SVC_TEST_WIFI_PASSWORD | mypassword | WPA/WPA2 密码 |
ESP_SVC_TEST_ENABLE_LIFECYCLE_STRESS | y | 传输生命周期压力测试(create/start/stop/destroy 反复循环) |
ESP_SVC_TEST_LIFECYCLE_STRESS_CYCLES | 100(范围 1–1000) | 每个传输的循环次数 |
ESP_SVC_TEST_ENABLE_HTTP_TRANSPORT | n | 启用板上 HTTP 烟测(依赖ESP_MCP_TRANSPORT_HTTP) |
ESP_SVC_TEST_ENABLE_SSE_TRANSPORT | n | 启用板上 SSE 烟测(依赖ESP_MCP_TRANSPORT_SSE) |
ESP_SVC_TEST_SSE_HOLD_AFTER_SMOKE | n | 烟测后保持SSE MCP 服务器运行,供 PC 脚本连接 |
ESP_SVC_TEST_ENABLE_WS_TRANSPORT | n | 启用板上 WS 烟测(依赖ESP_MCP_TRANSPORT_WS) |
ESP_SVC_TEST_WS_HOLD_AFTER_SMOKE | n | 烟测后保持 WS 服务器运行 |
ESP_SVC_TEST_ENABLE_UART_TRANSPORT | n | 启用板上 UART 烟测(依赖ESP_MCP_TRANSPORT_UART) |
ESP_SVC_TEST_UART_HOLD_AFTER_SMOKE | n | 烟测后保持 UART 传输运行 |
ESP_SVC_TEST_UART_TX_PIN | 43 | UART TX GPIO(ESP32-S3 U0TXD) |
ESP_SVC_TEST_UART_RX_PIN | 44 | UART RX GPIO(ESP32-S3 U0RXD) |
ESP_SVC_TEST_UART_BAUD_RATE | 115200 | UART 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 115200Python 依赖按脚本分装: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 的工具面:
| 分组 | 用例 |
|---|---|
| Player | player_service_play、player_service_set_volume(volume: 75)、player_service_get_volume、player_service_get_status |
| LED | led_service_blink(period_ms: 1000)、led_service_get_state、led_service_set_brightness(brightness: 80) |
| WiFi | wifi_service_get_status、wifi_service_get_ssid、wifi_service_get_rssi、wifi_service_reconnect(切到GuestNetwork)并再次查询 SSID 验证生效 |
| OTA | ota_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 会话,通信分两路:
- 上行 GET 长连接:
GET /mcp/sse建立持久 SSE 流,等待服务器推送必选的endpoint事件,其data中包含带sessionId的 POST 地址(POST /mcp/messages?sessionId=<id>); - 下行 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 Tools | tools/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 Clients | 3 个客户端并行各自发 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。
服务端默认配置对照(源码佐证)
脚本里的端口、路径并非魔法数字,全部对应各传输的默认配置宏:
| 传输 | 默认配置 | 源码位置 |
|---|---|---|
| HTTP | port=8080、uri_path="/mcp"、max_request_size=4096、stack_size=4096、max_uri_handlers=8 | include/esp_service_mcp_trans_http.h |
| SSE | port=8080、sse_path="/mcp/sse"、messages_path="/mcp/messages"、max_request_size=4096、stack_size=8192 | include/esp_service_mcp_trans_sse.h |
| WebSocket | port=8081、uri_path="/mcp"、max_request_size=4096、max_clients=4(代码中 0 会被重置为 4) | include/esp_service_mcp_trans_ws.h |
| UART | uart_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
相关推荐
AgentScope MCP客户端:HTTP、SSE、StdIO传输
AgentScope MCP客户端:HTTP、SSE、StdIO传输 引言:为什么需要多协议MCP客户端? 在现代AI代理系统中,模型上下文协议(Model C
人工智能大模型AI Agent多智能体Agent 框架Agent 编排工具调用RAGMCP 服务语音esp_service 测试与仿真框架实战:从 IDF 设备端 Unity 测试到 PC 主机端 MCP 客户端验证
esp_service 测试与仿真框架实战:从 IDF 设备端 Unity 测试到 PC 主机端 MCP 客户端验证 本文是 esp adf 中 esp_ser
音视频音频处理嵌入式物联网语音MCP Python SDK 客户端传输层全解:Streamable HTTP、stdio、内存直连与 SSE
MCP Python SDK 客户端传输层全解:Streamable HTTP、stdio、内存直连与 SSE 本篇文章围绕官方 Python SDK(pyth
人工智能MCP 服务MCP Clients
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考