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

```bash
pip install kafka-dataops-mcp
```

## Usage with Claude Code

Add to your `.mcp.json`:

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