Kafka Observer

An MCP (Model Context Protocol) server that gives AI agents real-time observability into Apache Kafka clusters. Monitor broker health, track consumer lag, and diagnose issues — all through natural language.

Why?

Kafka monitoring typically requires juggling multiple dashboards. This MCP server lets any AI assistant (Claude, ChatGPT, Cursor, VS Code Copilot) query your Kafka cluster directly:

• "Is my Kafka cluster healthy?"
• "What's the consumer lag for payment-processor group?"
• "Why is lag spiking on the orders topic?"

Tools

Tool	Description
get_broker_health	Cluster state: brokers, controller, under-replicated partitions
list_topics	All topics with partition counts and replication factors
describe_topic	Detailed config and partition assignments for a topic
get_consumer_lag	Per-partition lag for a consumer group
diagnose_lag_spike	Automated root-cause analysis for lag issues
get_cache_stats	Cache hit/miss statistics for observability

Resources

Resource URI	Description
kafka://cluster/overview	High-level cluster summary

Prompts

Prompt	Description
investigate_lag	Step-by-step workflow for diagnosing consumer lag
capacity_review	Template for cluster capacity planning

Quick Start

Prerequisites

• Python 3.12+
• Docker (for local Kafka)
• uv package manager

Setup

git clone https://github.com/Rushi264/mcp-kafka-observer.git
cd mcp-kafka-observer

# Install dependencies
uv sync

# Start local Kafka
docker compose up -d

# Run tests
uv run pytest -v

Claude Desktop Integration

Add to your claude_desktop_config.json:

{
  "mcpServers": {
    "kafka-observer": {
      "command": "uv",
      "args": [
        "--directory", "/path/to/mcp-kafka-observer",
        "run", "python", "-m", "mcp_kafka_observer.server"
      ],
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "localhost:9092"
      }
    }
  }
}

Architecture

MCP Client (Claude / Cursor / VS Code Copilot)
    │
    │  MCP Protocol (stdio)
    ▼
mcp-kafka-observer
    ├── Tools (get_broker_health, get_consumer_lag, ...)
    ├── Resources (kafka://cluster/overview)
    ├── Prompts (investigate_lag, capacity_review)
    ├── TTL Cache (prevents thundering herd on admin API)
    └── Analyzer (automated lag diagnosis)
    │
    │  confluent-kafka AdminClient
    ▼
Kafka Cluster

Tech Stack

• Python 3.12 with async/await
• MCP SDK (FastMCP) — official Anthropic SDK
• confluent-kafka — production-grade Kafka client (librdkafka)
• Pydantic — structured output validation
• Docker Compose — local Kafka for development

Testing

# Unit tests (no Kafka needed)
uv run pytest tests/test_server.py -v

# Integration tests (needs Docker Kafka running)
docker compose up -d
uv run pytest tests/test_kafka_client.py -v

# All tests
uv run pytest -v

# Linter
uv run ruff check src/ tests/

Configuration

Set via environment variables or .env file:

Variable	Default	Description
KAFKA_BOOTSTRAP_SERVERS	localhost:9092	Kafka broker addresses
KAFKA_SASL_MECHANISM	—	SASL auth mechanism (PLAIN, SCRAM-SHA-256)
KAFKA_SASL_USERNAME	—	SASL username
KAFKA_SASL_PASSWORD	—	SASL password
KAFKA_SECURITY_PROTOCOL	—	Security protocol (SASL_SSL, SASL_PLAINTEXT)

License

MIT