Kafka Sentinel

kafka-sentinel-mcp

AI エージェントに、Kafka クラスターを安全に読み取り専用で監視する目を与えます。

Kafka クラスターの健全性、コンシューマーのラグ、パーティションの状態、リプレイ可能性を構造化されたツールとして公開するMCP（Model Context Protocol）サーバーです。これにより、LLM エージェント（Claude、または任意の MCP クライアント）が、何かを壊すことなくストリーミング障害を診断できます。

99.999% の可用性で Kafka ベースの金融メッセージングを10年間運用し、「AI + Kafka」のデモがどれも本番環境への書き込みアクセスを前提としていることにうんざりしたエンジニアが開発しました。

このプロジェクトを作った理由

午前3時にコンシューマーグループが停止するとき、疑問はいつも同じです。ラグでしょうか？パーティションが停止しているのでしょうか？リバランスが頻発しているのでしょうか？オフセットのリセットに失敗したのでしょうか？こうした問題はパターンマッチングで解決でき、まさに LLM エージェントが得意とするものです。しかし、本番クラスターの管理者権限をエージェントに渡すオペレーターはいません。

kafka-sentinel-mcp は明確な境界を設けています。すべてのツールは設計上、読み取り専用です。これはクライアント設定レベルで強制されます（管理操作はインポートすらされません）。エージェントは監視、相関分析、推奨を行い、人間が実行します。

ツール

ツール	返される内容
list_topics	すべての非内部トピックと、パーティション数およびレプリケーション係数。トピック名が分からない場合はここから始めます
list_consumer_groups	すべてのコンシューマーグループ ID と状態。グループ名が分からない場合はここから始めます
cluster_health	ブローカー数、コントローラーの状態、レプリカ不足／オフラインのパーティション数
consumer_lag	グループ、トピック、パーティションごとのラグと、コミット済みオフセットと末尾オフセット
topic_audit	レプリケーション係数、min.insync.replicas、保持期間、および耐久性のベストプラクティスに反する設定を示すフラグ
partition_state	リーダー、ISR の縮小、ブローカー間の偏り
replay_readiness	グループとトピックについて、利用可能な最古のオフセットとコミット済みオフセットを比較します。つまり、「未処理分をまだリプレイできるか？」を確認します
incident_snapshot	上記すべてをまとめて1回の呼び出しで取得し、タイムスタンプを付加します。事後分析への貼り付けを想定しています

クイックスタート

pip install kafka-sentinel-mcp   # (or: uv tool install)

# Run against your cluster (read-only credentials!)
KAFKA_BOOTSTRAP=localhost:9092 kafka-sentinel-mcp

Claude Desktop／任意の MCP クライアントに追加します。

{
  "mcpServers": {
    "kafka-sentinel": {
      "command": "kafka-sentinel-mcp",
      "env": { "KAFKA_BOOTSTRAP": "broker1:9092,broker2:9092" }
    }
  }
}

次に、エージェントに尋ねてみましょう。「payments-consumer グループの処理が遅れている理由は何ですか？また、停止した時点からまだリプレイできますか？」

セキュリティ方針

• **構造上、読み取り専用です。**プロデュース、トピック／設定の変更、オフセットのコミット、ACL 操作は行いません。変更用 API は一切インポートされず、CI のテストで実行のたびにサーバーのソースを検索し、この状態を維持します。

• 監視用コンシューマーは enable.auto.commit=False で実行され、コミットは一切行いません。実際のブローカーで検証済みであり、単に主張しているだけではありません。

• SASL／SSL をサポートしています。認証情報は環境変数からのみ読み込まれ、ログに記録されることはありません。

• すべてのツール呼び出しは、パラメーターとともに監査用に記録されます。

• **最小権限：**クラスターとトピックに対する Describe、およびコンシューマーグループに対する Describe のみを持つプリンシパルで実行してください。ACL によって操作が拒否された場合、ツールはスタックトレースではなく、構造化された結果を返します。

  {
    "error": "permission_denied",
    "operation": "list_consumer_groups",
    "detail": "...",
    "hint": "The Kafka principal in use lacks the ACL required for this operation. ..."
  }
  

  これにより、エージェントは動作していないように見せるのではなく、不足している ACL をオペレーターに伝えられます。認可に関係しないエラーは意図的に握りつぶしません。伝播させることで、無関係なエラーが起きた際に黙って機能を低下させ、実際の問題を隠すことを防ぎます。

テスト

pip install -e ".[dev]"

pytest -m "not integration"   # fast, fully mocked — no Docker needed
pytest -m integration         # starts a real Kafka via testcontainers (needs Docker)
pytest                        # both

ユニットテストでは librdkafka を完全にモックし、ツールのロジックを網羅しています。統合テストでは実際のブローカーを起動し、実際のレコードを生成して、ツールが正しいラグ、ISR の状態、耐久性フラグ、リプレイ準備状態を返すことを検証します。また、オブザーバーがコミット済みオフセットを残さないことも確認します。どちらも CI で実行されます。

ステータス

初期段階ですが、テスト済みです。ROADMAP.md をご覧ください。イシューや PR を歓迎します。特に、インシデント中にエージェントから教えてほしかったことについての実体験をお寄せください。

このプロジェクトの引用

学術研究でこのプロジェクトを引用する場合は、CITATION.cff をご覧いただくか、GitHub の「Cite this repository」ボタンをご利用ください。

ライセンス

MIT