TopicForge

TopicForge

一个只读的 MCP（Model Context Protocol）服务，可让 AI Agent 检查 ROS2 图、录制的 bag 文件，以及 ROS 下层的 DDS 层。代码没有写入路径：它无法向总线发布消息或控制机器人，也无需配置权限系统。

每个工具也会在 MCP 协议中声明自己为只读（readOnlyHint、destructiveHint=false、idempotentHint，以及如实填写的 openWorldHint），并且只要有任何工具缺少这些声明，测试就会失败。有关这一保证及其局限，请参阅 SECURITY.md。

它为 Agent 提供 16 个类型化工具，返回冻结的 Pydantic 架构（输出契约 2：每个工具返回一个对象、一个 note，单位写在字段名称中；参见 docs/CONTRACT.md）。无论服务连接真实机器人还是使用内置模拟固件，输出都完全相同。询问 nav_planner 为什么收不到扫描数据，Agent 就会读取总线，发现 BEST_EFFORT 写入端对应的是 RELIABLE 读取端，并指出不兼容的策略（参见 examples/02-debug-qos-mismatch.md）。它面向 ROS2 开发者、机器人 ML/CV 工程师，以及无法接受生产技术栈中存在写入路径的团队。

在 DDS 方面，TopicForge 通过一个开源绑定（来自 PyPI 的 Eclipse CycloneDDS）以只读参与者身份加入域，并读取 OMG DDS-RTPS 协议标准化的内置发现主题。在实时总线上已观测到 Cyclone DDS、Fast DDS 和 Dust DDS 参与者：Fast DDS 参与者出现在 ROS 2 测试台的每次 CI 运行中（使用 rmw_fastrtps_cpp 的 Humble 和 Jazzy），也出现在 OmniSim 验证中。RTI Connext、OpenDDS 和 CoreDX 会通过同一标准发现机制进行声明，但目前尚未观测到其中任何一种。此功能仅涵盖发现信息：参与者、读取端、写入端及其 QoS。参见 docs/dds-interop-matrix.md。

快速开始

无需 ROS2；模拟适配器为小型差速机器人（激光雷达 + RGB 摄像头）提供确定性测试数据。支持 Python 3.10 至 3.13。

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

服务通过 stdio 使用 MCP 协议，并等待客户端连接，因此请将它接入一个客户端。对于 Claude Desktop，请将以下内容添加到 claude_desktop_config.json：

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

然后让它列出话题，或分析 /tmp/demo.mcap。对于 Claude Code：claude mcp add topicforge -- topicforge。如果客户端无法启动服务（例如从 WSL2 连接到 Windows，或通过 SSH 隧道连接机器人），topicforge --transport streamable-http --port 8765 会在回环地址上提供 http://127.0.0.1:8765/mcp（使用其他 --host 时，需在 TOPICFORGE_HTTP_TOKEN 中设置 bearer 令牌）；请参阅 docs/CLIENTS.md。

其他 MCP 客户端。 Claude Desktop、Cursor、VS Code、Windsurf、Cline、Roo Code、Continue、Zed、JetBrains、Codex、Gemini CLI、Goose 等客户端的即贴即用配置，请参阅 docs/CLIENTS.md。一键安装：添加到 Cursor | 安装到 VS Code。

真实 ROS2 环境（WSL2、Linux、Docker、原生 Windows）的设置说明见 docs/TESTING.md；定期监控提示词和隐私约定见 docs/TUTORIEL.md。

工具

除 health_check 外，每个响应都包含 mode_effective（"live" 或 "mock"），以便调用方区分真实图与模拟数据，并包含一条说明异常情况的 note。health_check.contract_version 为 2。完整参考文档（含所有字段）由 docs/TOOLS.md 生成；从 0.6.x 升级：docs/MIGRATION-0.7.md。

