mcp-server-celery

mcp-server-celery

実行中の Celery デプロイメントを AI エージェントがデバッグできる MCP サーバーです。稼働中のワーカー、それぞれが処理しているキュー、メッセージが滞留している場所を確認できます。何かが詰まるたびに手動で celery inspect や rabbitmqctl を実行していた確認作業を代わりに行います。

デフォルトでは読み取り専用です。Celery 自身の inspect stats が平文で返すブローカー情報も含め、すべての応答でブローカーのパスワードがマスクされます。

RabbitMQ（amqp://）と Redis（redis://、rediss://）に対応しています。

ツール

ツール	確認できること
list_workers	応答するワーカー、そのプールと同時実行数、稼働時間、プリフェッチ数、処理したタスク数、および各ワーカーが監視するキュー。
queue_depths	各キューで待機しているメッセージ数と、それを処理するワーカー。引数を指定しない場合、設定済みのキューに加えて、稼働中のワーカーが処理するすべてのキューを確認します。

近日追加予定：active_tasks、reserved_tasks、scheduled_tasks、task_status、diagnose_stuck。これらは上記の情報を相互参照し、どのワーカーも監視していないキュー、コンシューマーのいないキューにルーティングされたタスク、応答を停止したワーカーを検出します。ロードマップをご覧ください。

インストール

Python 3.10 以降と uv が必要です。

uvx mcp-server-celery

設定

設定はすべて環境変数で行います。

変数	デフォルト	用途
CELERY_BROKER_URL	amqp://guest:guest@localhost:5672//	確認するブローカー。
CELERY_RESULT_BACKEND	—	結果バックエンド（次のリリースで追加予定の task_status が使用します）。
CELERY_APP	—	アプリのインポートパス。例：proj.celery:app。サーバーがアプリの task_queues とルートを参照できるようになります。両方が設定されている場合も、CELERY_BROKER_URL がアプリのブローカー設定より優先されます。
CELERY_INSPECT_TIMEOUT	1.0	ワーカーからの応答を待つ秒数。大規模または応答の遅いクラスターでは値を大きくしてください。

CELERY_APP はインポート可能である必要があるため、プロジェクトのディレクトリと環境からサーバーを実行してください。

uv run --with mcp-server-celery mcp-server-celery

接続

Claude Code

claude mcp add celery -e CELERY_BROKER_URL=redis://localhost:6379/0 -- uvx mcp-server-celery

Cursor

.cursor/mcp.json に記述します。

{
  "mcpServers": {
    "celery": {
      "command": "uvx",
      "args": ["mcp-server-celery"],
      "env": { "CELERY_BROKER_URL": "redis://localhost:6379/0" }
    }
  }
}

Claude Desktop

claude_desktop_config.json の mcpServers 内に、Cursor と同じ command、args、env ブロックを使用します。

例

エージェントに「メールタスクが実行されないのはなぜ？」と尋ねると、queue_depths を呼び出せます。

{
  "broker": "amqp://app:****@rabbit:5672//",
  "queues": [
    { "queue": "celery", "exists": true, "messages": 0, "consumers": 2, "workers_consuming": ["w1@web", "w2@web"] },
    { "queue": "emails", "exists": true, "messages": 1843, "consumers": 0, "workers_consuming": [] }
  ]
}

emails キューに未処理のタスクがあり、それを処理するワーカーがいません。ワーカーは -Q emails を指定せずに起動されていました。

データに関する注意

• Redis にはコンシューマーという概念がないため、consumers は null になります。ワーカー自身から取得する workers_consuming を使用してください。また、Redis はキューが空になるとそのキーを削除するため、空になったキューは "messages": 0, "exists": null と表示されます。
• RabbitMQ の確認ではパッシブ宣言を使用するため、サーバーがキューを作成することはありません。存在しないキューは "exists": false と表示されます。
• 忙しすぎて CELERY_INSPECT_TIMEOUT 以内に応答できないワーカーは、list_workers に表示されません。想定しているワーカーが見つからない場合は、停止したと判断する前に timeout 引数を大きくしてみてください。

安全性

• すべてのツールに readOnlyHint: true の注釈が付いています。書き込み操作（取り消し、再試行、パージ）は、明示的なオプトインフラグを介してのみ利用できるようになります。
• URL の認証情報と、password／secret／token フィールドは、すべての出力とエラーメッセージでマスクされます。

ロードマップ

v0.1 — list_workers、queue_depths ✅

v1 — active_tasks、reserved_tasks、scheduled_tasks、task_status(task_id)、diagnose_stuck

v2 — summarize_failures：最近の失敗をタスク名と例外タイプごとにグループ化します。Celery はデフォルトでは失敗履歴を保持しないため、イベントリスナーが必要です。

開発

uv sync
uv run pytest tests/unit
uv run ruff check . && uv run ruff format --check .

統合テストでは実際のワーカーを起動するため、ブローカーが必要です。

CELERY_BROKER_URL=redis://localhost:6379/0 uv run pytest tests/integration -m integration

CI では、プッシュのたびに Redis と RabbitMQ を使ってテストを実行します。

リリース

pyproject.toml のバージョンと、server.json の両方の version フィールドを更新してコミットし、vX.Y.Z タグをプッシュします。Release ワークフローでは、3 つのバージョンすべてがタグと一致していることを確認してから、PyPI に公開し、その後 server.json を MCP レジストリに公開します。

ライセンス

MIT