Kafka Sentinel

kafka-sentinel-mcp

让 AI Agent 安全、只读地观察你的 Kafka 集群。

一个 MCP（Model Context Protocol） 服务，将 Kafka 集群健康状况、消费者延迟、分区状态和重放就绪情况作为结构化工具提供出来，让大模型 Agent（Claude 或任何 MCP 客户端）能够诊断流处理故障，而不会造成任何破坏。

由一位工程师打造。他曾花十年时间运维基于 Kafka 的金融消息系统，实现 99.999% 的可用性，后来厌倦了每个“AI + Kafka”演示都假设可以写入生产环境。

为什么要做这个项目

当一个消费者组在凌晨 3 点停滞时，问题总是一样：是延迟吗？是分区卡住了吗？是发生了再均衡风暴吗？还是偏移量重置出了问题？这些都是模式匹配问题，恰恰是大模型 Agent 擅长处理的；但没有运维人员会把生产集群的管理权限交给 Agent。

kafka-sentinel-mcp 划定了一条明确的界线：每个工具在设计上都只读，并通过客户端配置强制执行（甚至不会导入任何管理操作）。Agent 可以观察、关联并提出建议；由人工执行操作。

工具

工具	返回内容
list_topics	所有非内部主题及其分区数和复制因子——不知道主题名称时从这里开始
list_consumer_groups	所有消费者组 ID 及其状态——不知道组名称时从这里开始
cluster_health	Broker 数量、控制器状态、副本不足/离线分区数量
consumer_lag	按组、主题和分区显示的延迟，以及已提交偏移量与末尾偏移量
topic_audit	复制因子、min.insync.replicas、保留时间，并标记违反持久性最佳实践的配置
partition_state	Leader、副本集（ISR）缩减情况、各 Broker 之间的分布偏斜
replay_readiness	对于一个组和主题：最早可用偏移量与已提交偏移量的对比，也就是“我们还能重放错过的内容吗？”
incident_snapshot	一次调用即可获取上述所有信息，并附带时间戳——专为粘贴到事后复盘报告中而设计

快速开始

pip install kafka-sentinel-mcp   # (or: uv tool install)

# Run against your cluster (read-only credentials!)
KAFKA_BOOTSTRAP=localhost:9092 kafka-sentinel-mcp

添加到 Claude Desktop / 任意 MCP 客户端：

{
  "mcpServers": {
    "kafka-sentinel": {
      "command": "kafka-sentinel-mcp",
      "env": { "KAFKA_BOOTSTRAP": "broker1:9092,broker2:9092" }
    }
  }
}

然后询问你的 Agent：“为什么 payments-consumer 组越来越落后？我们还能从它停滞的位置开始重放吗？”

安全措施

• **从设计上只读：**不生产消息、不修改主题/配置、不提交偏移量、不执行 ACL 操作。永远不会导入修改 API；并且 CI 中的测试 每次运行都会扫描服务源代码，确保这一点。

• 观察者消费者以 enable.auto.commit=False 运行，且从不提交——已通过真实 Broker 验证，而非仅仅断言。

• 支持 SASL/SSL；凭据只从环境中读取，绝不会记录到日志中。

• 每次工具调用都会连同其参数一起记录，以便审计。

• **最小权限：**使用仅在集群和主题上具有 Describe 权限、在消费者组上具有 Describe 权限的主体运行。当 ACL 拒绝某项操作时，工具会返回结构化结果，而不是堆栈跟踪：

  {
    "error": "permission_denied",
    "operation": "list_consumer_groups",
    "detail": "...",
    "hint": "The Kafka principal in use lacks the ACL required for this operation. ..."
  }
  

  Agent 随后可以告知运维人员缺少哪项 ACL，而不会显得像是程序故障。非授权错误不会被刻意吞掉，而是会继续传播，因为在无关错误发生时静默降级会掩盖真正的问题。

测试

pip install -e ".[dev]"

pytest -m "not integration"   # fast, fully mocked — no Docker needed
pytest -m integration         # starts a real Kafka via testcontainers (needs Docker)
pytest                        # both

单元测试套件完全模拟 librdkafka，并覆盖工具逻辑。集成测试套件会启动实际的代理，生成真实记录，并断言工具会返回正确的滞后量、ISR 状态、持久性标志和重放就绪状态——同时确认观察器不会留下已提交的偏移量。两者都会在 CI 中运行。

状态

项目尚处早期，但已经过测试。请参阅 ROADMAP.md。欢迎提交问题和 PR——尤其欢迎分享你希望 Agent 在事故期间能告诉你的经验教训。

引用本项目

如果你在学术作品中引用本项目，请参阅 CITATION.cff，或使用 GitHub 上的“引用此仓库”按钮。

许可证

MIT