TopicForge

TopicForge

AI エージェントが ROS2 グラフ、記録済みの bag ファイル、および ROS の基盤となる DDS 層を調査できる、読み取り専用の MCP（Model Context Protocol）サーバーです。コードには書き込み経路がなく、バスへのパブリッシュやロボットへの指令はできません。また、設定が必要な権限システムもありません。

すべてのツールは、MCP プロトコル上でも読み取り専用であることを宣言します（readOnlyHint、destructiveHint=false、idempotentHint、および適切な openWorldHint）。いずれかのツールにこの宣言がない場合、テストは失敗します。保証内容とその限界については、SECURITY.md を参照してください。

このサーバーは、フリーズされた Pydantic スキーマを返す、型付きツールを16個エージェントに提供します（出力コントラクト 2：各ツールにつき1つのオブジェクト、1つの note、フィールド名に単位を含む。詳しくは docs/CONTRACT.md を参照）。実際のロボットを相手にする場合でも、組み込みのモックフィクスチャを使用する場合でも、スキーマは同一です。nav_planner に scan が届かない理由を尋ねると、エージェントはバスを読み取り、BEST_EFFORT の writer が RELIABLE の reader に接続されていることを見つけ、互換性のないポリシーを特定します（examples/02-debug-qos-mismatch.md を参照）。ROS2 開発者、ロボティクスの ML/CV エンジニア、および本番環境のスタックに書き込み経路を設けることを許容できないチームを対象としています。

DDS については、TopicForge は1つのオープンソースバインディング（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 は同じ標準ディスカバリーを通じて自身を通知しますが、これらはまだ確認されていません。対象となるのはディスカバリーのみで、参加者、reader、writer、およびそれらの QoS を扱います。詳しくは docs/dds-interop-matrix.md を参照してください。

クイックスタート

ROS2 は不要です。モックアダプターが、小型の差動ロボット（LIDAR + 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 にベアラートークンが必要です）。詳しくは 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 が1つ含まれます。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	1つのトピックのメッセージ型、パブリッシャー／サブスクライバー数、QoSを取得します
list_nodes	グラフ上の ROS 2 ノード：名前、名前空間、重複した名前（ros2 node list の呼び出しは1回）
get_node_info	1つのノードのパブリッシャー、サブスクライバー、サービス、アクション、パラメーター（シークレットはマスクされます）、および use_sim_time を取得します。応答しないノードはエラーではなく、報告されます
inspect_tf	/tf_static と /tf の TF ツリー：ルート、古いフレームと親が2つあるフレーム、切断されたツリー、シミュレーション時間に対応した経過時間とレート、2つのフレーム間のオプションのパス
graph_diff	前回の動作時からの変更点：ROS 2 グラフ（またはバッグ）のスナップショットを作成し、previous、snap-N ID、またはバッグとの差分を取得します。対象はノード、トピック、型、書き込み側QoS、件数、バッグのレートです
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 サンプル。3つの組み込みディスカバリートピックでは構造化データ、ユーザートピックでは存在のみを確認します
participant_events	参加者の discovered／lost イベントのタイムライン
topic_metrics	頻度、シーケンス欠落、レイテンシーのスキーマ。データは組み込みディスカバリートピックのみです
peek_bag_samples	記録されたバッグからデコードしたサンプル。ROS 1 .bag も対象です（rosbags はパッケージに含まれています）
list_endpoints	構造化QoSを含む DDS ライターとリーダー、および孤立したエンドポイント（リーダーのいないライター、ライターのいないリーダー）を示すトピックごとの集計

このサーバーは、3つの 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 participant
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 の順で選択）を受け付けます。デフォルトの mock は DDS バックエンドを選択しません。Cyclone バインディングをインストールするだけでは不十分で、TOPICFORGE_DDS_BACKEND=cyclone も設定する必要があります。TOPICFORGE_MODE=live でバックエンドが選択されていない場合、DDS ツールは実際の原因（バックエンドが未選択、バインディングが未インストール、またはバインディングはインストール済みだがアダプターの起動に失敗）を含む DDS module is not active: ... を発生させ、health_check は dds_backend: "none" と dds_inactive_note を報告します。明示的に指定した値は、ros2 が PATH 上にあるかどうかにかかわらず、mock 以外のすべてのモードで尊重されます。バインディングがない場合、または participant を起動できない場合、サーバーは原因を示す警告をログに記録し、ROS2 CLI のみを使用するか、mock フィクスチャにフォールバックします。ros2 と DDS バックエンドの両方が起動している場合、複合アダプターは8つの ros2 CLI ツール（list_topics、get_topic_info、list_nodes、get_node_info、inspect_tf、graph_diff、sample_messages、analyze_bag）を CLI に、6つの DDS ツールを DDS バックエンドに振り分けます。peek_bag_samples は CLI を介さずにバッグを読み取ります。

Fast DDS アダプターは存在しますが、バスに接続して動作したことはありません。また、その fastdds Python バインディングは PyPI にありません。eProsima のソースからビルドし、TopicForge の隣にインストールしてください。[dds-fast] extra はありません。その他の値（rti、opensplice、coredx、intercom、opendds、dust）は、使用可能な値を列挙したエラーとともに起動時に拒否されます。Cyclone は標準のディスカバリーを通じて、これらのベンダーの participant をすでに検出します。バックエンドの選択方法、ルーティングテーブル、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件あたりのサイズ上限（1 KiB～64 MiB）。1回の呼び出しで返されるサイズは最大でその4倍です。上限を超えるメッセージは注記を付けて破棄されます