工具	用途
health_check	环境和模式检查。始终成功；在 requested_mode 旁报告 mode
list_topics	探索 ROS2 图
get_topic_info	单个话题的消息类型、发布者／订阅者数量和 QoS
list_nodes	图上的 ROS 2 节点：名称、命名空间、重复名称（调用一次 ros2 node list）
get_node_info	单个节点的发布者、订阅者、服务、动作、参数（机密信息已屏蔽）和 use_sim_time；无响应的节点会被报告，而非报错
inspect_tf	来自 /tf_static 和 /tf 的 TF 树：根节点、过期帧和双父帧、断连的树、适用于仿真时间的时间间隔和频率，以及可选的两帧间路径
graph_diff	自上次正常运行以来发生的变化：对 ROS 2 图（或 bag）进行快照，并与 previous、snap-N ID 或 bag 进行差异比较：节点、话题、类型、写入者 QoS、计数、bag 频率
sample_messages	查看 ROS2 话题的近期消息（数量上限为 50）
analyze_bag	汇总 .mcap / .db3 录制文件或 rosbag2_* 目录（通过 ros2 bag info）
list_participants	域中的 DDS 参与者：厂商、name（EntityName QoS，Cyclone）和 hostname
detect_qos_mismatches	DDS 读取者与写入者之间不兼容的 QoS 配对
peek_dds_samples	原始 DDS 样本；对三个内置发现话题提供结构化数据，对用户话题仅报告是否存在
participant_events	参与者 discovered／lost 事件的时间线
topic_metrics	频率、序列间隙和延迟结构；仅为内置发现话题提供数据
peek_bag_samples	录制 bag 中解码后的样本，包括 ROS 1 .bag（软件包附带 rosbags）
list_endpoints	具有结构化 QoS 的 DDS 写入者和读取者，以及按话题汇总、标记孤立端点（没有读取者的写入者、没有写入者的读取者）

该服务还提供三个 MCP 提示词：diagnose-dds-bus、inspect-ros2-robot 和 diagnose-nav2，为客户端提供只读调用顺序；此外还提供简短的服务 instructions，供不显示提示词的客户端使用（参见 docs/CLIENTS.md）。

examples/ 中提供了针对模拟环境的分步示例，每个示例都包含准确的工具调用和载荷。若要在包含多个程序和厂商的真实总线上运行 DDS 工具，请参阅 examples/dds/README.md（python examples/dds/run_all.py）。

模式

模式	使用场景	后端
mock	开发、演示、CI、录屏	确定性测试数据
live	ROS2 已加载环境并位于 PATH 中，和／或存在 DDS 后端	ros2 CLI、DDS 参与者
auto	检测可用资源，否则使用 mock（默认）	最合适的可用后端

live 和 auto 会降级运行，而不是直接失败：如果 ros2 和 DDS 后端均无法启动，服务会提供 mock 测试数据，并且 health_check 会在 requested_mode 旁报告 mode: "mock"。只有 TOPICFORGE_MODE=mock 会无条件强制使用测试数据。实时 ROS2 适配器会调用 ros2 CLI，因此无需能够导入 rclpy。

DDS 后端

pip install topicforge                           # includes Eclipse CycloneDDS where wheels exist (Python 3.10-3.13; Linux x86_64, Windows x64, macOS)
pip install "topicforge[dds]"                    # alias, kept so older commands work; forces a source build elsewhere (needs the Cyclone C library)
TOPICFORGE_DDS_BACKEND=cyclone python -m topicforge

TOPICFORGE_DDS_BACKEND 接受 mock（默认值）、cyclone、fast 和 auto（依次尝试 fast、cyclone，然后使用能成功导入绑定的后端）。默认的 mock 不会选择任何 DDS 后端：仅安装 Cyclone 绑定还不够，还必须设置 TOPICFORGE_DDS_BACKEND=cyclone。当设置了 TOPICFORGE_MODE=live 但未选择后端时，DDS 工具会报错 DDS module is not active: ...，并说明实际原因（未选择后端、未安装绑定，或绑定已安装但适配器启动失败）；health_check 会报告 dds_backend: "none"，并包含 dds_inactive_note。除 mock 模式外，无论 PATH 中是否存在 ros2，显式设置的值都会生效。如果缺少绑定或参与者无法启动，服务会记录一条指出原因的警告，并仅使用 ROS2 CLI，或回退到 mock 测试数据。如果 ros2 和 DDS 后端均已启动，复合适配器会将八个 ros2 CLI 工具（list_topics、get_topic_info、list_nodes、get_node_info、inspect_tf、graph_diff、sample_messages、analyze_bag）路由到 CLI，并将六个 DDS 工具路由到 DDS 后端；peek_bag_samples 无需 CLI 即可读取 bag。

