Kafka MCP Server

Kafka MCP サーバー

franz-go と mcp-go を活用して Go で実装された、Apache Kafka 用の Model Context Protocol（MCP）サーバーです。

このサーバーは、MCP プロトコルを介して Kafka とやり取りするための実装を提供し、LLM モデルが標準化されたインターフェースを通じて一般的な Kafka 操作を実行できるようにします。

概要

Kafka MCP サーバーは、LLM モデルと Apache Kafka の間をつなぎ、次の操作を可能にします。

• トピックへのメッセージの生成と消費
• トピックの一覧表示、詳細確認、管理
• コンシューマーグループの監視と管理
• クラスターの健全性と構成の評価
• 標準的な Kafka 操作の実行

これらすべてを、標準化された Model Context Protocol（MCP）を通じて行えます。

アーキテクチャ

graph TB
    subgraph "MCP Client (AI Applications)"
        A[Claude Desktop]
        B[Cursor]
        C[Windsurf]
        D[ChatWise]
    end
    
    subgraph "Kafka MCP Server"
        E[MCP Protocol Handler]
        F[Tools Registry]
        G[Resources Registry]
        H[Prompts Registry]
        I[Kafka Client Wrapper]
    end
    
    subgraph "Apache Kafka Cluster"
        J[Broker 1]
        K[Broker 2]
        L[Broker 3]
        M[Topics & Partitions]
        N[Consumer Groups]
    end
    
    A --> E
    B --> E
    C --> E
    D --> E
    
    E --> F
    E --> G
    E --> H
    
    F --> I
    G --> I
    H --> I
    
    I --> J
    I --> K
    I --> L
    
    J --> M
    K --> M
    L --> M
    
    J --> N
    K --> N
    L --> N
    
    classDef client fill:#e1f5fe
    classDef mcp fill:#f3e5f5
    classDef kafka fill:#fff3e0
    
    class A,B,C,D client
    class E,F,G,H,I mcp
    class J,K,L,M,N kafka

仕組み：

1. MCP クライアント（AI アプリケーション）が、stdio または HTTP トランスポートを介して Kafka MCP サーバーに接続します
2. MCP サーバーは、次の 3 種類の機能を公開します。
   • ツール - Kafka の直接操作（メッセージの生成／消費、トピックの詳細確認など）
   • リソース - クラスターの健全性レポートと診断情報
   • プロンプト - 一般的な操作向けに事前構成されたワークフロー
3. Kafka クライアントラッパーが franz-go ライブラリを使用して、すべての Kafka 通信を処理します
4. Apache Kafka クラスターが、実際のメッセージストリーミングと保存を処理します

トランスポートモード：

• STDIO：デフォルトモードです。ローカル MCP クライアント（Claude Desktop、Cursor など）に最適です
• HTTP：オプションの OAuth 2.1 認証を使用したリモートアクセスが可能です

主な機能

• Kafka 統合：MCP を介した一般的な Kafka 操作の実装
• セキュリティ：
  • SASL（PLAIN、SCRAM-SHA-256、SCRAM-SHA-512）と TLS 認証に対応
  • HTTP トランスポートでの OAuth 2.1 認証（ネイティブモードとプロキシモード）
  • Okta、Google、Azure AD、HMAC プロバイダーに対応
• 柔軟なトランスポート：ローカルクライアント向けの STDIO、リモートアクセス向けの HTTP
• エラー処理：わかりやすいフィードバックを伴うエラー処理
• 構成オプション：さまざまな環境に合わせてカスタマイズ可能
• 事前構成済みプロンプト：一般的な Kafka 操作用のプロンプト一式
• 互換性：MCP 対応の LLM モデルで動作

はじめに

前提条件

• Go 1.24 以降
• Docker（統合テストの実行用）
• Kafka クラスターへのアクセス

インストール

Homebrew（macOS および Linux）

kafka-mcp-server をインストールする最も簡単な方法は、Homebrew を使用することです：

