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。