Kafka MCP Server

Kafka MCP サーバー

このプロジェクトは、標準的なツールベースのインターフェースを通じて Kafka クラスターを操作できる Model Context Protocol（MCP） サーバーを提供します。メッセージの生成や消費、トピックの一覧表示など、さまざまな Kafka 操作を公開し、Claude Desktop または MCP 対応クライアントから利用できます。

🛠 特長

• Kafka トピックの一覧表示
• トピックの作成と削除
• メッセージの生成

---

🚀 クイックスタート

1. リポジトリをクローンする

git clone https://github.com/joel-hanson/kafka-mcp-server.git
cd kafka-mcp-server

2. Python 環境をセットアップする（conda を使用）

conda create -n kafka-mcp python=3.10 -y
conda activate kafka-mcp
pip install -r requirements.txt

3. MCP サーバーを起動する（ローカル開発用）

mcp dev

これによりサーバーが起動し、ツールのプロンプトと応答をテストできる MCP Inspector が開きます。

---

使用方法

サーバーの実行

python server.py

利用可能なツール

1. kafka_initialize_connection - プロパティファイルを使用して Kafka に接続します
2. kafka_list_topics - クラスター内のすべてのトピックを一覧表示します
3. kafka_create_topic - 新しいトピックを作成します
4. kafka_delete_topic - 既存のトピックを削除します
5. kafka_get_topic_info - トピックの詳細情報を取得します

使用例の流れ

1. まず、Kafka に接続します：

   Use kafka_initialize_connection with your properties file <path>
   OR
   Initialize connection with kafka using the properties file <path>
   

2. 既存のトピックを一覧表示します：

   Use kafka_list_topics to see all topics
   OR
   List all the topics
   

3. 新しいトピックを作成します：

   Use kafka_create_topic with name "my-topic", partitions 3, replication_factor 1
   OR
   Create me a topic with name "my-topic"
   

4. トピックの詳細を取得します：

   Use kafka_get_topic_info for "my-topic"
   

機能

• 構成管理：標準的なプロパティファイルから Kafka 接続の詳細を読み込みます
• トピック管理：トピックの一覧表示、作成、削除、確認を行います
• エラー処理：わかりやすいメッセージを含む包括的なエラー処理を提供します
• 接続管理：効率的な接続プーリングとクリーンアップを行います
• 拡張性：Kafka の操作を簡単に追加できます

対応する Kafka 操作

現在

• トピックの一覧表示
• パーティション数とレプリケーション係数を指定したトピックの作成
• トピックの削除
• トピックの確認（パーティション、レプリカ、ISR）

今後の予定

• メッセージの生成
• メッセージの消費
• コンシューマーグループの管理
• オフセット管理
• クラスター情報
• パフォーマンス指標

トラブルシューティング

よくある問題

1. 接続失敗：プロパティファイルの bootstrap.servers を確認してください
2. 認証エラー：SASL/SSL の構成を確認してください
3. トピックがすでに存在します：既存のトピックを確認するには kafka_get_topic_info を使用してください
4. 権限がありません：Kafka ユーザーに必要な権限があることを確認してください

ログ

サーバーは stdout にログを出力します。コード内の logging.basicConfig(level=logging.INFO) 行を変更すると、ログレベルを調整できます。

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

• 機密性の高い認証情報は安全に保管してください
• 本番環境では SSL/TLS を使用してください
• 適切な認証と認可を実装してください
• 認証情報は定期的にローテーションしてください

🧠 Claude Desktop での使用

Kafka MCP サーバーを Claude Desktop と統合するには：

Claude にインストールします：

mcp install <file_spec> -f <optional env file>

注意：依存関係に関する問題が発生する場合は、claud_desktop_config.json に Python の絶対パスを設定し、実行するファイルを指定してください。

MCP クライアントの構成

MCP クライアントの構成（例：Claude Desktop）に次の内容を追加します：

{
  "mcpServers": {
    "kafka": {
      "command": "python",
      "args": ["/path/to/server.py"]
    }
  }
}

{
"mcpServers": {
      "Kafka MCP Server": {
         "command": "/opt/miniconda3/envs/kafka-mcp/bin/python",
         "args": [
         "/path/to/server.py"
         ]
      }
   }
}

これで Claude は、Python サーバーで @mcp.prompt() を通じて定義したツールを検出し、操作できるようになります。

構成

Kafka の接続情報を含む Kafka プロパティファイル（例：kafka.properties）を作成します：

# Basic Kafka connection
bootstrap.servers=localhost:9092
client.id=kafka-mcp-client

# Security configuration (if needed)
# security.protocol=SASL_SSL
# sasl.mechanism=PLAIN
# sasl.username=your-username
# sasl.password=your-password

# SSL configuration (if needed)
# ssl.ca.location=/path/to/ca-cert
# ssl.certificate.location=/path/to/client-cert
# ssl.key.location=/path/to/client-key

---

📂 プロジェクト構成

kafka_mcp_server/
├── server.py               # Main FastMCP server defining tools
├── kafka_utils.py
├── requirements.txt
├── README.md
└── ...

---

🧩 ツールのデバッグに関するヒント

• mcp dev を使用して、プロンプトの動作を対話的に確認します
• ツール関数に print() または logging.debug() を追加します
• 正しく解析されるよう、構造化された @prompt 関数シグネチャを使用します

---

📦 要件

次の内容を requirements.txt に追加します：

kafka-python
mcp[cli]
pydantic
uv

---

テスト

docker compose up を実行して、単一ブローカーの Kafka クラスターを起動してください。ブートストラップサーバー localhost:9092 を使用して操作できます。

---

🔗 リソース

• MCP プロトコルのドキュメント
• Claude + MCP ガイド
• mcp CLI リファレンス