Kafka DataOps 诊断与监控 MCP 服务

一个面向 Kafka DataOps 的 MCP 服务器,提供消费者滞后诊断、Topic/ISR 健康检查与 Broker 集群状态监控,适用于 Kafka CDC 管道运维和故障排查。

  • by Aguantar
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 Aguantar/kafka-mcp-server

使用说明

kafka-dataops-mcp

mcp-name: io.github.Aguantar/kafka-dataops-mcp

A DataOps-focused Kafka MCP server with consumer lag diagnosis and broker health monitoring. Diagnosis logic is based on actual CDC pipeline operational experience.

Features

  • kafka_consumer_lag — Consumer group lag with incident-pattern diagnosis
  • kafka_topic_info — Topic details with ISR/replication health checks
  • kafka_broker_status — Cluster health: brokers, controller, under-replicated partitions
  • kafka_list_topics — Topic catalog with built-in descriptions

Diagnosis based on real incidents

The diagnosis logic is not generic — it's based on actual operational experience:

  • Flink crash detection: "no active members" + growing lag = likely Flink Job failure (based on a 46-hour outage caused by MySQL DELETE → Debezium tombstone → Flink NPE)
  • Checkpoint vs consumer group: warns that Kafka consumer group reset alone is insufficient for Flink — checkpoints must be deleted first
  • ClusterIdMismatch: detects missing brokers and suggests Docker volume conflict as root cause
  • ISR monitoring: ISR < min.insync.replicas = write failures (critical)

Installation

pip install kafka-dataops-mcp

Usage with Claude Code

Add to your .mcp.json:

{
  "mcpServers": {
    "kafka": {
      "command": "kafka-dataops-mcp",
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "localhost:9092"
      }
    }
  }
}

Environment Variables

Variable Default Description
KAFKA_BOOTSTRAP_SERVERS localhost:9092 Kafka bootstrap servers
KAFKA_COMMAND_TIMEOUT 10 Command timeout in seconds

License

MIT