TopicForge
[PyPI version]
[CI]
[Python versions]
[License: MIT]
[Read-only]
A read-only MCP (Model Context Protocol) server that lets an AI agent inspect a ROS2 graph, recorded bag files and the DDS layer underneath ROS. The code has no write path: it cannot publish to the bus or command a robot, and there is no permission system to configure.
Every tool also declares itself read-only in the MCP protocol (readOnlyHint, destructiveHint=false, idempotentHint, with an honest openWorldHint), and a test fails if any tool lacks it. See SECURITY.md for the guarantee and its limits.
It gives the agent seventeen typed tools that return frozen Pydantic schemas (output contract 2: one object per tool, one note, units in the field names; see docs/CONTRACT.md), identical whether the server talks to a real robot or to its built-in mock fixtures. Ask why nav_planner gets no scan, and the agent reads the bus, finds the BEST_EFFORT writer facing a RELIABLE reader and names the incompatible policy (see examples/02-debug-qos-mismatch.md). It is meant for ROS2 developers, robotics ML/CV engineers and teams that cannot accept a write path into a production stack.
For DDS, TopicForge joins a domain as a read-only participant through one open-source binding (Eclipse CycloneDDS from PyPI) and reads the builtin discovery topics that the OMG DDS-RTPS protocol standardizes. Cyclone DDS, Fast DDS and Dust DDS participants have been observed on live buses: the Fast DDS ones on every CI run of the ROS 2 test bench (Humble and Jazzy with rmw_fastrtps_cpp) and in the OmniSim validation. RTI Connext, OpenDDS and CoreDX announce themselves through the same standard discovery, but none of them has been observed yet. This covers discovery only: participants, readers, writers and their QoS. See docs/dds-interop-matrix.md.
Quickstart
No ROS2 is needed; the mock adapter serves deterministic fixtures for a small differential robot (LIDAR + RGB camera). Python 3.10 to 3.13.
The server speaks MCP over stdio and waits for a client, so wire it into one. For Claude Desktop, add to claude_desktop_config.json:
Then ask it to list the topics or to analyze /tmp/demo.mcap. For Claude Code: claude mcp add topicforge -- topicforge. If the client cannot launch the server (WSL2 to Windows, a robot over an SSH tunnel), topicforge --transport streamable-http --port 8765 serves http://127.0.0.1:8765/mcp on loopback (another --host needs a bearer token in TOPICFORGE_HTTP_TOKEN); see docs/CLIENTS.md.
Other MCP clients. Ready-to-paste configs for Claude Desktop, Cursor, VS Code, Windsurf, Cline, Roo Code, Continue, Zed, JetBrains, Codex, Gemini CLI, Goose and more are in docs/CLIENTS.md. One-click: Add to Cursor | Install in VS Code.
Setup for a real ROS2 environment (WSL2, Linux, Docker, native Windows) is in docs/TESTING.md; recurring monitoring prompts and the privacy contract are in docs/TUTORIEL.md.
Tools
Every response except health_check carries mode_effective ("live" or "mock"), so a caller can tell a real graph from fixtures, and a single note that explains anything unusual. health_check.contract_version is 2. The full reference, with every field, is generated in docs/TOOLS.md; upgrading from 0.6.x: docs/MIGRATION-0.7.md.
The server also offers three MCP prompts, diagnose-dds-bus, inspect-ros2-robot and diagnose-nav2, that give a client the read-only call order, and short server instructions for clients that do not show prompts (see docs/CLIENTS.md).
Walkthroughs against the mock, each with the exact tool calls and payloads, are in examples/. To run the DDS tools against a real bus with several programs and vendors, see examples/dds/README.md (python examples/dds/run_all.py).
Modes
live and auto degrade instead of failing: if neither ros2 nor a DDS backend comes up, the server serves the mock fixtures and health_check reports mode: "mock" next to requested_mode. Only TOPICFORGE_MODE=mock forces fixtures unconditionally. The live ROS2 adapter shells out to the ros2 CLI, so rclpy does not need to be importable.
DDS backends
TOPICFORGE_DDS_BACKEND accepts mock (default), cyclone and auto (cyclone if the binding imports, else mock). The default mock selects no DDS backend: installing the Cyclone binding is not enough, you must also set TOPICFORGE_DDS_BACKEND=cyclone. With TOPICFORGE_MODE=live and no backend selected, the DDS tools raise DDS module is not active: ... with the actual cause (backend not selected, binding not installed, or binding installed but the adapter failed to start), and health_check reports dds_backend: "none" plus dds_inactive_note. An explicit value is honoured with or without ros2 on PATH, in any mode except mock. If the binding is missing or the participant cannot start, the server logs a warning naming the cause and falls back to the ROS2 CLI alone, or to the mock fixtures. When both ros2 and a DDS backend are up, a composite adapter routes the nine ros2 CLI tools (list_topics, get_topic_info, list_nodes, get_node_info, inspect_tf, graph_diff, diagnose, sample_messages, analyze_bag) to the CLI and the six DDS tools to the DDS backend; peek_bag_samples goes to the ROS half but runs no ros2 process.
Fast DDS participants (ROS 2 nodes on rmw_fastrtps_cpp) need no backend of their own: the Cyclone backend observes them through standard RTPS discovery. Since 0.9.0 TOPICFORGE_DDS_BACKEND=fast is refused, like rti, opensplice, coredx, intercom, opendds and dust: startup stops with an error that lists the valid values (see docs/TROUBLESHOOTING.md). Full backend selection, the routing table and the QoS mismatch scenario are in docs/DDS_QUICKSTART.md.
Configuration reference
Command-line options (not environment variables): --transport stdio|streamable-http (default stdio), --port N (default 8765) and --host ADDR (default 127.0.0.1), the last two only with streamable-http. The HTTP transport binds 127.0.0.1 with DNS-rebinding protection; TOPICFORGE_HTTP_TOKEN (at least 32 characters) makes every request carry Authorization: Bearer <token> and is required for any other --host (plain HTTP, so prefer an SSH tunnel); see docs/CLIENTS.md.
Samples with comments are in .env.example. Any invalid value stops the server with topicforge: configuration error: ... and exit code 2, so a typo cannot silently change behaviour.
Limitations
External validation against a simulated robot's ground truth (OmniSim): docs/VALIDATION.md.
- DDS validation is partial. The Cyclone adapter has run against real buses with Cyclone, Fast DDS and Dust DDS participants, on Windows and in CI on Ubuntu and Windows (
.github/workflows/demo.yml, .github/workflows/ros2-live.yml). No RTI, OpenDDS, CoreDX or OpenSplice participant has been observed by this project. The multi-vendor claim rests on the RTPS protocol guarantee, not on a recorded cross-vendor run.
- User-topic payloads are not decoded.
peek_dds_samples on a user topic returns count 0 and a note that the topic is announced on the bus; no traffic is read. topic_metrics therefore has data only for the builtin discovery topics and says so in its status. It is a discovery-layer probe, not a publish-rate monitor.
- Liveliness at runtime is not observed. A writer that is alive but silent (a hung process whose lease is still renewed) looks healthy, because TopicForge reads discovery, not data. An opt-in data probe is planned. A crash and a clean leave cannot be told apart, and
lost_ns is an upper bound of the death.
- Cyclone vendor ids: participants that do not follow the RTPS vendor-id convention in their GUID prefix (Dust DDS, and RTI by default) are reported with vendor
unknown.
- Single domain: the server observes the domain it joined at startup; changing it needs a restart.
- DDS Security is not handled. A participant without credentials sees an empty secure bus.
detect_qos_mismatches checks Partition, type name, Reliability, Durability, Deadline, Liveliness, LatencyBudget, Ownership (kind), DestinationOrder and DataRepresentation (History as a risk); Presentation, XTypes assignability and runtime behavior are not checked, and the result lists them in policies_unchecked. It returns a MismatchScan envelope: read reports for the mismatches.
sample_messages (live) streams ros2 topic echo until count messages arrive or timeout_s (1..40, default 10, for the whole call) runs out, and returns what arrived with a note (N of M messages within T s, saying whether a publisher exists). QoS is matched to the publishers, so latched topics work. The payload has nested named fields (payload.header.stamp.sec); timestamp_ns is header.stamp (the publisher's clock, sim time on a simulation), the time in the body of /clock, /tf and /rosout messages (stamp_source payload), or 0 for other headerless types, with stamp_source and received_ns (wall clock when the CLI printed it). Arrays are cut at 128 elements by default; max_array_length (1..65536, or null for no cut) and arrays_summary_only change that, and a cut is listed under _truncated_fields. nan and inf come back as strings. count above 50 is capped with a note.analyze_bag (live) parses ros2 bag info text for the totals and counts; anomaly detection is mock-only. Per-topic times, rates ((n - 1) / span) and latched are added when the bag can be read locally (.db3 with the standard library, .mcap with rosbags), else rates fall back to count / bag duration (frequency_basis). peek_bag_samples reads the file itself, through rosbags, returns the first count messages in recording order (not the last), and is served only by the ROS2 CLI adapter or the mock; its timestamp_ns is the message's header.stamp (stamp_source header), the body time of Clock, TFMessage and Log messages (payload) or, without either, the bag record time (stamp_source recorded), and recorded_ns is always the bag record time; bags that embed no message definitions (Humble .db3) are decoded with the Humble definitions, or the distro the bag records, and note says so. Without ros2, bag tools return fixtures: check health_check for mode: "mock" before trusting bag output. health_check also reports sim_clock_published (live only): true when /clock has a publisher, a hint that header stamps may be simulation time.- Calls are serialized per backend (one
ros2 call at a time, one DDS call at a time) and each call has a 45 s wall budget, lock wait included; a call that cannot get its lane fails at once with busy. health_check and peek_bag_samples take no lock.
- No streaming or push subscriptions: tools are strictly request/response.
Not done yet: an opt-in probe to tell a hung writer from a healthy one, wider real-bus validation (RTI, OpenDDS) and DDS Security. Open work is tracked in issues.
Telemetry
Opt-in, anonymous and off by default. When off, instrumentation returns the handler unchanged: no event is built, no transport is constructed, no network code runs (pinned by tests/test_telemetry.py::test_build_app_off_makes_no_transport_calls).
On-values: on, 1, true, yes, enabled. Off-values: unset, off, 0, false, no, disabled. Anything else is a configuration error rather than a silent "off".
When on, each tool call emits one event with exactly six fields:
Never sent: topic names, message types or payloads, bag paths or contents, hostnames, usernames, IP addresses, environment variables, error messages. The field set is fenced by tests/test_telemetry.py::test_payload_contains_only_whitelisted_keys; adding a field requires updating this section. The default transport is a structured log line; there is no HTTP endpoint yet. The implementation is in src/topicforge/telemetry/.
Security model
TopicForge is designed for local trust: it runs as a subprocess of your MCP client on a machine you control and inspects your own ROS2 graph, DDS domain and bag files. It is not hardened for adversarial inputs.
TOPICFORGE_ROS2_BIN accepts an arbitrary path; treat it the way you treat PATH.analyze_bag and peek_bag_samples open whatever path the client passes (no workspace isolation, no symlink restriction).- All
ros2 invocations use subprocess.run with an argument list, never shell=True. ROS2 topic names are validated against ^/[A-Za-z0-9_/]+$ first.
- The server loads no third-party code at startup.
- No outbound network calls unless telemetry is turned on.
Before exposing TopicForge to untrusted MCP clients (hosted endpoints, shared environments), add path isolation and revisit the TOPICFORGE_ROS2_BIN policy. Vulnerability reports: see SECURITY.md.
Development
Tests run against the mock adapter, the live adapter's pure parsers and the binding-free DDS helpers; they never need a running ROS graph. Tests needing the cyclonedds binding skip themselves when it is absent, and the integration marker (real-bus tests) is deselected by default. The Makefile (make check) uses POSIX shell syntax; on plain PowerShell run the commands above. See CONTRIBUTING.md.
Upgrading
TopicForge is pre-1.0; CHANGELOG.md lists every change, including the yanked releases and removed extras. 0.7.0 changed the shape of tool results (contract_version 2): docs/MIGRATION-0.7.md maps every old field to its new one.
Layout
Layers are strictly separated: handlers never call subprocess, adapters are the only code that talks to a backend, and new backends implement the MiddlewareAdapter protocol in adapters/base.py.
License
MIT, see LICENSE. Integration or support work for a specific ROS 2 / DDS setup: [email protected].