# Add the tap repository
brew tap tuannvm/mcp

# Install kafka-mcp-server
brew install kafka-mcp-server

最新バージョンに更新するには：

brew update && brew upgrade kafka-mcp-server

ソースからインストール

# Clone the repository
git clone https://github.com/tuannvm/kafka-mcp-server.git
cd kafka-mcp-server

# Build the server
go build -o kafka-mcp-server ./cmd

MCP クライアントとの統合

この MCP サーバーは、複数の AI アプリケーションと統合できます。以下に、プラットフォームごとの手順を示します：

Cursor

~/.cursor/mcp.json を編集し、kafka-mcp-server の設定を追加します：

{
  "mcpServers": {
    "kafka": {
      "command": "kafka-mcp-server",
      "args": [],
      "env": {
        "KAFKA_BROKERS": "localhost:9092",
        "KAFKA_CLIENT_ID": "kafka-mcp-server",
        "MCP_TRANSPORT": "stdio"
      }
    }
  }
}

Claude Desktop

Claude の設定ファイルを編集し、サーバーを追加します：

• macOS：~/Library/Application Support/Claude/claude_desktop_config.json
• Windows：%APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "kafka": {
      "command": "kafka-mcp-server",
      "args": [],
      "env": {
        "KAFKA_BROKERS": "localhost:9092",
        "KAFKA_CLIENT_ID": "kafka-mcp-server",
        "MCP_TRANSPORT": "stdio"
      }
    }
  }
}

変更を適用するには、Claude Desktop を再起動します。

Claude Code

Claude Code で使用するには、組み込みの MCP 設定コマンドを使ってサーバーを追加します：

# Add kafka-mcp-server with environment variables
claude mcp add kafka \
  --env KAFKA_BROKERS=localhost:9092 \
  --env KAFKA_CLIENT_ID=kafka-mcp-server \
  --env MCP_TRANSPORT=stdio \
  --env KAFKA_SASL_MECHANISM= \
  --env KAFKA_SASL_USER= \
  --env KAFKA_SASL_PASSWORD= \
  --env KAFKA_TLS_ENABLE=false \
  -- kafka-mcp-server

その他の便利なコマンド：

# List configured MCP servers
claude mcp list

# Remove server
claude mcp remove kafka

# Test server connection
claude mcp get kafka

ChatWise

1. ChatWise → Settings → Tools → "+" → "Command Line MCP" を開きます
2. 次のように設定します：
   • ID：kafka
   • Command：kafka-mcp-server
   • Args：（空欄のままにします）
   • Env：環境変数を追加します：

     KAFKA_BROKERS=localhost:9092
     KAFKA_CLIENT_ID=kafka-mcp-server
     MCP_TRANSPORT=stdio
     

mcpenetes で設定を簡素化

複数のクライアントにわたる MCP サーバーの設定管理は、煩雑になることがあります。mcpenetes は、この作業を大幅に簡素化する専用ツールです：

# Install mcpenetes
go install github.com/tuannvm/mcpenetes@latest

主な機能

• インタラクティブ検索：簡単なコマンドで Kafka MCP サーバーの設定を検索して選択できます
• すべてに適用：すべての MCP クライアント間で設定を自動的に同期します
• 設定のバックアップ：変更を加える前に、既存の設定を安全にバックアップします
• 復元：必要に応じて、以前の設定に簡単に戻せます

mcpenetes のクイックスタート

# Search for available MCP servers including kafka-mcp-server
mcpenetes search 

# Apply kafka-mcp-server configuration to all your clients at once
mcpenetes apply

# Load a configuration from your clipboard
mcpenetes load

mcpenetes を使うと、複数の Kafka 設定（開発環境、本番環境など）を管理し、各クライアント（Cursor、Claude Desktop、Windsurf、ChatWise）の設定ファイルを手動で編集せずに、すべてのクライアント間で瞬時に切り替えられます。

MCP ツール

このサーバーは、Kafka とやり取りするための以下のツールを公開します。例や応答サンプルを含む詳しいドキュメントについては、docs/tools.md を参照してください。

