Kafka MCP Server

Kafka MCP 服务

本项目提供一个 Model Context Protocol（MCP） 服务，可通过基于标准工具的接口与 Kafka 集群交互。它提供 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.properties），其中包含 Kafka 连接详情：

# 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，启动一个包含单个 broker 的 Kafka 集群，然后使用 bootstrap.server localhost:9092 与其交互。

---

🔗 资源

• MCP 协议文档
• Claude + MCP 指南
• mcp 命令行参考