Fast DDS 适配器已经实现，但从未在总线上运行过，其 fastdds Python 绑定不在 PyPI 上：请从 eProsima 的源码构建，并安装到 TopicForge 旁边。没有 [dds-fast] extra。其他任何值（rti、opensplice、coredx、intercom、opendds、dust）都会在启动时被拒绝，并报错列出有效值；Cyclone 已能通过标准发现机制发现这些厂商的参与者。完整的后端选择、路由表和 QoS 不匹配场景请参阅 docs/DDS_QUICKSTART.md；错误消息请参阅 docs/TROUBLESHOOTING.md。

配置参考

变量	默认值	描述
TOPICFORGE_MODE	auto	mock、live 或 auto
TOPICFORGE_LOG_LEVEL	INFO	DEBUG、INFO、WARNING、ERROR
TOPICFORGE_ROS2_BIN	ros2	ROS2 CLI 二进制文件的名称或路径
TOPICFORGE_TELEMETRY	off	选择启用匿名遥测；无法识别的值会中止启动。参见遥测
TOPICFORGE_DDS_BACKEND	mock	mock（无 DDS 后端）、cyclone、fast、auto；其他值均会被拒绝
TOPICFORGE_DDS_DOMAIN_ID	0	要观察的 DDS 域（0..232）。启动时加入；更改后需要重启
TOPICFORGE_MAX_SAMPLE_BYTES	1048576	单条采样消息的大小上限（1 KiB..64 MiB）；一次调用最多返回上限的 4 倍。超出上限的消息会被丢弃并附带说明

命令行选项（非环境变量）：--transport stdio|streamable-http（默认值为 stdio）、--port N（默认值为 8765）和 --host ADDR（默认值为 127.0.0.1），后两项仅适用于 streamable-http。HTTP 传输会绑定到 127.0.0.1 并启用 DNS 重绑定保护；设置 TOPICFORGE_HTTP_TOKEN（至少 32 个字符）后，每个请求都必须携带 Authorization: Bearer <token>，而使用任何其他 --host 时则必须设置该变量（这是纯 HTTP，因此建议使用 SSH 隧道）；参见 docs/CLIENTS.md。

带注释的示例见 .env.example。任何无效值都会使服务停止，并显示 topicforge: configuration error: ...，退出代码为 2，因此拼写错误不会悄悄改变行为。

限制

针对模拟机器人的真实数据（OmniSim）进行外部验证：docs/VALIDATION.md。

