Kafka MCP Server

Kafka MCP サーバー

Apache Kafka と統合して、LLM および Agentic アプリケーションの公開および消費機能を提供するメッセージ コンテキスト プロトコル (MCP) サーバー。

概要

このプロジェクトは、AIモデルが標準化されたインターフェースを介してKafkaトピックと対話できるようにするサーバーを実装します。サポート対象は以下のとおりです。

• Kafkaトピックへのメッセージの公開
• Kafkaトピックからのメッセージの消費

前提条件

• Python 3.8以上
• Apache Kafka インスタンス
• Python の依存関係 (インストールのセクションを参照)

インストール

1. リポジトリをクローンします。
   git clone <repository-url> cd <repository-directory>
2. 仮想環境を作成してアクティブ化します。
   python -m venv venv source venv/bin/activate # On Windows, use: venv\Scripts\activate
3. 必要な依存関係をインストールします。
   pip install -r requirements.txt
   requirements.txt が存在しない場合は、次のパッケージをインストールします。
   pip install aiokafka python-dotenv pydantic-settings mcp-server

構成

次の変数を含む.envファイルをプロジェクト ルートに作成します。

使用法

サーバーの実行

提供されているmain.pyスクリプトを使用してサーバーを実行できます。

利用可能な交通手段:

• stdio : 標準入出力（デフォルト）
• sse : サーバー送信イベント

Claude Desktopとの統合

この Kafka MCP サーバーを Claude Desktop で使用するには、Claude Desktop 構成ファイルに次の構成を追加します。

<PATH TO PROJECTS>プロジェクト ディレクトリへの絶対パスに置き換えます。

プロジェクト構造

• main.py : アプリケーションのエントリポイント
• kafka.py : Kafka コネクタの実装
• server.py : Kafka とのやり取りのためのツールを備えた MCP サーバーの実装
• settings.py : Pydantic を使用した構成管理

利用可能なツール

kafka-publish

構成された Kafka トピックに情報を公開します。

kafka-consume

構成された Kafka トピックから情報を消費します。

• 注意: 一度トピックからメッセージを読み取ると、同じグループIDを使用して再度読み取ることはできません。

トピックの作成

指定されたパラメータを使用して新しい Kafka トピックを作成します。

• オプション:
  • --topic作成するトピックの名前
  • --partitions割り当てるパーティションの数
  • --replication-factorブローカー間のレプリケーション係数
  • --config (オプション) トピックレベルの構成のオーバーライド (例: retention.ms=604800000 )

トピックを削除

既存の Kafka トピックを削除します。

• オプション:
  • --topic削除するトピックの名前
  • --timeout (オプション) 削除が完了するまでの待機時間

リストトピック

クラスター内のすべてのトピックを一覧表示します (またはパターンでフィルタリングします)。

• オプション:
  • --bootstrap-serverブローカーアドレス
  • --pattern (オプション) トピック名をフィルタリングする正規表現
  • --exclude-internal (オプション) 内部トピックを除外する (デフォルト: true)

トピック構成

1 つ以上のトピックの構成を表示または変更します。

• オプション:
  • --describeトピックの現在の設定を表示する
  • --alter構成を変更します (例: --add-config retention.ms=86400000,--delete-config cleanup.policy )
  • --topicトピックの名前

トピックメタデータ

トピックまたはクラスターに関するメタデータを取得します。

• オプション:
  • --topic (指定されている場合) このトピックのメタデータのみを取得します
  • --bootstrap-serverブローカーアドレス
  • --include-offline (オプション) オフラインのブローカーまたはパーティションを含める