Kafka MCP Server

Kafka MCP サーバー

LLM およびエージェント型アプリケーションにパブリッシュ機能とコンシューム機能を提供する、Apache Kafka と連携する Message Context Protocol（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 ファイルを作成し、次の変数を設定します。

# Kafka Configuration
KAFKA_BOOTSTRAP_SERVERS=localhost:9092
TOPIC_NAME=your-topic-name
IS_TOPIC_READ_FROM_BEGINNING=False
DEFAULT_GROUP_ID_FOR_CONSUMER=kafka-mcp-group

# Optional: Custom Tool Descriptions
# TOOL_PUBLISH_DESCRIPTION="Custom description for the publish tool"
# TOOL_CONSUME_DESCRIPTION="Custom description for the consume tool"

使用方法

サーバーの実行

付属の main.py スクリプトを使用してサーバーを実行できます。

python main.py --transport stdio

使用できるトランスポートオプション：

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

Claude Desktop との統合

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

{
    "mcpServers": {
        "kafka": {
            "command": "python",
            "args": [
                "<PATH TO PROJECTS>/main.py"
            ]
        }
    }
}

<PATH TO PROJECTS> をプロジェクトディレクトリの絶対パスに置き換えてください。

プロジェクト構成

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

使用可能なツール

kafka-publish

設定された Kafka トピックに情報をパブリッシュします。

kafka-consume

設定された Kafka トピックから情報をコンシュームします。

• 注：トピックから読み取ったメッセージは、同じ groupid を使用して再度読み取ることはできません。

Create-Topic

指定したパラメーターで新しい Kafka トピックを作成します。

• オプション：
  • --topic作成するトピックの名前
  • --partitions割り当てるパーティション数
  • --replication-factorブローカー間のレプリケーション係数
  • --config（任意）トピック単位の設定の上書き（例：retention.ms=604800000）

Delete-Topic

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

• オプション：
  • --topic削除するトピックの名前
  • --timeout（任意）削除の完了を待つ時間

List-Topics

クラスター内のすべてのトピックを一覧表示します（またはパターンで絞り込みます）。

• オプション：
  • --bootstrap-serverブローカーのアドレス
  • --pattern（任意）トピック名を絞り込む正規表現
  • --exclude-internal（任意）内部トピックを除外します（デフォルト：true）

Topic-Configuration

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

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

Topic-Metadata

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

• オプション：
  • --topic（指定した場合）このトピックのみのメタデータを取得します
  • --bootstrap-serverブローカーのアドレス
  • --include-offline（任意）オフラインのブローカーまたはパーティションを含めます