rm_mcp: ROS Model Context Protocol Server for RoboMaster Workspace

English | 中文

English

1. Project Background

In complex robotics development like RoboMaster (RM), debugging controllers, tuning PID parameters, and inspecting system states traditionally require engineers to manually execute CLI commands, launch rqt/PlotJuggler GUIs, and inspect rosbags.

rm_mcp bridges ROS Noetic and ROS 2 (Humble/Jazzy) with Model Context Protocol (MCP), providing AI Agents (such as Claude, Cursor, and Antigravity) with structured "eyes" (state & system inspection) and "hands" (controller actuation, signal injection, rosbag recording, and automated PID response analysis).

Built specifically for the rm_ws workspace, rm_mcp solves critical real-world robotics debugging challenges:

• Cross-Platform Readiness (ROS 1 + ROS 2): Uses a pure Rosbridge WebSocket JSON-RPC architecture (ws://host:9090), ensuring 100% zero-code-change portability when migrating from ROS 1 Noetic to ROS 2.

• Bandwidth Bottlenecks: Prevents high-frequency ROS topics (IMU, point clouds, joint states) from choking WiFi connections by processing and recording data on-vehicle before transfer.

• Safety Hazards: Prevents hallucinated or accidental execution of high-risk motor control actions on physical robots through a two-phase confirmation mechanism (confirm_token).

• Closed-Loop Automation: Automates the "Parameter Update → Signal Excitation → Bag Recording → PID Analysis" loop against strict domain thresholds (e.g., 0.006 rad gimbal tracking error).

2. Architecture & Security Tiers

rm_mcp implements a 4-tier Trust Ladder (L0 to L3) and a Two-Phase Handshake (confirm_token):

+-------------------------------------------------------------------------------+
|                            Trust Ladder Architecture                          |
+-------------------+-----------------------------------------------------------+
| Tier              | Description & Operations                                  |
+-------------------+-----------------------------------------------------------+
| L0: Read-Only     | Topic/node inspection, param get, preflight status checks  |
| L1: Sim Write     | Parameter tuning and controller testing in Gazebo         |

> **Sim-mode gating**: red-line keywords (`chassis`/`shooter`/`friction`/`balance`/`legged`) and inject amplitudes > 1.0 rad/s still require `confirm_token` in sim by design; every `PENDING_CONFIRMATION` response includes a `trigger_reason` explaining exactly why the gate fired.
| L2: Real Power-off| Vehicle status verification without motor output          |
| L3: Real Power-on | Physical robot actuation (launch, shooter, chassis)       |
+-------------------+-----------------------------------------------------------+

Two-Phase Handshake Mechanism

When an operation targets a physical robot (TARGET_MODE=real) or matches sensitive keywords (e.g., shooter, friction, chassis, balance, legged), rm_mcp intercepts the call:

1. Phase 1 (Pending Confirmation): Returns a structured request with a 6-character confirm_token (valid for 5 minutes) and a human-readable execution summary.

2. Phase 2 (Execution): The Agent prompts the user for approval and re-invokes the tool with the valid confirm_token

3. Key Features & Tools (25 MCP Tools)

rm_mcp exposes 25 production-ready MCP tools divided into three suites:

🌐 Cross-Platform Introspection Suite (Pure Rosbridge WebSocket)

All five service/echo tools self-heal: when rosbridge is unreachable they first try to pull it up via ensure_infrastructure("rosbridge_websocket") (rate-limited, 30s cooldown), retry once, and only then fail honestly — the ROS Master XML-RPC fallback is reserved for "rosbridge up, rosapi node missing".

1. get_topic_info(topic): Inspect topic type, publishers, and subscribers (answers "Who is publishing to this topic?").

2. get_node_info(node): Inspect active node details (published topics, subscribed topics, services).

3. list_topics(): List all active ROS topics and their message types.

4. list_nodes(): List all active ROS nodes in the system.

5. echo_topic(topic, duration_sec, max_messages): Sample ROS topic messages with safety throttling (max 5.0s duration). Returns JSON objects.

⚙️ Operations, Signal Injection & Bag Suite

6. preflight(): Inspect system prerequisite states without side effects (ROS Master socket, Systemd services, vehicle ROBOT_TYPE, robot_state_publisher TF readiness, gimbal controller TF-dependency chain — e.g. chassis_controller stopped removing the odom -> base_link TF — motor effort_limit vs PID gains, and safety_controller soft-limit stall detection).

7. launch(package, launch_file, args, confirm_token): Start ROS nodes — local roslaunch in sim mode, SSH to vehicle NUC on real targets. args accepts both list and dict formats. Preferred tuning bring-up: load_controllers.launch loads all controllers (it does NOT start them — follow with switch_controller), or manual.launch for remote-control sessions.

8. list_launches(): List running roslaunch / mon processes with handles (host:package:launch_file:PID).

9. kill_launch(handle, confirm_token): Terminate a process group by handle (SIGTERM -> SIGKILL progression and verification).

10. ensure_infrastructure(target_services, confirm_token): Ensure essential infrastructure services (rosapi, rosbridge_websocket, robot_state_publisher) are running.

11. set_param(param_name, value, confirm_token): Safe parameter writing (XML-RPC & rospy) with pre-write dry-run state capture and post-write re-read verification. Warns when the path is a controller runtime parameter that only set_dynamic_param can hot-reload.

12. inject_signal(topic, field, waveform, amplitude, frequency, duration_sec, publish_rate_hz, extra_fields, dry_run, force, confirm_token): Step, sine, square, or triangle waveform generator with exclusive publisher arbitration and quiet-state (0.0 odom reference) restoration. After extra_fields are applied, a zero/absent top-level stamp or header.stamp is auto-filled with rospy.Time.now() on every published message (quiet-state included; explicit nonzero stamps preserved), eliminating 0-timestamp artifacts in rosbag analysis.

13. record_bag(topics, duration_sec, mode, output_path, background, trigger_on_topic): Synchronous or asynchronous (background=True) recording. trigger_on_topic appends the named topic to the recorded topic list; rosbag record --duration starts counting once the first message arrives on any recorded topic, so use background=True and choose duration_sec long enough to cover inject latency + signal duration + margin.

14. stop_record_bag(bag_handle, handle, confirm_token): Stop a running background recording handle, clean up processes, retrieve remote bag files via scp, and return metadata inspection.

15. capture_topic_sync(topic, fields, duration_sec, max_messages, timeout_sec): Zero disk I/O in-memory ROS topic sampling via rospy.Subscriber for ultra-fast steady-state or external-input diagnostics.

16. capture_inject_sync(capture_topic, inject_topic, field, waveform, amplitude, frequency, duration_sec, publish_rate_hz, quiet_window_sec, fields, max_messages, extra_fields, dry_run, force, confirm_token): Preferred dynamic closed-loop path: waits for an observed capture callback, delegates injection security/confirmation/arbitration unchanged to inject_signal, and captures the full quiet window after injection completes. Readiness-only data, no injection-period samples, no post-injection samples when quiet_window_sec > 0, or max_messages truncation all return success=False with partial data; an explicit zero quiet window does not require a post-injection callback. No .bag is written. Use record_bag instead only when persistent/offline evidence is required.

17. analyze_step_response(bag_path, setpoint_field, feedback_field, mode, topic_filter, signal_type): Automated step, tracking, or sine frequency response metrics evaluation (signal_type="auto"|"step"|"sine").

18. analyze_gimbal_pid(bag_path, axis, loop, skip_initial_sec, signal_type): Domain-specific RoboMaster Gimbal PID tracking analysis supporting step (0.006 rad SOP-04 threshold) and sine wave frequency response.

19. diagnose_gimbal_tuning(bag_path, axis, step_threshold_rad, settle_criterion): Detailed Gimbal PID diagnosis tool analyzing time-domain step metrics (overshoot %, 5% & threshold settling time, steady-state error), velocity loop zero crossings / oscillation severity, position loop windup, effort limit saturation %, and actionable tuning recommendations. Detects false-deadlock step tests — feedback already starting at the final setpoint (step_info.initial_position_at_final_value + warnings code) — and prepends reset guidance (send traj_<axis>=0 first, then re-run the step).

🔧 Parameter, Controller & Target Configuration Suite

20. get_param(param_name): L0 read-only parameter access via rospy / XML-RPC master API (name accepted as an alias).

21. set_dynamic_param(node, params, confirm_token, dry_run): Hot-update a running node's runtime parameters via dynamic_reconfigure. dry_run=True or parameter set failure returns structured available_keys schema (doubles, bools, ints, strs).

22. list_controllers(): L0 read-only listing of ros_control controllers and their states via controller_manager.

23. switch_controller(start_controllers, stop_controllers, strictness, confirm_token): Start/stop ros_control controllers.

24. get_joint_limits(joint_name, robot_type): Query URDF physical and soft joint limits (lower/upper rad, effort, velocity, safety_controller k_position).

25. configure_target(mode, rosbridge_uri, ssh_host, confirm_token): Runtime sim/real target switch with asymmetric gating.

4. Core Advantages

• 🚀 ROS 1 & ROS 2 Unified: Pure Rosbridge WebSocket protocol eliminates dual API code churn during ROS 2 migration.

• 🛡️ Safety First: Physical robot write operations are protected by two-phase token confirmation and automatic dry-run verification.

• ⏱️ Hang-Proof Server: Every tool runs under a wall-clock watchdog — a single hung call is abandoned with a structured error instead of stalling the stdio stream; the embedded rospy node detects roscore restarts (master registration check — OS pids get recycled) and force re-initializes instead of going zombie.

• 🔍 Honest Failures, Never Fabricated Data: Unreachable rosbridge/master yields success=False with a remediation hint — synthetic offline data exists only behind the explicit RM_MCP_FAKE_FALLBACK=1 unit-test switch, so agents can never mistake fake topology for live state. When the rosapi node is absent, service introspection falls back to the ROS Master XML-RPC API (transport: ros_master_xmlrpc); when rosbridge itself is down, tools first attempt a self-heal via ensure_infrastructure before failing.

• 📡 WiFi Bandwidth Efficient: High-frequency topic stream throttling and on-vehicle bag recording protect wireless network links.

• 🔄 Closed-Loop Tuning: Automated pipeline from signal injection to bag analysis against domain acceptance criteria.

• 🤐 Stdio Protocol Pureness: All logging is strictly directed to sys.stderr to prevent JSON-RPC pipeline corruption.

• 🧪 Developer-Friendly: Built with Python 3.10+, uv, and FastMCP. Four-tier reliability suite: unit / component (fake rosbridge, fake XML-RPC master, fake SSH) / stdio protocol e2e (187 tests) plus Gazebo hardware-in-the-loop smoke (8 tests, RUN_HW_TESTS=1).

5. Quick Start & Testing

Prerequisites

• ROS Noetic (or ROS 2 Humble/Jazzy)

• Python >= 3.10

• uv package manager (curl -LsSf https://astral.sh/uv/install.sh | sh)

Installation

cd src/rm_mcp
uv venv
uv pip install -e .

Running Unit Tests

cd src/rm_mcp
uv run pytest

Hardware-in-the-Loop Smoke (Gazebo)

cd src/rm_mcp
RUN_HW_TESTS=1 uv run pytest tests/hardware/ -m hardware -v

Single-Step Debugging via MCP Inspector

cd src/rm_mcp
uv run mcp dev src/rm_mcp/server.py

Stdio Startup

uv --directory src/rm_mcp run python -m rm_mcp.server

Related MCP server: ROS1 Noetic MCP Server

中文

1. 项目背景

在 RoboMaster (RM) 等复杂机器人系统开发中，传统的控制器调试、PID 参数调优与系统状态排查极度依赖工程师手动敲击 CLI 命令、开启 rqt/PlotJuggler 图形界面以及人工拉取 rosbag 分析。

rm_mcp 是专为 rm_ws 工作区打造的 ROS Model Context Protocol (MCP) 服务，支持 ROS Noetic 与未来 ROS 2 无缝迁移，为 AI Agent（如 Claude、Cursor、Antigravity 等）赋予了结构化的“眼”（状态/节点/话题深度感知）和“手”（控制器启停、信号激振、录包与 PID 闭环分析）。

rm_mcp 解决了实车机器人调试中的四大核心痛点：

• ROS 1 / ROS 2 跨平台保值：基于纯正的 Rosbridge WebSocket JSON-RPC 架构（ws://host:9090），实现 ROS 1 到 ROS 2 迁移时内省与控制协议零改动。

• WiFi 带宽瓶颈：避免高频 ROS 话题（IMU、点云、关节状态）通过无线网络长流传输造成卡顿，采用“车上录制 + 后置 SCP 拉回”与“echo 限时限量防护”策略。

• 实车物理安全风险：防止 AI 幻觉或误触引发底盘/发射机构意外动作，引入两阶段 Token 确认机制 (confirm_token)。

• 调参闭环自动化：将“参数设置 → 激振输入 → 数据录制 → PID 性能分析”全流程自动化，并对照团队 SOP 标准（如云台 0.006 rad 跟踪误差）自动得出判定结论。

2. 架构与信任阶梯

rm_mcp 设计了 4 级 信任阶梯 (Trust Ladder)（L0 至 L3）与 两阶段确认机制：

+-------------------------------------------------------------------------------+
|                                信任阶梯架构                                   |
+-------------------+-----------------------------------------------------------+
| 层级              | 作用说明与涵盖操作                                        |
+-------------------+-----------------------------------------------------------+
| L0: 只读层        | 话题/节点内省、参数查询、preflight 前置检查               |
| L1: 仿真写操作    | Gazebo 仿真环境下的参数修改与控制器测试                   |

> **仿真模式门禁**：红线关键字（`chassis`/`shooter`/`friction`/`balance`/`legged`）与幅值 > 1.0 rad/s 的 inject 在仿真下依然需要 `confirm_token`（设计使然）；每个 `PENDING_CONFIRMATION` 响应都带 `trigger_reason`，明确说明触发原因。
| L2: 真车无动力    | 实车断电/无动力状态下的传感器与接口验证                   |
| L3: 真车动力      | 实车上电（包含 launch、发射机构、底盘等高危操作）         |
+-------------------+-----------------------------------------------------------+

两阶段确认机制 (Two-Phase Handshake)

当目标为实车（TARGET_MODE=real）或涉及敏感红线关键字（如 shooter、friction、chassis、balance、legged）时，rm_mcp 会自动拦截：

1. 第一阶段 (待确认)：MCP 返回结构化确认请求，生成 6 位 confirm_token（有效期 5 分钟）与可读的动作摘要。

2. 第二阶段 (确认执行)：Agent 向用户展示确认提示，用户同意后 Agent 携带有效的 confirm_token 重新发起调用。

3. 功能特点与工具清单 (全量 25 项 MCP 工具)

rm_mcp 内置 25 项开箱即用的 MCP 工具：

🌐 跨平台内省套件 (纯 Rosbridge WebSocket)

五个服务/采样工具均带 Rosbridge 自愈：不可达时先经 ensure_infrastructure("rosbridge_websocket") 尝试拉起（30s 冷却限流），重试一次，仍失败才诚实报错——ROS Master XML-RPC 回退仅保留给"rosbridge 在线但 rosapi 节点缺失"场景。

1. get_topic_info(topic)：查询话题消息类型、发布者与订阅者节点列表。

2. get_node_info(node)：查询节点活动详情。

3. list_topics()：列出当前 ROS 系统中所有活动的话题及其消息类型。

4. list_nodes()：列出当前 ROS 系统中所有活跃的节点。

5. echo_topic(topic, duration_sec, max_messages)：安全采样指定话题的内容。

⚙️ 控制启停、激振录包与 PID 分析套件

6. preflight()：环境前提状态检查（ROS Master、Systemd 服务、ROBOT_TYPE 匹配、robot_state_publisher TF 准备状态、云台控制器 TF 依赖链诊断——如 chassis_controller 停止导致 odom -> base_link TF 缺失、电机 effort_limit 与 PID 增益比对、safety_controller 软限位卡死告警）。

7. launch(package, launch_file, args, confirm_token)：仿真模式下本地启动 roslaunch；实车模式经 SSH 启动。调参起手式推荐：load_controllers.launch 整体加载控制器（注意只 load 不 start，需再用 switch_controller 启动），需要遥控器时改用 manual.launch。

8. list_launches()：活动进程句柄发现 (host:package:launch_file:PID)。

9. kill_launch(handle, confirm_token)：句柄终止整个进程组。

10. ensure_infrastructure(target_services, confirm_token)：幂等检查并启动基础设施服务 (rosapi, rosbridge_websocket, robot_state_publisher)。

11. set_param(param_name, value, confirm_token)：带写前状态捕获与写后自动回读校验的参数修改。

12. inject_signal(topic, field, waveform, amplitude, frequency, duration_sec, publish_rate_hz, extra_fields, dry_run, force, confirm_token)：阶跃/正弦/方波/三角波信号注入与 0.0 归零复原。在 extra_fields 应用后自动将缺失/全零的顶层 stamp / header.stamp 补全为 rospy.Time.now()（含归零消息；显式非零 stamp 保留），消除 rosbag 分析中的 0 时间戳问题。

13. record_bag(topics, duration_sec, mode, output_path, background, trigger_on_topic)：同步或后台异步录包 (background=True)。trigger_on_topic 会把该 topic 追加进 rosbag 录制列表；rosbag record --duration 在任一被录 topic 首消息到达后才开始倒计时，因此 duration_sec 需覆盖注入延迟 + 信号持续 + 余量。

14. stop_record_bag(bag_handle, handle, confirm_token)：优雅停止后台录包句柄并拉回文件解析元数据。

15. capture_topic_sync(topic, fields, duration_sec, max_messages, timeout_sec)：零磁盘 I/O 内存同步话题采样，适用于稳态或外部输入诊断。

16. capture_inject_sync(capture_topic, inject_topic, field, waveform, amplitude, frequency, duration_sec, publish_rate_hz, quiet_window_sec, fields, max_messages, extra_fields, dry_run, force, confirm_token)：动态闭环首选路径：确认采集回调已实际到达后才开始激振，安全确认与独占发布仲裁原样委托给 inject_signal，并从注入完成时刻起完整采集 Quiet State 窗口。仅有 readiness 样本、注入期间无新样本、quiet_window_sec > 0 时注入后无新样本或 max_messages 提前截断，均携带部分数据返回 success=False；显式零 Quiet 窗口不要求注入后回调。全程不写 .bag；仅需持久化/离线证据时使用 record_bag。

17. analyze_step_response(bag_path, setpoint_field, feedback_field, mode, topic_filter, signal_type)：阶跃、跟踪与正弦频域响应指标评估。

18. analyze_gimbal_pid(bag_path, axis, loop, skip_initial_sec, signal_type)：云台 PID 轴向隔离 pass/fail 快速筛查。

19. diagnose_gimbal_tuning(bag_path, axis, step_threshold_rad, settle_criterion)：云台 PID 详细诊断工具，评估时域阶跃指标（超调 %、5% 与阈值调节时间、稳态误差）、速度环零交越震荡严重度、位置环 windup、硬件饱和 % 与可操作调参建议。可检出阶跃假死——注入前未复位、反馈起点已处终值（step_info.initial_position_at_final_value + warnings 码），并在建议首位给出复位指引（先发 traj_<axis>=0，再重做阶跃测试）。

🔧 参数、控制器与目标配置套件

20. get_param(param_name)：L0 只读参数查询。

21. set_dynamic_param(node, params, confirm_token, dry_run)：经 dynamic_reconfigure 热更新运行中节点的运行时参数。dry_run=True 或失败时返回结构化 available_keys schema (doubles, bools, ints, strs)。

22. list_controllers()：L0 只读列出 ros_control 控制器及其运行状态。

23. switch_controller(start_controllers, stop_controllers, strictness, confirm_token)：启停 ros_control 控制器。

24. get_joint_limits(joint_name, robot_type)：解析 URDF 关节物理与软限位（lower/upper_limit, soft_lower/upper_limit, effort, velocity, k_position）。

25. configure_target(mode, rosbridge_uri, ssh_host, confirm_token)：运行时切换 sim/real 目标。

4. 项目优势

• 🚀 ROS 1 / ROS 2 无缝兼顾：基于纯正 Rosbridge WebSocket 协议，避免在两套原生 API 间重复开发。

• 🛡️ 安全第一：实车写操作红线防护，两阶段 Token 确认与写后自动校验双重保险。

• 🔍 诚实失败，绝不造假：rosbridge/master 不可达时返回 success=False + remediation 排查建议——离线假数据仅存在于显式单测开关 RM_MCP_FAKE_FALLBACK=1 之下，Agent 永远不会把编造拓扑当成真实系统状态。

• 📡 网络高效：高频数据限时限量，车载录包后置拉回，不占用调试 WiFi 带宽。

• 🔄 闭环调参：从激振到分析全程自动化，指标直接对齐团队 SOP 验收标准。

• 🤐 Stdio 纯净：日志全量重定向至 sys.stderr，绝不污染 JSON-RPC 标准输入输出管道。

• 🧪 现代化工程：基于 Python 3.10+、uv 与 FastMCP 构建，四层可靠性测试体系：单元 / 组件（fake rosbridge、fake XML-RPC master、fake SSH）/ stdio 协议 e2e（187 项）+ Gazebo 硬件在环冒烟（8 项，RUN_HW_TESTS=1 门控）。

5. 快速开始与测试

环境要求

• ROS Noetic (或 ROS 2 Humble/Jazzy)

• Python >= 3.10

• uv 包管理器 (curl -LsSf https://astral.sh/uv/install.sh | sh)

安装说明

cd src/rm_mcp
uv venv
uv pip install -e .

运行单元测试

cd src/rm_mcp
uv run pytest

Gazebo 硬件在环冒烟

cd src/rm_mcp
RUN_HW_TESTS=1 uv run pytest tests/hardware/ -m hardware -v

通过 MCP Inspector 可视化调试

cd src/rm_mcp
uv run mcp dev src/rm_mcp/server.py

Stdio 模式启动

uv --directory src/rm_mcp run python -m rm_mcp.server

6. 许可证

本项目采用 BSD 许可证开源。