• DDS 验证不完整。Cyclone 适配器已在真实总线上与 Cyclone、Fast DDS 和 Dust DDS 参与者一起运行，并在 Ubuntu 和 Windows 上通过 CI 运行（.github/workflows/demo.yml、.github/workflows/ros2-live.yml）。Fast DDS 适配器（TopicForge 通过 fastdds 绑定接入）从未在总线上运行过，本项目也未观察到任何 RTI、OpenDDS、CoreDX 或 OpenSplice 参与者。跨厂商兼容性的说法基于 RTPS 协议保证，而非有记录的跨厂商运行结果。
• 不会解码用户主题的载荷。对用户主题调用 peek_dds_samples 会返回计数 0，并附注说明该主题已在总线上发布；不会读取任何流量。因此，topic_metrics 仅包含内置发现主题的数据，并会在其 status 中说明这一点。它是发现层探测工具，不是发布速率监视器。
• 不会在运行时观察存活状态。仍存活但保持静默的写入者（租约仍在续期的挂起进程）会显示为健康，因为 TopicForge 读取的是发现信息，而非数据。计划提供可选的数据探测功能。无法区分崩溃与正常离开，lost_ns 是死亡时间的上限。
• Cyclone 厂商 ID：GUID 前缀中的厂商 ID 不遵循 RTPS 约定的参与者（Dust DDS，以及默认配置下的 RTI）会被报告为厂商 unknown。
• 单一域：服务会观察启动时加入的域；更改域需要重启。
• 不处理 DDS Security。没有凭据的参与者看到的是空的安全总线。detect_qos_mismatches 会检查 Partition、类型名称、Reliability、Durability、Deadline、Liveliness、LatencyBudget、Ownership（kind）、DestinationOrder 和 DataRepresentation（将 History 视为风险）；不会检查 Presentation、XTypes 可赋值性和运行时行为，并会在 policies_unchecked 中列出这些项目。它会返回一个 MismatchScan 信封：请查看 reports 获取不匹配项。
• Fast DDS 不提供 list_endpoints。
• sample_messages（实时）会持续运行 ros2 topic echo，直到收到 count 条消息或 timeout_s（1..40，默认值为 10，适用于整个调用）耗尽，并返回收到的消息以及一条 note（N of M messages within T s，说明是否存在发布者）。QoS 会与发布者匹配，因此锁存主题也能正常工作。载荷包含嵌套的命名字段（payload.header.stamp.sec）；timestamp_ns 为 header.stamp（发布者的时钟，在仿真中为仿真时间）、/clock、/tf 和 /rosout 消息正文中的时间（stamp_source 为 payload），或其他无标头类型的 0，并包含 stamp_source 和 received_ns（CLI 打印消息时的墙上时钟时间）。数组默认截断至 128 个元素；max_array_length（1..65536，或设为 null 表示不截断）和 arrays_summary_only 可更改此设置，截断项会列在 _truncated_fields 下。nan 和 inf 会以字符串形式返回。count 超过 50 时会封顶，并附带说明。
• analyze_bag（实时）会解析 ros2 bag info 文本以获取总数和计数；异常检测仅为模拟数据。若能在本地读取包文件（.db3 使用标准库，.mcap 使用 rosbags），则会添加每个主题的时间、速率（(n - 1) / span）和 latched；否则速率会退回到计数 / 包持续时间（frequency_basis）。peek_bag_samples 通过 rosbags 直接读取文件，按录制顺序返回前 count 条消息（而非最后几条），仅由 ROS2 CLI 适配器或模拟模式提供；其 timestamp_ns 为消息的 header.stamp（stamp_source 为 header）、Clock、TFMessage 和 Log 消息正文中的时间（payload），或者在两者都不存在时为包记录时间（stamp_source 为 recorded）；recorded_ns 始终为包记录时间。对于未嵌入消息定义的包（Humble .db3），会使用 Humble 定义或包中记录的发行版定义进行解码，并在 note 中说明。没有 ros2 时，包工具会返回固定样例：在信任包输出之前，请检查 health_check 的 mode: "mock"。health_check 还会报告 sim_clock_published（仅实时模式）：当 /clock 存在发布者时为 true，这提示标头时间戳可能是仿真时间。
• 每个后端的调用会串行执行（一次只能进行一个 ros2 调用，一次只能进行一个 DDS 调用）；每次调用的墙上时钟预算为 45 s，其中包括等待锁的时间；无法获取执行通道的调用会立即以 busy 失败。health_check 和 peek_bag_samples 不获取锁。
• 不支持流式处理或推送订阅：工具严格采用请求/响应模式。 下一步：添加可选探测，以区分写入器挂起还是运行正常；扩大真实总线验证范围（Fast DDS、RTI、OpenDDS）；以及支持 DDS Security。未完成工作见 issues。

遥测

可选、匿名，默认关闭。关闭时，插桩会原样返回处理器：不会构建事件、不会创建传输，也不会运行网络代码（由 tests/test_telemetry.py::test_build_app_off_makes_no_transport_calls 确保）。

TOPICFORGE_TELEMETRY=on python -m topicforge

