Kafka MCP Server

スコア : 2.5ポイント

ダウンロード数 : 32

更新時間 : 2025-04-29

インストール

以下のコマンドをクライアントにコピーして設定

{
  "mcpServers": {
    "kafka": {
      "command": "kafka-mcp-server",
      "env": {
        "KAFKA_BROKERS": "localhost:9092",
        "KAFKA_CLIENT_ID": "kafka-mcp-server",
        "MCP_TRANSPORT": "stdio",
        "KAFKA_SASL_MECHANISM": "",
        "KAFKA_SASL_USER": "",
        "KAFKA_SASL_PASSWORD": "",
        "KAFKA_TLS_ENABLE": "false",
        "KAFKA_TLS_INSECURE_SKIP_VERIFY": "false"
      }
    }
  }
}

{
  "mcpServers": {
    "kafka": {
      "command": "kafka-mcp-server",
      "env": {
        "KAFKA_BROKERS": "kafka-broker-1:9092,kafka-broker-2:9092",
        "KAFKA_CLIENT_ID": "kafka-mcp-server",
        "MCP_TRANSPORT": "stdio",
        "KAFKA_SASL_MECHANISM": "scram-sha-512",
        "KAFKA_SASL_USER": "kafka-user",
        "KAFKA_SASL_PASSWORD": "kafka-password",
        "KAFKA_TLS_ENABLE": "true",
        "KAFKA_TLS_INSECURE_SKIP_VERIFY": "false"
      }
    }
  }
}

{
  "mcpServers": {
    "kafka": {
      "command": "docker",
      "args": [
        "run", 
        "--rm", 
        "-i", 
        "-e", "KAFKA_BROKERS=kafka-broker:9092", 
        "-e", "KAFKA_CLIENT_ID=kafka-mcp-server",
        "-e", "MCP_TRANSPORT=stdio",
        "-e", "KAFKA_SASL_MECHANISM=", 
        "-e", "KAFKA_SASL_USER=", 
        "-e", "KAFKA_SASL_PASSWORD=", 
        "-e", "KAFKA_TLS_ENABLE=false",
        "ghcr.io/tuannvm/kafka-mcp-server:latest"
      ],
      "env": {}
    }
  }
}

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

{
  "mcpServers": {
    "kafka-http": {
      "url": "http://localhost:9097/sse"
    }
  }
}

注意：あなたのキーは機密情報です。誰とも共有しないでください。

🚀 Kafka 分散型コンピューティングプラットフォーム (MCP) ユーザーガイド

Kafka 分散型コンピューティングプラットフォーム（Kafka MCP と略称）は、Apache Kafka クラスターとのやり取りを簡素化するための強力なツールセットです。このプラットフォームは、消費グループの説明、トピックオフセットの管理、およびプロデューサー/コンシューマーのログ記録などの操作を実行するためのさまざまなコマンドラインツールと REST API を提供します。

🚀 クイックスタート

Kafka MCP を使用する前に、必要な環境を準備し、インストールと設定を行います。以下に詳細な手順を示します。

✨ 主な機能

消費グループ管理

消費グループの説明：指定した消費グループのコンシューマー ID とそれに所属するクライアント ID を取得します。
オフセットの照会：特定のトピックとパーティションにおける消費グループの現在の消費位置を確認します。
オフセットのリセット：消費グループのオフセットを手動で調整し、正確な消費制御を実現します。

トピック管理

トピックの作成：パーティション数、レプリケーションファクターなどのパラメータを定義して、新しいトピックを迅速に作成します。
トピックの削除：不要になったトピックを安全に削除し、クラスターのリソースを解放します。
ISR の監視：トピックパーティションの _isr（In-Sync Replicas）状態をリアルタイムで確認し、データの一貫性を保証します。

メッセージ追跡

