MCP Kafka Server

🚀 MCP Kafka ツール

これはMCP（モデルコンテキストプロトコル）プロトコルに基づくKafkaツールで、大規模モデルを通じてKafkaに接続し、以下の機能を提供します。

1. 指定されたトピックの情報を照会する
2. 指定されたトピックにデータを読み書きする
3. 指定された時間、指定されたトピックのデータを再生する

📦 インストール

pip install -r requirements.txt

💻 使用例

ローカルモード（標準入出力）

mcp_kafka_server.pyファイルを直接実行します。

python mcp_kafka_server.py

その後、MCPクライアント（Claude、Cherry Studioなど）でローカルのMCPサーバーに接続します。

Webサービスモード（SSE）

mcp_kafka_sse_server.pyファイルを実行してWebサービスを起動します。

python mcp_kafka_sse_server.py

サービスは http://localhost:8000 で起動し、MCPエンドポイントは http://localhost:8000/mcp です。 その後、MCPクライアントでこのリモートのMCPサーバーに接続します。

✨ 主な機能

1. すべてのトピックをリストする

list_topics()

Kafka内のすべての利用可能なトピックのリストを返します。

2. トピック情報を取得する

get_topic_info(topic_name)

指定されたトピックの詳細情報を取得します。これには、パーティション数、各パーティションの開始と終了オフセット、およびメッセージ数が含まれます。

3. メッセージを送信する

send_message(topic_name, message, key)

指定されたトピックにメッセージを送信します。messageはJSON文字列または通常の文字列で、keyはオプションのパラメータです。

4. メッセージを読み取る

read_messages(topic_name, max_messages, timeout_ms)

指定されたトピックから最新のメッセージを読み取ります。max_messagesは最大読み取りメッセージ数を指定し（デフォルトは10件）、timeout_msはタイムアウト時間を指定します（デフォルトは1000ミリ秒）。

5. 過去のメッセージを再生する

replay_messages(topic_name, start_time, end_time, max_messages)

指定された期間内のメッセージを再生します。start_timeとend_timeはISO形式の時間文字列（例：'2023-01-01T00:00:00'）で、max_messagesは最大再生メッセージ数を指定します（デフォルトは100件）。

🔧 Kafka接続の設定

デフォルトではlocalhost:9092に接続します。接続先を変更する場合は、KafkaManagerクラスの初期化パラメータでbootstrap_serversを変更してください。