开启值：on、1、true、yes、enabled。关闭值：未设置、off、0、false、no、disabled。其他值会导致配置错误，而不是静默地视为“关闭”。

开启后，每次工具调用都会发出一个事件，且恰好包含六个字段：

字段	示例	说明
tool_name	"list_topics"	十六种工具之一，绝不包含参数值
latency_ms	12.34	处理器的墙钟耗时，保留 2 位小数
mode	"mock"	实际提供服务的适配器模式：mock 或 live
version	"0.8.1"	TopicForge 服务版本
session_id	"a1b2c3..."	每个进程随机生成的 UUID，绝不持久化
success	true	处理器是否正常返回或引发异常

绝不发送：主题名称、消息类型或负载、bag 路径或内容、主机名、用户名、IP 地址、环境变量、错误消息。字段集合由 tests/test_telemetry.py::test_payload_contains_only_whitelisted_keys 限定；添加字段时必须更新本节。默认传输方式为结构化日志行；目前尚无 HTTP 端点。实现位于 src/topicforge/telemetry/。

安全模型

TopicForge 面向本地可信环境设计：它作为 MCP 客户端的子进程运行在你控制的机器上，并检查你自己的 ROS2 图、DDS 域和 bag 文件。它并未针对恶意输入进行加固。

• TOPICFORGE_ROS2_BIN 接受任意路径；请像对待 PATH 一样对待它。
• analyze_bag 和 peek_bag_samples 会打开客户端传入的任意路径（没有工作区隔离，也没有符号链接限制）。
• 所有 ros2 调用都使用带参数列表的 subprocess.run，绝不使用 shell=True。ROS2 主题名称会先通过 ^/[A-Za-z0-9_/]+$ 验证。
• 服务启动时不会加载第三方代码。
• 除非开启遥测，否则不会进行出站网络调用。

在向不受信任的 MCP 客户端公开 TopicForge（托管端点、共享环境）之前，请添加路径隔离，并重新评估 TOPICFORGE_ROS2_BIN 策略。漏洞报告：请参阅 SECURITY.md。

开发

git clone https://github.com/yaniswav/TopicForge.git && cd TopicForge
python -m venv .venv && source .venv/bin/activate     # Windows: .venv\Scripts\Activate.ps1
pip install -e ".[dev]"
python -m ruff check src tests
python -m pytest

测试针对 mock 适配器、live 适配器的纯解析器以及不依赖绑定的 DDS 辅助工具运行；它们从不需要正在运行的 ROS 图。需要 cyclonedds 或 fastdds 绑定的测试在缺少相应绑定时会自动跳过；integration 标记（真实总线测试）默认取消选择。Makefile（make check）使用 POSIX shell 语法；在纯 PowerShell 中，请运行上面的命令。请参阅 CONTRIBUTING.md。

升级

TopicForge 尚未达到 1.0 版本；CHANGELOG.md 列出了所有变更，包括已撤回的版本和已移除的额外依赖。0.7.0 更改了工具结果的结构（contract_version 2）：docs/MIGRATION-0.7.md 将每个旧字段对应到新字段。

目录结构

src/topicforge/
  server/      MCP bootstrap, build_app(settings), prompts and instructions
  tools/       thin MCPServer handlers, no backend logic
  services/    input validation, orchestration, adapter factory
  adapters/    ros2_live, ros2_mock, dds_cyclone, dds_fast, common/ (binding-free logic)
  models/      frozen Pydantic schemas, the contract with MCP clients
  config/      settings and mode resolution
  telemetry/   opt-in, off by default
examples/      mock walkthroughs (*.md) and runnable live DDS examples (dds/)
scripts/       real-bus interop checks
docs/          guides, product plan
tests/         pytest suite, mock-only, no ROS2 or DDS required

各层严格分离：处理器绝不调用 subprocess，只有适配器代码会与后端交互，新后端需在 adapters/base.py 中实现 MiddlewareAdapter 协议。

许可证

MIT，详见 LICENSE。针对特定 ROS 2 / DDS 配置的集成或支持工作：[email protected]。