コマンドラインオプション（環境変数ではありません）：--transport stdio|streamable-http（デフォルトは stdio）、--port N（デフォルトは8765）、--host ADDR（デフォルトは 127.0.0.1）。後者2つは streamable-http の場合にのみ使用できます。HTTP トランスポートは DNS リバインディング対策を有効にして 127.0.0.1 にバインドします。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 アダプター（fastdds バインディングを通じて参加する TopicForge）は、バスに対して実行されたことがなく、RTI、OpenDDS、CoreDX、OpenSplice の参加者も、このプロジェクトでは観測されていません。複数ベンダーに対応するという主張の根拠は、記録されたベンダー間の実行結果ではなく、RTPS プロトコルの保証です。
• ユーザートピックのペイロードはデコードされません。ユーザートピックに対する peek_dds_samples は、カウント 0 と、そのトピックがバス上でアナウンスされていることを示す注記を返します。トラフィックは読み取られません。そのため、topic_metrics がデータを持つのは組み込みのディスカバリートピックのみであり、その旨は status に記載されています。これはディスカバリーレイヤーのプローブであり、パブリッシュレートのモニターではありません。
• 実行時の活性は観測されません。生存していても無通信のライター（リースを更新し続けているハングしたプロセス）は、TopicForge がデータではなくディスカバリーを読み取るため、正常に見えます。オプトインのデータプローブを計画しています。クラッシュと正常な離脱は区別できず、lost_ns は死亡時刻の上限値です。
• Cyclone のベンダー ID：GUID プレフィックスで RTPS のベンダー ID 規約に従わない参加者（Dust DDS、およびデフォルト設定の RTI）は、ベンダー unknown として報告されます。
• ドメインは 1 つのみです。サーバーは起動時に参加したドメインを監視します。ドメインを変更するには再起動が必要です。
• 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（ライブ）は、count 件のメッセージが届くまで、または timeout_s（1～40、デフォルトは 10、呼び出し全体に適用）が経過するまで ros2 topic echo をストリーミングし、届いたメッセージと 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 を超える場合は、注記を付けて 50 に制限されます。
• analyze_bag（ライブ）は、合計と件数について ros2 bag info のテキストを解析します。異常検出はモックのみです。バッグをローカルで読み取れる場合（標準ライブラリを使う .db3、rosbags を使う .mcap）、トピックごとの時刻、レート（(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 呼び出しは一度に 1 件、DDS 呼び出しも一度に 1 件）。各呼び出しにはロック待ち時間を含む 45 秒の壁時計時間の制限があります。レーンを確保できない呼び出しは、直ちに busy で失敗します。health_check と peek_bag_samples はロックを取得しません。
• ストリーミングやプッシュ型のサブスクリプションはありません。ツールは厳密にリクエスト／レスポンス方式です。 次の予定：ハングした書き込み側と正常な書き込み側を見分けるオプトインのプローブ、実際のバスを使ったより広範な検証（Fast DDS、RTI、OpenDDS）、および DDS Security です。未対応の作業はイシューで管理しています。

テレメトリ

オプトイン方式で、匿名で、デフォルトでは無効です。無効の場合、計測処理はハンドラーをそのまま返します。イベントは作成されず、トランスポートも構築されず、ネットワークコードも実行されません（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。これ以外の値は、黙って「無効」とするのではなく、設定エラーになります。

有効にすると、各ツール呼び出しで次の6つのフィールドだけを持つイベントを1つ送信します。

フィールド	例	備考
tool_name	"list_topics"	16個のツールのいずれか。引数の値は含みません
latency_ms	12.34	ハンドラーの実時間。小数点以下2桁
mode	"mock"	実際に処理を行うアダプターのモード：mock または live
version	"0.8.1"	TopicForge サーバーのバージョン
session_id	"a1b2c3..."	プロセスごとにランダムな UUID。永続化されません
success	true	ハンドラーが値を返したか、例外を発生させたか

送信されることがない情報：トピック名、メッセージ型やペイロード、バッグのパスや内容、ホスト名、ユーザー名、IP アドレス、環境変数、エラーメッセージ。フィールドの集合はtests/test_telemetry.py::test_payload_contains_only_whitelisted_keysで制限されています。フィールドを追加する場合は、このセクションも更新してください。デフォルトのトランスポートは構造化ログ行です。HTTP エンドポイントはまだありません。実装はsrc/topicforge/telemetry/にあります。

セキュリティモデル

TopicForge はローカル環境での信頼を前提に設計されています。ユーザーが管理するマシン上で MCP クライアントのサブプロセスとして実行され、ユーザー自身の ROS2 グラフ、DDS ドメイン、バッグファイルを調べます。悪意ある入力への対策は強化されていません。

• TOPICFORGE_ROS2_BINには任意のパスを指定できます。PATHと同様に扱ってください。
• analyze_bagとpeek_bag_samplesは、クライアントが渡した任意のパスを開きます（ワークスペースの分離やシンボリックリンクの制限はありません）。
• すべてのros2呼び出しでは、shell=Trueではなく引数リストを使ってsubprocess.runを実行します。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

テストでは、モックアダプター、ライブアダプターの純粋なパーサー、バインディング不要の DDS ヘルパーを対象とします。ROS グラフの実行は不要です。cycloneddsまたはfastddsバインディングが必要なテストは、バインディングがない場合に自動的にスキップされます。また、integrationマーカー（実際のバスを使うテスト）はデフォルトで選択されません。Makefile（make check）では POSIX シェル構文を使用します。通常の 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] までお問い合わせください。