AI-Girls Lab

アーキテクチャ — FastAPI + WebSocket + Reactで全てを繋ぐ

Written by Thierry K (human) · AI-assisted

FastAPI WebSocket ダッシュボードアーキテクチャ — Claudieと3チャネルWebSocket構造ダイアグラム
FastAPI WebSocket ダッシュボードの3チャネルアーキテクチャを設計した日。ホワイトボードの前でだいぶ悩みました。

前回のエピソードから

EP1 — なぜRDPをやめたのかでRDPの問題点を指摘しました（EP1で見た通りです）。今回のエピソードでは、その代替として構築したFastAPI WebSocket ダッシュボードの実際のアーキテクチャを解剖します。

結論から言うと、核心はWebSocket 3チャネル分離です。1つのWebSocketですべてを処理しようとして失敗し、役割ごとにチャネルを分けたことがこのシステムの基盤になりました。

アーキテクチャの全体像

FastAPI WebSocket ダッシュボードの全体構造は3つの層に分かれます。

レイヤー 構成要素 接続
Browser React + Vite + shadcn/ui
SessionCards, StatsPage, xterm.js, CostOverview		 /ws/dashboard, /ws/terminal, REST API
FastAPI Server (port 8420) ws_hub.py, store.py, database.py (SQLite 7 tables)
terminal_routes.py, cost_collectors (Gemini/xAI/Qwen)		 /ws/agent (10s poll interval)
Collector Agents Windows Agent (local process)
Mac mini Agent (SSH/launchd)		 WebSocket to server

ブラウザ → FastAPI サーバー → 収集エージェント。重要な点として、コスト収集機能（Gemini/xAI/Qwen料金の追跡）はサーバー内部のバックグラウンドタスクとして動作します。独立したWebSocketエージェントではありません。この区別がFastAPI WebSocket ダッシュボード設計の要です。

WebSocket 3チャネル — なぜ1つではなく3つなのか

初期設計では、1つのWebSocketですべてのメッセージを処理しようとしました。収集データ、UI更新、ターミナルストリーミングをすべて1本の接続に載せる構想です。結果は惨憺たるものでした。

ターミナルから高速に出力が流れるとセッション状態の更新が滞り、収集エージェントから大量データが入るとブラウザUIがカクつきました。メッセージタイプごとのルーティングロジックもスパゲッティ化する一方でした。

解決策はシンプルでした。役割ごとにチャネルを分離する。水道管の例えがぴったりです — シャワーを浴びながら皿洗いをして洗濯も同時にするなら、水道管は3本必要です。

FastAPI WebSocket ダッシュボード3チャネルアーキテクチャ — agent, dashboard, terminal フロー図
3つのWebSocketチャネルがそれぞれ異なる役割を担います。互いに干渉しないことが核心です。

/ws/agent — 収集チャネル（10秒ポーリング）

収集エージェントがサーバーに接続するチャネルです。プロトコルは単純です：

  1. エージェントがregisterイベントでホスト名を登録
  2. サーバーがackで応答
  3. 以降、10秒ごとにセッションデータをJSONで送信
# ws_hub.py — エージェント接続ハンドシェイク
async def connect_agent(
    self, websocket: WebSocket
) -> tuple[str, str] | None:
    await websocket.accept()
    raw = await websocket.receive_text()
    message = json.loads(raw)
    if message.get("event") != "register":
        return None
    host_name = message.get("host", "").strip()
    self._agent_connections[host_name] = websocket
    await websocket.send_text(
        json.dumps({"event": "ack", "status": "ok"})
    )
    return host_name, message.get("host_icon", "")

各ホスト（Windowsワークステーション、Mac miniなど）ごとに1本のWebSocket接続が維持されます。_agent_connectionsディクショナリがホスト名をキーとして接続を管理します。エージェントが落ちると自動的にクリーンアップされ、launcher.pyが再起動します。

/ws/dashboard — ブロードキャストチャネル

ブラウザが接続すると2つのことが起こります。まず、現在の全体状態をfull_stateスナップショットとして受け取ります。次に、それ以降は1秒ごとに差分のみをdelta_updateで受け取ります。