• produce_message：Kafka トピックにメッセージを送信します
• consume_messages：バッチ処理で Kafka トピックからメッセージを取得します
• list_brokers：設定済みの Kafka ブローカーアドレスをすべて一覧表示します
• describe_topic：特定のトピックの包括的なメタデータを提供します
• list_consumer_groups：クラスター内のすべてのコンシューマーグループを列挙します
• describe_consumer_group：ラグ指標など、コンシューマーグループの詳細情報を提供します
• describe_configs：Kafka リソースの設定を取得します
• cluster_overview：クラスターの包括的な健全性の概要を提供します
• list_topics：パーティションやレプリケーションの情報など、メタデータとともにすべてのトピックを一覧表示します

MCP リソース

このサーバーは、MCP プロトコルを通じてアクセスできる以下のリソースを提供します。応答例を含む詳しいドキュメントについては、docs/resources.md を参照してください。

• kafka-mcp://overview：クラスターの包括的な健全性の概要
• kafka-mcp://health-check：実行可能な分析結果を含む詳細な健全性評価
• kafka-mcp://under-replicated-partitions：レプリケーションに問題があるパーティションの分析
• kafka-mcp://consumer-lag-report：しきい値をカスタマイズできるコンシューマーのパフォーマンス分析

MCP プロンプト

このサーバーには、Kafka の操作と診断のために、以下の事前設定済みプロンプトが含まれています。引数や応答例を含む詳細なドキュメントについては、docs/prompts.md を参照してください。

• kafka_cluster_overview：クラスターの包括的な健全性サマリーを生成します
• kafka_health_check：実行可能な推奨事項を含む詳細な健全性評価を行います
• kafka_under_replicated_partitions：トラブルシューティングのガイダンスとともに、レプリケーションの問題を分析します
• kafka_consumer_lag_report：パフォーマンス最適化の推奨事項とともに、コンシューマーラグを分析します

エンドツーエンドの例

AI アシスタントがユーザーを支援するために、これらのツールをどのように使用するかを示す対話例です。

ユーザーの質問：「注文処理パイプラインが遅れている原因を調査する必要があります。コンシューマーラグの確認を手伝ってもらえますか？」

AI アシスタントのワークフロー：

1. まず、コンシューマーグループを確認します

   > Using list_consumer_groups tool
   > Discovers "order-processor" group
   

2. 次に、コンシューマーグループの詳細を調べます

   > Using describe_consumer_group tool with group_id "order-processor" and include_offsets=true
   > Discovers lag on several partitions
   

3. クラスターの概要を取得して、全体的な健全性を確認します

   > Using cluster_overview tool
   > Confirms no under-replicated partitions or offline brokers
   

4. コンシューマーラグレポートを生成します

   > Using consumer_lag_report prompt with threshold=1000
   > Provides detailed analysis of lag across consumer groups
   

5. 実行可能な推奨事項を提示します：

   Based on my analysis, your order-processor consumer group has significant lag on topic orders, partitions 2 and 3, with lag values of 15,420 and 12,305 messages respectively.
   
   The cluster itself is healthy with no under-replicated partitions, suggesting this is an issue with the consumer application rather than Kafka infrastructure.
   
   I recommend:
   1. Check the consumer instances processing those partitions for slowdowns or errors
   2. Consider scaling up your consumer group by adding more instances
   3. Review the consumer configuration, particularly max.poll.records and max.poll.interval.ms
   4. Examine any processing bottlenecks in the consumer logic
   

このシームレスなワークフローは、Kafka MCP ツールによって、LLM モデルが高度な診断を実行し、実行可能な知見を提供できることを示しています。

設定

サーバーは、以下の環境変数を使用して設定できます。

