AI导哈

Kafka MCP服务

一个全面的Apache Kafka操作模型上下文协议（MCP）服务器，支持与Claude Desktop及其他MCP客户端的无缝集成。

  • by aswinayyolath
  • Python
  • 本地部署
  • 2 星标
  • 更新于
在 GitHub 查看 aswinayyolath/kafka-mcp-server

使用说明

Kafka MCP服务器

用于Apache Kafka操作的全面模型上下文协议（MCP）服务器，可与Claude Desktop和其他MCP客户端无缝集成。

🚀 特性

  • 群集管理：监视群集运行状况、代理信息和元数据
  • 主题操作：创建、列出、描述和删除Kafka主题
  • 消息操作：通过灵活的配置发送和使用消息
  • 消费者群体管理：列出并描述消费者群体
  • 实时监控：获取主题指标、偏移量和性能数据
  • 健康检查：全面的集群健康监测
  • 安全配置：支持SASL、SSL和各种身份验证方法

📋 先决条件

  • Python 3.8+
  • Apache Kafka集群（本地或远程）
  • pip包管理器

🛠️ 安装

  1. 克隆存储库:

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

  2. 创建虚拟环境:

    python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

  3. 安装依赖项:

    pip install mcp kafka-python

  4. 配置环境变量:

    cp .env.template .env
# Edit .env with your Kafka configuration

⚙️ 配置

环境变量

创建一个 .env 文件基于 .env.template:

# Basic Kafka Configuration
KAFKA_BOOTSTRAP_SERVERS=localhost:9092
KAFKA_SECURITY_PROTOCOL=PLAINTEXT

# SASL Configuration (if needed)
KAFKA_SASL_MECHANISM=PLAIN
KAFKA_SASL_USERNAME=your_username
KAFKA_SASL_PASSWORD=your_password

# SSL Configuration (if needed)
KAFKA_SSL_CAFILE=/path/to/ca.pem
KAFKA_SSL_CERTFILE=/path/to/cert.pem
KAFKA_SSL_KEYFILE=/path/to/key.pem

Docker设置（可选）

使用Docker启动本地Kafka集群：

docker-compose up -d

这将启动Kafka localhost:9092.

🧪 测试

运行全面的测试套件来验证您的设置：

# Set environment variables
export KAFKA_BOOTSTRAP_SERVERS=localhost:9092
export KAFKA_SECURITY_PROTOCOL=PLAINTEXT

# Run tests
python test_kafka_mcp.py

测试套件验证：

  • Kafka连接
  • MCP服务器启动
  • 所有可用工具
  • 主题操作
  • 消息操作
  • 资源和提示

🔧 用法

独立服务器

直接运行MCP服务器：

python kafka_mcp_server.py

Claude桌面集成

添加到您的Claude Desktop配置(claude_desktop_config.json):

{
  "mcpServers": {
    "kafka": {
      "command": "python",
      "args": ["/path/to/kafka_mcp_server.py"],
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "localhost:9092",
        "KAFKA_SECURITY_PROTOCOL": "PLAINTEXT"
      }
    }
  }
}

🛠️ 可用工具

集群操作

  • get_cluster_info -获取全面的集群信息
  • health_check -执行群集运行状况检查

主题操作

  • list_topics -列出集群中的所有主题
  • create_topic -创建新主题
  • describe_topic -获取详细的主题信息
  • delete_topic -删除主题
  • get_topic_metrics -获取全面的主题指标
  • get_topic_offsets -获取分区偏移量

消息操作

  • send_message -发送一条消息
  • send_batch_messages -发送多条消息
  • consume_messages -消费来自某个主题的消息

消费者群体运营

  • list_consumer_groups -列出所有消费者群体
  • describe_consumer_group -获取详细的消费者群体信息

📚 资源

服务器提供MCP资源，便于访问集群信息：

  • kafka://cluster/info -实时集群信息
  • kafka://topics/list -当前主题列表

🎯 鼓励

常见场景的内置提示：

  • kafka_monitoring_prompt -生成监控和故障排除指南
  • kafka_troubleshooting_prompt -获取特定问题的帮助

💡 Claude使用示例

一旦与Claude Desktop集成，您可以问：

  • “你能检查我的Kafka集群的健康状况吗？”
  • “列出所有主题及其分区计数”
  • “创建一个名为“用户事件”的主题，包含3个分区”
  • “向用户事件主题发送测试消息”
  • “显示用户事件的最后10条消息”
  • “哪些消费群体很活跃？”

🔒 安全

认证

服务器支持多种Kafka身份验证方法：

  • 明文：无身份验证（仅用于开发）
  • SASL_PLAINTEXT：通过普通连接进行SASL身份验证
  • SASL_SSL：SSL上的SASL身份验证
  • 安全套接层：SSL客户端证书身份验证

最佳实践

  1. 永不承诺 .env 文件 -使用 .env.template 例如
  2. 在生产中使用SSL -始终加密与生产集群的连接
  3. 限制权限 -使用具有最低所需权限的专用服务帐户
  4. 监控访问 -记录并监控MCP服务器的使用情况

🐛 故障排除

常见问题

  1. 连接被拒绝:

    • 验证Kafka是否正在运行
    • 检查 KAFKA_BOOTSTRAP_SERVERS 配置
    • 确保网络连接

  2. 身份验证失败:

    • 验证SASL凭据
    • 检查SSL证书路径
    • 验证安全协议设置

  3. 未找到主题:

    • 确保主题存在
    • 检查主题名称拼写
    • 验证权限

调试模式

通过设置启用调试日志记录：

export PYTHONPATH=.
python -c "import logging; logging.basicConfig(level=logging.DEBUG)"
python kafka_mcp_server.py

🤝 贡献

  1. 分叉存储库
  2. 创建要素分支： git checkout -b feature-name
  3. 进行更改
  4. 运行测试： python test_kafka_mcp.py
  5. 提交更改： git commit -am 'Add feature'
  6. 推送到分支： git push origin feature-name
  7. 提交拉取请求

📄 许可证

此项目根据Apache许可证2.0获得许可-请参阅 许可证 文件以获取详细信息。

🙏 致谢

📞 支持

对于问题和疑问：

  1. 检查 故障排除部分
  2. 运行测试套件以验证您的设置
  3. 查看Kafka和MCP文档
  4. 打开一个包含详细错误信息的问题

使用MCP享受Kafka流媒体的快乐！ 🎉

相关 MCP