Pavanjava Kafka MCP Server

🚀 Kafka MCP サーバー

Apache Kafka と統合されたメッセージコンテキストプロトコル（MCP）サーバーです。大規模言語モデル（LLM）やエージェントアプリケーションに対し、メッセージの発行と消費機能を提供します。

🚀 クイックスタート

このKafka MCPサーバーは、AIモデルが標準化されたインターフェースを通じてKafkaトピックとやり取りできるようにするサーバーを実装しています。以下の機能をサポートしています。

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

✨ 主な機能

• AIモデルが標準化されたインターフェースを通じて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  # Windowsの場合は：venv\Scripts\activate
   

3. 必要な依存関係をインストールします。

   pip install -r requirements.txt
   

   requirements.txt ファイルがない場合は、以下のパッケージをインストールします。

   pip install aiokafka python-dotenv pydantic-settings mcp-server
   

📚 ドキュメント

設定

プロジェクトのルートディレクトリに .env ファイルを作成し、以下の変数を追加します。

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

# オプション：カスタムツールの説明
# TOOL_PUBLISH_DESCRIPTION="カスタム発行ツールの説明"
# TOOL_CONSUME_DESCRIPTION="カスタム消費ツールの説明"

使用方法

サーバーの起動

提供されている 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 で再度読み取ることはできません。