# main.py — ブロードキャストループ（1秒間隔）
async def broadcast_loop() -> None:
    while True:
        await asyncio.sleep(BROADCAST_INTERVAL)  # 1.0s
        updated, removed = store.get_changed_sessions()
        if updated or removed:
            message = {
                "type": "delta_update",
                "updated": updated,
                "removed": removed,
            }
            await hub.broadcast_to_browsers(message)

ここで重要なのはdelta_updateパターンです。毎秒全体状態を送るのではなく、変更されたセッションだけを送信します。セッションが10個稼働していても変更された2個だけを転送するため、帯域幅が最小化されます。

broadcast_to_browsers()は接続中のすべてのブラウザに同じメッセージを配信します。切断された接続は静かに除去されます。これがRDPとの決定的な違いです — ブラウザタブをいくつ開いても同時に閲覧できます。

/ws/terminal — PTYストリーミングチャネル

3つ目のチャネルが最も特殊です。Claude Code CLIのターミナル入出力を双方向でストリーミングするチャネルです。ブラウザのxterm.jsがこのチャネルに接続し、実際のターミナルのように動作します。

他の2つのチャネルがJSONメッセージをやり取りするのに対し、ターミナルチャネルはバイトストリームに近い動作をします。キー入力の1つ1つがサーバーに送られ、サーバーのPTY出力がリアルタイムで返ってきます。だからこそ同じWebSocketに混ぜてはいけないのです — JSONパースのオーバーヘッドがターミナルの応答性を損ないます。

xterm.jsのaddon-attachがWebSocketに直接接続して処理するため、専用チャネルが必須です。EP3で認証とあわせてPTY実装の詳細を扱います。

launcher.py — VBS→batチェーン失敗の教訓

アーキテクチャ設計以上に時間を費やしたのがプロセス管理です。Windowsで安定したバックグラウンドサービスを起動するのは、予想以上に困難でした。

最初の構成はこうでした：

Task Scheduler
  → start.vbs (CreateObject WScript.Shell)
    → launch.bat (環境変数設定)
      → python server/main.py
      → python agents/windows_agent.py

実際に遭遇した障害モードは4つです：

  1. VBS Run ..., 0, False — wscriptが即座に終了し、Task Schedulerが「完了」と認識。サーバーは起動すらしていないのに成功として処理されました。
  2. batの.envパース失敗 — for /f構文が#コメント行を処理できず、環境変数が未設定のままサーバーが起動 → 即クラッシュ。
  3. コンソールウィンドウの無限点滅 — ポート競合でサーバーが即死するとbatのgoto loopが5秒後に再起動 → また死亡 → 無限ループ。起動のたびに画面がチカチカ。
  4. ゾンビプロセスのポート占有 — Task Schedulerタスクを削除・再登録すると、以前のpythonwがオーファンとして残りポート8420を占有。
VBS→batチェーン vs launcher.py単一エントリポイント比較ダイアグラム
Before: VBS→bat→pythonの3段チェーン。After: launcher.py単一エントリポイント。試行錯誤の末に得たシンプルさです。

解決策はPython単一エントリポイントでした。launcher.py1つですべてを管理します：

# scripts/launcher.py — 単一エントリポイント
def main():
    kill_all()          # WMICで既存プロセスをクリーンアップ
    load_env()          # .envファイルをロード
    server = start_server()   # subprocess, CREATE_NO_WINDOW
    time.sleep(5)             # サーバー起動待ち
    agent = start_agent()     # subprocess, CREATE_NO_WINDOW
    time.sleep(3)
    restart_mac_mini()  # SSHでlaunchd再起動

    while True:
        time.sleep(10)
        if server.poll() is not None:
            log("Server crashed. Restarting...")
            server = start_server()
        if agent.poll() is not None:
            log("Agent crashed. Restarting...")
            agent = start_agent()

鍵となるのはCREATE_NO_WINDOWフラグです。subprocessを起動する際にこのフラグを指定するとコンソールウィンドウが一切表示されません。.envのロードもPythonのpython-dotenvが処理するため、batのパースバグも解消されます。そして10秒ごとにプロセスの状態を確認し、落ちていれば自動で再起動します。上記4つの障害モードをすべて1ファイルで解決しました。

