MCP Kafka Server

🚀 MCP Kafka 工具

MCP Kafka 工具基於 MCP（模型上下文協議）協議，藉助大模型實現與 Kafka 的連接。它具備查詢主題信息、讀寫指定主題數據以及回放特定時間和主題數據等功能，為 Kafka 數據操作提供了便捷途徑。

🚀 快速開始

本工具支持本地模式（標準輸入輸出）和 Web 服務模式（SSE）兩種使用方式，你可按需選擇：

本地模式（標準輸入輸出）

1. 安裝依賴：

pip install -r requirements.txt

2. 運行服務：

python mcp_kafka_server.py

3. 連接服務：在 MCP 客戶端（如 Claude、Cherry Studio 等）中連接本地 MCP 服務器。

Web 服務模式（SSE）

1. 安裝依賴：

pip install -r requirements.txt

2. 運行服務：

python mcp_kafka_sse_server.py

服務將啟動在 http://localhost:8000，MCP 端點為 http://localhost:8000/mcp 3. 連接服務：在 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 條）。

📦 安裝指南

pip install -r requirements.txt

💻 使用示例

基礎用法

# 列出所有主題
list_topics()

# 獲取指定主題信息
get_topic_info('example_topic')

# 發送消息
send_message('example_topic', 'Hello, Kafka!', 'key')

# 讀取消息
read_messages('example_topic')

# 回放歷史消息
replay_messages('example_topic', '2023-01-01T00:00:00', '2023-01-02T00:00:00')

高級用法

# 讀取消息時指定最大消息數和超時時間
read_messages('example_topic', max_messages=20, timeout_ms=2000)

# 回放歷史消息時指定最大回放消息數
replay_messages('example_topic', '2023-01-01T00:00:00', '2023-01-02T00:00:00', max_messages=200)

🔧 技術細節

默認連接到 localhost:9092，如需修改連接地址，請在 KafkaManager 類的初始化參數中修改 bootstrap_servers。