Kafka Dataops

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 50-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