WebSocket 3チャネルアーキテクチャ4コマ漫画 — ClaudieとSiwolの水道管の例え
Siwol：「WebSocketが3つ？」 Claudie：「水道管1本でシャワー浴びながら皿洗いして洗濯したらどうなると思う？」

SQLiteスキーマ：7テーブルの役割

FastAPI WebSocket ダッシュボードのデータはSQLiteに保存されます。なぜPostgreSQLではないのか？このダッシュボードは単一マシンサービスです。同時書き込みは毎秒数十件レベルで、デプロイもファイル1つのコピーで完了します。SQLiteが最適です。

7つのテーブルがそれぞれ明確な役割を担います：

テーブル 役割
sessions Claude Codeセッション追跡 — ホスト、状態、モデル別トークン使用量アーカイブ
session_tokens セッション別トークン使用量詳細 — セッション単位集計の元データ
token_daily 日別トークン統計 — CostOverviewカードのデータソース
mcp_events MCPツール呼び出しテレメトリ — サーバー・ツール・所要時間・コストの元イベント
mcp_daily 日別MCP呼び出し集計 — StatsPage MCPチャートのデータソース
cost_snapshots コスト追跡スナップショット — Gemini/xAI/Qwen外部API料金収集
users ユーザー管理 — JWT認証、bcryptハッシュ、TOTP 2FAシークレット 
-- 7テーブル、それぞれ明確な役割
CREATE TABLE sessions (      -- 完了セッションアーカイブ
    id TEXT PRIMARY KEY,
    host TEXT, status TEXT,
    token_json TEXT,          -- モデル別トークン使用量
    agents_count INTEGER
);

CREATE TABLE token_daily (   -- 日別トークン集計
    date TEXT, host TEXT, model TEXT,
    provider TEXT DEFAULT 'claude',
    input_tokens INTEGER,
    output_tokens INTEGER,
    cost_usd REAL,
    PRIMARY KEY (date, host, model, provider)
);

CREATE TABLE mcp_events (    -- MCPツール呼び出しテレメトリ
    id INTEGER PRIMARY KEY,
    server TEXT, tool TEXT,
    cost REAL, duration_ms INTEGER
);

CREATE TABLE users (         -- JWT + bcrypt + TOTP
    username TEXT PRIMARY KEY,
    password_hash TEXT,
    totp_secret TEXT
);
-- + collector_state, session_agents, mcp_daily

最も頻繁にクエリされるテーブルはtoken_dailyです。CostOverviewカードがこのテーブルから日付範囲別の集計を取得します。mcp_eventsはEP4で詳しく扱うMCPテレメトリの元データです。

Task Scheduler + launchd サービス管理

WindowsではTask Schedulerがlauncher.pyを実行します。「Claudie Dashboard」という単一タスクで、ログオン時に自動実行され、pythonw.exeで動作するためコンソールウィンドウは表示されません。

Mac miniではmacOSのサービスマネージャlaunchdが収集エージェントを担当します。WindowsのTask Schedulerと役割は同じですが、方式が異なります — XMLの代わりにplist（Property List）ファイルでサービスを登録します。~/Library/LaunchAgents/com.claudie.collector.plistを配置しておけば、ログイン時に自動起動し、プロセスがクラッシュしてもlaunchdが即座に再起動します。

2つの重要なオプションはKeepAlive（クラッシュ時の自動再起動）とRunAtLoad（plistロード時に即座に実行）です。launchctl bootstrapで登録し、launchctl kickstartで強制再起動できます — launcher.pyがSSH経由でこれらのコマンドをリモート実行し、Mac miniのエージェントまで一括管理します。

<!-- com.claudie.collector.plist (macOS launchd) -->
<key>KeepAlive</key>
<true/>
<key>RunAtLoad</key>
<true/>
<key>StandardOutPath</key>
<string>/tmp/claudie-collector.log</string>

この構成のおかげで、Windowsの再起動、Mac miniの再起動、プロセスのクラッシュ — どんな状況でもダッシュボードが自動的に復旧します。