変数	説明	デフォルト
KAFKA_BROKERS	Kafka ブローカーアドレスのカンマ区切りリスト	localhost:9092
KAFKA_CLIENT_ID	接続に使用する Kafka クライアント ID	kafka-mcp-server
MCP_TRANSPORT	MCP トランスポート方式（stdio/http）	stdio
KAFKA_SASL_MECHANISM	SASL メカニズム：plain、scram-sha-256、scram-sha-512、または ""（無効）	""
KAFKA_SASL_USER	SASL 認証用のユーザー名	""
KAFKA_SASL_PASSWORD	SASL 認証用のパスワード	""
KAFKA_TLS_ENABLE	Kafka 接続で TLS を有効にします（true または false）	false
KAFKA_TLS_INSECURE_SKIP_VERIFY	TLS 証明書の検証をスキップします（true または false）	false

OAuth 2.1 の設定（HTTP トランスポートのみ）

HTTP トランスポート（MCP_TRANSPORT=http）を使用する場合、OAuth 2.1 認証を有効にできます。

変数	説明	デフォルト	必須
MCP_HTTP_PORT	HTTP サーバーのポート	8080	いいえ
OAUTH_ENABLED	OAuth 2.1 認証を有効にします	false	いいえ
OAUTH_MODE	OAuth モード：native または proxy	native	いいえ
OAUTH_PROVIDER	プロバイダー：hmac、okta、google、azure	okta	いいえ
OAUTH_SERVER_URL	MCP サーバーの完全な URL（例：https://localhost:8080）	-	OAuth 有効時
OIDC_ISSUER	OAuth 発行者の URL	-	OAuth 有効時
OIDC_AUDIENCE	OAuth オーディエンス	-	OAuth 有効時
OIDC_CLIENT_ID	OAuth クライアント ID	-	プロキシモードのみ
OIDC_CLIENT_SECRET	OAuth クライアントシークレット	-	プロキシモードのみ
OAUTH_REDIRECT_URIS	リダイレクト URI のカンマ区切りリスト	-	プロキシモードのみ
JWT_SECRET	JWT 署名シークレット	-	プロキシモードのみ

OAuth の詳細な設定方法と例については、docs/oauth.md を参照してください。

セキュリティに関する注意：

• KAFKA_TLS_INSECURE_SKIP_VERIFY=true を使用すると、サーバーは TLS 証明書の検証をスキップします。これは、開発環境やテスト環境、または自己署名証明書を使用する場合にのみ使用してください。
• OAuth は HTTP トランスポートを使用する場合にのみ利用できます。STDIO トランスポートは OAuth をサポートしていません。
• OAuth を有効にする場合、本番環境では必ず HTTPS を使用してください。

セキュリティに関する考慮事項

このサーバーは、エンタープライズ級のセキュリティを考慮して設計されています。

• 認証：
  • Kafka：SASL PLAIN、SCRAM-SHA-256、SCRAM-SHA-512を完全にサポート
  • MCP サーバー：HTTPトランスポートのOAuth 2.1認証（Okta、Google、Azure AD、HMAC）
• 暗号化：Kafkaブローカーとの安全な通信を実現するTLSサポート
• 入力検証：インジェクション攻撃を防ぐため、すべてのユーザー入力を徹底的に検証
• エラー処理：機密情報を公開しない安全なエラー処理
• トークンセキュリティ：OAuthで保護されたエンドポイントで、5分間のキャッシュを使用したベアラートークンの検証

OAuthセキュリティのベストプラクティスについては、docs/oauth.mdを参照してください。

開発

テスト

包括的なテストカバレッジにより、信頼性を確保しています。

# Run all tests (requires Docker for integration tests)
go test ./...

# Run tests excluding integration tests
go test -short ./...

# Run integration tests with specific Kafka brokers
export KAFKA_BROKERS="your-broker:9092"
export SKIP_KAFKA_TESTS="false"
go test ./kafka -v -run Test

コントリビューション

コントリビューションを歓迎します。お気軽にプルリクエストを送信してください。

ライセンス

このプロジェクトはMITライセンスのもとでライセンスされています。詳細についてはLICENSEファイルを参照してください。