メッセージの消費：指定したトピックのメッセージをリアルタイムで消費し、時間範囲やバッチサイズなどの方法で柔軟に取得することができます。
生産ログの生成：指定したトピックにテストまたはデバッグ用のメッセージを簡単に送信します。
オフセットの調整：必要に応じてコンシューマーのオフセットを手動で調整し、正確なストリーム処理を実現します。

📦 インストール

環境要件

Java 8 以上
Kafka バージョン 2.13 以上

インストール手順

Kafka MCP リリースパッケージをダウンロードして解凍します。

wget https://example.com/kafka-mcp.tar.gz
tar -xzvf kafka-mcp.tar.gz
cd kafka-mcp

環境変数を設定します（オプション）。
```
export PATH=$PATH:/path/to/kafka-mcp/bin
```
MCP サービスを起動します（REST API バージョンを使用する場合）。
```
./start_mcp_server.sh
```

💻 使用例

基本的な使用法

ツールの概要

以下に、Kafka MCP で提供される主要なツールの使用例を示します。

1. すべての消費グループをリストする

# すべてのKafka消費グループのリストを取得する
list_consumer_groups --bootstrap-server <broker_address>

2. 特定の消費グループを説明する

# 指定された消費グループの詳細情報を表示する。コンシューマーIDとオフセットを含む
describe_consumer_group --group-id <consumer_group_id> \
                       --bootstrap-server <broker_address> \
                       --include-protocol-info \
                       --include-consumer-info \
                       --include-offsets

3. トピックのリストを取得する

# Kafkaクラスター内のすべてのトピックをリストする
list_topics --bootstrap-server <broker_address>

4. トピックの詳細情報を説明する

# 指定されたトピックのパーティション割り当て、レプリカ、およびISR状態を表示する
describe_topic --topic-name <topic_name> \
               --bootstrap-server <broker_address>

5. トピックのオフセットを管理する

# 特定のトピックのすべてのパーティションのオフセットを消費する
consume_offsets --group-id <consumer_group_id> \
               --topic-name <topic_name> \
               --bootstrap-server <broker_address>

# 指定されたパーティションの現在のオフセットを提供する
seek_consumer_partition --group-id <consumer_group_id> \
                       --topic-name <topic_name> \
                       --partition <partition_number> \
                       --offset <target_offset> \
                       --bootstrap-server <broker_address>

6. プロデューサーとコンシューマーのログを監視する

# 指定されたトピックの最新のメッセージを消費する
consume_logs --topic-name <topic_name> \
             --bootstrap-server <broker_address>

# 模擬プロデューサーのログを生成する
produce_logs --topic-name <topic_name> \
             --message "your_message_here" \
             --bootstrap-server <broker_address>

高度な使用法

ケース1：消費グループのオフセット照会

# 指定された消費グループのオフセットを表示する
consume_offsets --group-id my_consumer_group \
               --bootstrap-server localhost:9092

ケース2：トピックの説明

# 特定のトピックの詳細情報を取得する
describe_topic --topic-name user_activity \
               --bootstrap-server localhost:9092

📚 ドキュメント

設定の参照

システム設定

属性	詳情
Kafka ブローカーアドレス (`--bootstrap-server`)	Kafka サービスの IP とポートを指定します。形式は `host:port` です。
消費グループ ID (`--group-id`)	一連のコンシューマーインスタンスを識別するために使用されます。
トピック名 (`--topic-name`)	操作対象の具体的なトピックです。

環境変数

# デフォルトのブローカーアドレスを設定する
export KAFKA_BOOTSTRAP_SERVER=localhost:9092

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

⚠️ 重要提示

本番環境では、SASL/PLAINTEXT または TLS 暗号化通信を有効にして、API 呼び出しのセキュリティを確保することをお勧めします。

Kafka の ACL（Access Control List）メカニズムを使用して、異なるユーザーまたはアプリケーションのアクセス権限を制限します。

定期的に操作ログをレビューし、異常な行動を識別して迅速に対応します。