# TopicForge

**Category:** 💻 Developer Tools  
**Repository:** https://github.com/yaniswav/TopicForge  
**Views:** 0  
**Installs:** 0  
**Upvotes:** 0  
**Directory Page:** https://allmcps.com/mcp/topicforge

## Description
Read-only MCP server for ROS2 and DDS: topics, participants, QoS mismatches, bags.

## Claude Desktop Quick Installation
Heuristic fallback — verify the package name and runner against the repository README before running it. Uses `npx` (confidence: low):

```json
"mcpServers": {
  "topicforge": {
    "command": "npx",
    "args": ["-y","topicforge"]
  }
}
```

## Documentation & README

# TopicForge

<!-- mcp-name: io.github.yaniswav/topicforge -->

[![PyPI version](https://img.shields.io/pypi/v/topicforge.svg)](https://pypi.org/project/topicforge/)
[![CI](https://github.com/yaniswav/TopicForge/actions/workflows/ci.yml/badge.svg?branch=main)](https://github.com/yaniswav/TopicForge/actions/workflows/ci.yml)
[![Python versions](https://img.shields.io/pypi/pyversions/topicforge.svg)](https://pypi.org/project/topicforge/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://github.com/yaniswav/TopicForge/blob/main/LICENSE)
[![Read-only by architecture](https://img.shields.io/badge/safety-read--only_by_architecture-2563eb)](https://github.com/yaniswav/TopicForge#security-model)

> **The safety-first read-only MCP for ROS2 robotics: graph introspection, bag analysis, and multi-vendor OMG DDS-RTPS observability.** TopicForge lets AI agents inspect your ROS2 graph, recorded bag files, and the raw DDS layer beneath ROS, without ever publishing back to the bus. **Eleven typed read-only tools** (5 ROS2 graph + 3 DDS + 3 observability/bag) share a single Pydantic envelope so an LLM caller reads one schema across the whole stack. It joins a domain through one OSS Python participant (Eclipse CycloneDDS or eProsima Fast DDS) and reads the builtin discovery topics the OMG protocol standardizes, so it observes **every conformant vendor on the wire** (RTI Connext, OpenDDS, CoreDX, Dust DDS in Rust) with no proprietary binding. See [`docs/dds-interop-matrix.md`](https://github.com/yaniswav/TopicForge/blob/HEAD/docs/dds-interop-matrix.md) for the canonical multi-vendor positioning and the OMG May 2025 interop reference.

TopicForge is a production-minded MCP (Model Context Protocol) server that lets AI agents (such as Claude) inspect ROS2 topics, analyze ROS bag files, and (since v0.2.0) observe the raw DDS layer through a clean, structured tool interface. It is read-only by **architecture**, not by configuration: there is no write path to misconfigure, no permission system to audit, no liability conversation to have. The MCP client can see the robot stack; it cannot touch it.

This stance matters because the ROS-MCP space is no longer empty: general-purpose ROS-MCP servers exist that let an LLM publish topics, call services, and command robots. That shape is fine for demos; it is untenable for production fleets, defense systems, automotive AUTOSAR Adaptive surfaces, or anything safety-certified. TopicForge is the read-only alternative for those audiences, plus the robotics developers, ML/CV engineers, and teams that want their AI tooling to *understand* their robotics stack without commanding it.

## Why it exists

LLM agents are good at reasoning over text, but ROS2 introspection lives in a CLI + DDS world they cannot directly reach. Without grounding, an LLM will hallucinate topic names, message types, and bag contents. TopicForge bridges that gap with a small, well-typed set of MCP tools, all read-only, all returning frozen Pydantic schemas that a downstream agent can parse without ambiguity:

| Tool              | Purpose                                                |
| ----------------- | ------------------------------------------------------ |
| `health_check`    | Environment & mode introspection                       |
| `list_topics`     | Discover the ROS graph                                 |
| `get_topic_info`  | Structured info for a single topic                     |
| `sample_messages` | Peek recent messages on a topic (publish-time timestamps for `Header`-stamped types) |
| `analyze_bag`     | Summarize a `.mcap` / `.db3` / `.bag` recording        |

Outputs are structured, JSON-serializable, and stable across runtime modes - they look the same whether the server is talking to a real robot or to its built-in mock fixtures. Every response carries a `mode_effective` field (`"live"` or `"mock"`) so a downstream LLM can tell a real graph from the demo fixtures without re-reading `health_check`.

## 30-second demo without ROS2

The mock adapter ships deterministic fixtures for a small differential robot (LIDAR + RGB camera). You do not need ROS2 installed to try the full tool surface - a clean Python 3.11 venv is enough.

```bash
pip install topicforge
TOPICFORGE_MODE=mock python -m topicforge
# Windows PowerShell: $env:TOPICFORGE_MODE="mock"; python -m topicforge
```

Point any MCP client (Claude Desktop, see below) at this server and ask it to *list the topics* or *analyze `/tmp/demo.mcap`* - every tool returns realistic, typed payloads.

## Quickstart

```bash
pip install topicforge
python -m topicforge --help
TOPICFORGE_MODE=mock python -m topicforge
```

## Architecture

```
+----------------------+
|   MCP client (LLM)   |
+----------+-----------+
           |  (stdio, MCP protocol)
           v
+----------+-----------+
|  topicforge.server   |   FastMCP entrypoint, lifecycle, tool registration
+----------+-----------+
           |
           v
+----------+-----------+
|  topicforge.tools    |   Thin handlers - validate, delegate, serialize
+----------+-----------+
           |
           v
+----------+-----------+
| topicforge.services  |   Inspector / Health - orchestration & validation
+----------+-----------+
           |
           v
+----------+-----------+
| topicforge.adapters  |   ros2_live  - subprocess wrappers over `ros2` CLI
|                      |   ros2_mock  - deterministic fixtures
+----------------------+
```

Layers are strictly separated:

- **`server/`** wires the whole graph and exposes `build_app(settings)`.
- **`tools/`** registers MCP tools on FastMCP. Handlers never call ROS directly.
- **`services/`** validate inputs and orchestrate calls.
- **`adapters/`** are the *only* code that knows how to talk to a specific backend. New backends (e.g. an `rclpy`-based adapter) plug in by implementing the `RosAdapter` protocol.
- **`models/`** holds Pydantic schemas - the contract with MCP clients.
- **`config/`** resolves runtime settings from the environment.

## Runtime modes

| Mode    | When to use                                         | Backend                       |
| ------- | --------------------------------------------------- | ----------------------------- |
| `mock`  | Local development, demos, CI, screencasts           | Deterministic fixtures        |
| `live`  | A machine with ROS2 installed and sourced           | `ros2` CLI wrappers           |
| `auto`  | Detect ROS2; fall back to mock if not present       | Best available (default)      |

Mode is selected via the `TOPICFORGE_MODE` environment variable.

## Install from source

Requires Python 3.11+.

```bash
git clone https://github.com/yaniswav/TopicForge.git
cd TopicForge
python -m venv .venv
source .venv/bin/activate          # Linux / macOS
# .venv\Scripts\Activate.ps1       # Windows PowerShell
pip install -e ".[dev]"
```

Or, if you have `make`:

```bash
make dev
```

## Run

### Mock mode (no ROS2 required)

```bash
TOPICFORGE_MODE=mock python -m topicforge
```

Or:

```bash
make run-mock
```

### Live mode (requires ROS2)

Source your ROS2 distribution first, then:

```bash
source /opt/ros/humble/setup.bash
TOPICFORGE_MODE=live python -m topicforge
```

TopicForge invokes the `ros2` CLI under the hood, so it does **not** require `rclpy` to be importable. This keeps the live adapter portable across ROS2 distros.

### Multi-vendor DDS support (v0.3.0+)

Beyond ROS2 graph introspection, TopicForge observes the raw DDS bus directly via one of two OSS Python adapters (Eclipse CycloneDDS or eProsima Fast DDS) each joining as a **read-only DDS-RTPS participant**. By the OMG-DDS-RTPS protocol guarantee, both adapters see every conformant participant on the domain (RTI Connext, OpenDDS, CoreDX, Dust DDS in Rust, InterCOM, etc.) regardless of host language. See [`docs/dds-interop-matrix.md`](https://github.com/yaniswav/TopicForge/blob/HEAD/docs/dds-interop-matrix.md) for the canonical multi-vendor positioning and the [OMG May 2025 interop reference](https://github.com/yaniswav/TopicForge/blob/HEAD/docs/projet-file/references/omg-dds-interop-2025-05-08.xlsx).

Useful for non-ROS DDS stacks (defense, aerospace, automotive AUTOSAR Adaptive, industrial integration) and for diagnosing why a ROS2 subscriber isn't receiving when the graph says it should. Same safety-first contract : read-only by **architecture**. The `MiddlewareAdapter` protocol does not expose a write method on any backend.

Install one or both OSS backends. The v0.4.0 auto-detect framework picks whichever you actually installed:

```bash
# Single vendor: install only what you need
pip install topicforge[dds-cyclone]   # Eclipse CycloneDDS only
pip install topicforge[dds-fast]      # eProsima Fast DDS only
pip install topicforge[dds]           # both OSS backends (union)
```

Then select a backend (or let auto-detect pick):

```bash
TOPICFORGE_MODE=live TOPICFORGE_DDS_BACKEND=cyclone python -m topicforge
# or:
TOPICFORGE_MODE=live TOPICFORGE_DDS_BACKEND=fast python -m topicforge
# or, auto-select across the 8-vendor priority chain:
#   RTI > OpenSplice > CoreDX > InterCOM (Pro tier, if installed)
#   > OpenDDS > Fast > Cyclone > Dust > Mock (OSS fallback)
TOPICFORGE_MODE=live TOPICFORGE_DDS_BACKEND=auto python -m topicforge
```

**What the OSS install covers.** Cyclone and Fast DDS are the two working
backends; OpenDDS and Dust are stubs whose upstream Python bindings do not
exist yet, and their install extras fail today. That is not a limitation on
what you can observe: because discovery runs over the OMG-standardized wire
protocol, a Cyclone participant already sees RTI Connext, OpenDDS, CoreDX and
Fast endpoints on the same domain. **You do not need a commercial adapter to
observe a commercial bus.**

A native RTI Connext adapter exists for the cases where the standard route is
not enough (secure domains needing vendor credentials, shared-memory-only
deployments, vendor-specific extensions). It requires the RTI Python binding
and an RTI license you already hold. See [`docs/pro.md`](https://github.com/yaniswav/TopicForge/blob/HEAD/docs/pro.md) for
commercial support and integration.

Six DDS / observability tools (in addition to the five ROS2 tools above) :

| Tool                    | Since   | Purpose                                                                                                        |
| ----------------------- | ------- | -------------------------------------------------------------------------------------------------------------- |
| `list_participants`     | v0.2.0  | DDS participants discovered on a domain, with vendor, hostname, and (v0.4.0) lifecycle fields                  |
| `detect_qos_mismatches` | v0.2.0  | Reader/writer QoS incompatibilities preventing communication on a topic                                        |
| `peek_dds_samples`      | v0.2.0  | Recent samples on a raw DDS topic: v0.4.0 adds best-effort XTypes user-topic decode (distinct from `sample_messages` on ROS2 graph) |
| `participant_events`    | v0.4.0  | Lifecycle stream: `discovered` / `lost` participant events over a configurable window                         |
| `topic_metrics`         | v0.4.0  | Temporal metrics: observed frequency, sequence gaps, latency p50/p95/p99 over a sliding window                |
| `peek_bag_samples`      | v0.4.0  | Post-mortem inspection: decoded samples from a recorded `.mcap` / `.db3` / `.bag` file                        |

**Composite adapter (v0.4.0 Phase 1+).** When `TOPICFORGE_MODE=live` is paired with a DDS backend (`cyclone`, `fast`, ...), TopicForge instantiates **both** a ROS2 CLI adapter and the chosen DDS adapter and routes per-tool category: the 5 ROS2 tools hit the CLI, the DDS / observability tools hit the DDS backend. ROS2-only or DDS-only setups still work: the missing half is skipped and the present half serves what it can. The mock backend continues to expose all 11 tools against deterministic fixtures for local development.

**`peek_dds_samples` payload shape (v0.4.0 Phase 1.5).** Full-fidelity on the 4 builtin DCPS topics (`DCPSParticipant`, `DCPSSubscription`, `DCPSPublication`). Arbitrary user topics return best-effort decoded samples with a `_decode_status` annotation : `"full"` (every IDL field decoded; currently a v0.4.0+ Cyclone XTypes path), `"partial"` (some fields decoded, others opaque), or `"raw"` (binding could not resolve the dynamic XTypes). The diagnostic key `_decode_note` carries a short explanation when the status is non-`full`. The wire shape is identical across Cyclone and Fast backends. **Caveat (v0.5.x):** on the current user-topic *raw* path `_raw_bytes_hex` is **empty**. A `"raw"` status means "topic present on the bus but not decoded", not "here are the serialized bytes to re-decode". Capturing the on-wire CDR bytes into the fallback is roadmapped ; until then use the Cyclone full/partial XTypes path for actual user-topic payloads. The 4 builtin DCPS topics are unaffected (always structured).

**`RTI Connext`** is v0.4.0+ Pro tier (BYO license; see `docs/pro.md`).

**Full 5-minute walkthrough** (backend selection, the canonical QoS-mismatch debugging scenario, troubleshooting) lives in [`docs/DDS_QUICKSTART.md`](https://github.com/yaniswav/TopicForge/blob/HEAD/docs/DDS_QUICKSTART.md). Migration history : [v0.2 -> v0.3](https://github.com/yaniswav/TopicForge/blob/HEAD/docs/MIGRATION_v0.2_to_v0.3.md), [v0.3 -> v0.4](https://github.com/yaniswav/TopicForge/blob/HEAD/docs/MIGRATION_v0.3_to_v0.4.md).

### Configure with Claude Desktop

Add to your `claude_desktop_config.json`:

```json
{
  "mcpServers": {
    "topicforge": {
      "command": "python",
      "args": ["-m", "topicforge"],
      "env": { "TOPICFORGE_MODE": "auto" }
    }
  }
}
```

## Test

```bash
pytest
# or
make test
```

Tests run entirely against the mock adapter and the live adapter's pure parsers - they never require a running ROS graph. The full suite completes in well under a second.

## Lint & format

```bash
make lint     # ruff check
make fmt      # ruff format
make check    # both, plus tests (CI bundle)
```

> **Windows note.** The `Makefile` uses POSIX shell syntax (`VAR=value cmd`,
> `find ... -exec`). Run it from Git Bash, WSL, or MSYS2. From a plain
> PowerShell session, invoke the underlying commands directly:
>
> ```powershell
> python -m ruff check src tests
> python -m ruff format src tests
> python -m pytest
> $env:TOPICFORGE_MODE = "mock"; python -m topicforge   # equivalent of `make run-mock`
> ```

## Configuration reference

| Variable                    | Default | Description                                                                   |
| --------------------------- | ------- | ----------------------------------------------------------------------------- |
| `TOPICFORGE_MODE`           | `auto`  | `mock`, `live`, or `auto`                                                     |
| `TOPICFORGE_LOG_LEVEL`      | `INFO`  | `DEBUG`, `INFO`, `WARNING`, `ERROR`                                           |
| `TOPICFORGE_ROS2_BIN`       | `ros2`  | Name (or path) of the ROS2 CLI binary                                         |
| `TOPICFORGE_TELEMETRY`      | `off`   | Opt-in anonymous usage telemetry. See [Telemetry](#telemetry).                |
| `TOPICFORGE_DDS_BACKEND`    | `mock`  | DDS module backend: `mock`, `cyclone`, `fast`, `rti`, `opensplice`, `coredx`, `intercom`, `opendds`, `dust`, or `auto`. The v0.4.0 Phase 1.5 auto-detect chain resolves to: `rti > opensplice > coredx > intercom` (Pro tier, if installed) `> opendds > fast > cyclone > dust > mock`. See [Multi-vendor DDS support](#multi-vendor-dds-support-v030). |
| `TOPICFORGE_DDS_DOMAIN_ID`  | `0`     | DDS domain id observed (0..232) when a DDS backend is active.                 |

See [`.env.example`](https://github.com/yaniswav/TopicForge/blob/HEAD/.env.example).

## Telemetry

TopicForge ships an **opt-in, anonymous, minimal** telemetry hook. It is **off by default** and the OFF code path performs **zero network calls**: pinned by a unit test (`tests/test_telemetry.py::test_build_app_off_makes_no_transport_calls`).

### How to opt in

```bash
TOPICFORGE_TELEMETRY=on python -m topicforge
# Windows PowerShell: $env:TOPICFORGE_TELEMETRY="on"; python -m topicforge
```

Accepted on-values: `on`, `1`, `true`, `yes`, `enabled` (case-insensitive). Anything else (including unset) keeps telemetry off.

### How to opt out

Unset the variable, set it to `off`, or just don't touch it. Opt-out is the default.

### Exactly what is sent

When telemetry is on, each MCP tool call emits a single event with **only** these six fields:

| Field             | Example          | Notes                                                                  |
| ----------------- | ---------------- | ---------------------------------------------------------------------- |
| `tool_name`       | `"list_topics"`  | One of the eleven MCP tools, never argument values.                   |
| `latency_ms`      | `12.34`          | Wall-clock duration of the handler, rounded to 2 decimals.             |
| `mode`            | `"mock"`         | Effective runtime mode: `mock` or `live`.                              |
| `version`         | `"0.1.2"`        | TopicForge server version.                                             |
| `session_id`      | `"a1b2c3..."`      | Random UUID generated per process. Never persisted, never re-used.     |
| `success`         | `true`           | Whether the handler returned (true) or raised (false).                 |

### What is **never** sent

- Topic names, message types, message payloads
- Bag file paths or bag contents
- Hostnames, usernames, IP addresses, ROS distro, environment variables
- Stack traces, error messages, or any free-form text
- Any persistent identifier: `session_id` is regenerated on every server start

The payload shape is fenced by `tests/test_telemetry.py::test_payload_contains_only_whitelisted_keys`. Adding a field there requires a matching change in this section.

### Where the code lives

The complete telemetry implementation is in [`src/topicforge/telemetry/`](https://github.com/yaniswav/TopicForge/blob/HEAD/src/topicforge/telemetry/). Read it in under five minutes. The default transport is a structured log line (no HTTP endpoint yet); a future S3-backed endpoint will plug into the same `Transport` callable without touching tool handlers.

## Security model

TopicForge is designed for **local trust**: it runs as a subprocess of your MCP client (Claude Desktop, Claude Code) on a machine you control, and inspects your own ROS2 graph or your own bag files. It is not hardened for adversarial inputs.

- `TOPICFORGE_ROS2_BIN` accepts an arbitrary path - if you point it at a malicious binary, TopicForge will execute it. Treat the variable the way you treat `PATH`.
- `analyze_bag` opens whatever path the MCP client passes (no workspace isolation, no symlink restriction). The threat model assumes the client is your trusted agent acting on your behalf.
- All `ros2` CLI invocations use `subprocess.run` with an argument list - never `shell=True`. Topic names are validated against a strict allowlist (`^/[A-Za-z0-9_/]+$`) before being passed to the CLI.
- No outbound network calls by default. Since v0.1.1, opt-in anonymous usage telemetry is available behind `TOPICFORGE_TELEMETRY=on`: see [Telemetry](#telemetry) for the exact payload and opt-out instructions. When off (the default), the OFF code path is a verified no-op.

Before exposing TopicForge to *untrusted* MCP clients (hosted endpoints, shared environments), add path isolation and revisit the `TOPICFORGE_ROS2_BIN` policy.

## MVP limitations

- `sample_messages` in live mode uses `ros2 topic echo --csv --once` with a short timeout; topics with no current publisher will return an empty sample. `MessageSample.timestamp_ns` is the message's `header.stamp` (publish time) for `Header`-stamped messages and `0` for headerless types (`std_msgs/String`, `geometry_msgs/Twist`, ...); surfacing the rmw receive timestamp for arbitrary types waits on the future `rclpy`-backed adapter.
- `sample_messages` silently clamps `count` to 50 to keep tool output bounded; requests for more than 50 messages return at most 50 (the `SampleResult.count` field reflects what was actually returned).
- `analyze_bag` in live mode shells out to `ros2 bag info` and parses its text output. Deep anomaly detection is mock-only for now.
- No streaming / push subscriptions in the MVP. Tools are strictly request/response.
- Live adapter is CLI-based, not `rclpy`-based - by design, for portability.

## Roadmap

See [`docs/product-plan.md`](https://github.com/yaniswav/TopicForge/blob/HEAD/docs/product-plan.md) for the full product trajectory.

Near-term additions on the bench:

- `rclpy`-backed live adapter for faster & richer sampling (per-message rmw receive timestamps, windowed sampling): gated on external user demand
- Extended QoS coverage (Liveliness, Ownership, Partition, TimeBasedFilter, LatencyBudget): v0.5.x patch
- Real `RtiConnextAdapter` in the Pro tier (BYO RTI Connext license, gated by `TOPICFORGE_LICENSE_KEY`): the v0.4.0 Phase 1.5 framework is in place ; production binding pending Pro tier launch
- URDF inspector / validator MCP tools (Pro tier)
- Bag anomaly detection (clock jumps, gaps, dropped frames, TF tree health): Pro tier
- Dataset export helpers (rosbag -> COCO / HF Datasets)
- Synthetic data pipeline controller (Blender, Gazebo, Isaac Sim)
- Hosted MCP endpoint with auth (Phase 3; depends on Pro tier traction)

## Project layout

```
topicforge-mcp/
├── README.md                  # You are here
├── Makefile                   # Common developer tasks
├── pyproject.toml             # Build & tooling config
├── .env.example               # Example runtime configuration
├── docs/
│   └── product-plan.md        # Product strategy & roadmap
├── src/topicforge/
│   ├── __main__.py            # `python -m topicforge`
│   ├── server/                # MCP bootstrap & lifecycle
│   ├── tools/                 # MCP tool definitions
│   ├── services/              # Domain orchestration
│   ├── adapters/
│   │   ├── ros2_live/         # `ros2` CLI wrappers
│   │   ├── ros2_mock/         # Deterministic fixtures (incl. DDS)
│   │   ├── dds_cyclone/       # Eclipse CycloneDDS adapter (lazy import)
│   │   ├── dds_fast/          # eProsima Fast DDS adapter (lazy import)
│   │   └── common/            # Vendor-neutral helpers + pure QoS analyzer
│   ├── models/                # Pydantic schemas
│   └── config/                # Settings & mode resolution
└── tests/                     # Pytest suite (mock-only, no ROS2/DDS required)
```

## License

MIT - see [LICENSE](https://github.com/yaniswav/TopicForge/blob/HEAD/LICENSE).