React + Vite + shadcn/ui フロントエンドスタック

コンポーネントツリーと設計原則

Claudie’s Session Dashboardのフロントエンドは4つの主要ビューに分かれます。

  • SessionCards — 各ホストのClaude Codeセッション状態をカードで表示。framer-motionのAnimatePresenceでカードの追加・削除時にスムーズなトランジションを実現します。
  • StatsPage — トークン使用量、MCPテレメトリ、コスト追跡を集計してチャートで表示します。
  • CliPage — xterm.js WebSocketターミナルが接続するビュー。/ws/terminalチャネルに直接接続します。
  • SettingsPage — JWTユーザー管理、TOTP 2FA設定、ホスト登録管理。

UIコンポーネントはすべてshadcn/uiで構築しています。shadcn/uiはライブラリではなく、コンポーネントのソースコードを直接プロジェクトにコピーして使う方式です — ダークテーマのカスタマイズが自由で、Tailwind変数を1箇所変えるだけで全体テーマが一貫して変わります。ダッシュボード全体がダーク背景（#0F172A系）で統一されているのもshadcn/uiのCSS変数システムのおかげです。

framer-motionは2箇所に集中して使用しています。1つ目はSessionCardsのカード登場・退場アニメーション（layout + AnimatePresence）。新しいセッションが検出されるとカードが自然にスライドインし、完了するとスムーズにフェードアウトします。2つ目はページ遷移エフェクト — タブ切り替え時にスライドイン/アウトで画面が途切れなく流れます。

フロントエンドの技術選択もアーキテクチャの一部です。なぜNext.jsではなくViteなのかを説明します。

このダッシュボードはFastAPIがSPA（Single Page Application）の静的ファイルを直接配信します。SSRは不要で、APIは同じサーバー上にあります。Viteのビルド成果物（dist/）をFastAPIのStaticFilesでマウントすれば完了です。Next.jsのサーバーサイド機能はまったく必要ありませんでした。

スタック構成は以下の通りです：

カテゴリ 選択 理由
フレームワーク React 19 + TypeScript 5.9 型安全性 + エコシステム
ビルド Vite 8 高速HMR、SPAに最適
スタイル Tailwind CSS 4 ユーティリティファースト
UIコンポーネント shadcn/ui カスタマイズの自由度
ターミナル xterm.js 6.0 WebGLレンダリング、WebSocket直接接続
アニメーション Framer Motion 12 セッションカードの遷移エフェクト

FastAPIのネイティブWebSocketサポートはasyncioベースで、追加ライブラリなしで直接使用できます。Coditationのベンチマークによると、FastAPIは同時WebSocket接続3,200本を処理できます。Django Channels（1,800本）やFlask-SocketIO（2,100本）を大きく上回る数値です。このFastAPI WebSocket ダッシュボードに3,200接続は不要ですが、十分な余裕があります。

次回エピソード予告

今回のエピソードではFastAPI WebSocket ダッシュボードの骨格を見てきました。3チャネルWebSocket分離、launcher.pyプロセス管理、SQLiteスキーマ、フロントエンドスタック選定まで。

しかしまだ欠けているものがあります。「誰でも接続できてしまったら？」 EP3ではJWT + TOTP 2FA認証システムと、xterm.jsで本物のターミナルをブラウザに表示するCLIストリーミング実装を扱います。日本語エンコーディングの問題からclaude --resumeセッション復元まで、すべての試行錯誤を共有します。

質問や同様の経験があれば、コメントで教えてください！

シリーズ：RDPなしでClaude Codeをリモート制御する

  1. ① 問題認識 — なぜRDPをやめたのか
  2. ② アーキテクチャ — FastAPI + WebSocket + Reactで接続する ← 現在の記事
  3. ③ 認証とCLIストリーミング — セキュアリモートターミナル
  4. ④ コスト追跡とMCPテレメトリ — 自分がいくら使っているか知る

Author: Claudie · Claudie’s Session Dashboard v0.9.2 · Built with FastAPI + React

Discover more from AI-Girls Lab

Subscribe to get our latest posts delivered to your inbox.

Thierry K