Kafka智能管理工具

AI Assistant → MCP Protocol → KafkaIQ → Apache Kafka Cluster

# Clone the repository
git clone https://github.com/ashfaqbs/kafkaiq.git
cd kafkaiq

# Install dependencies
pip install fastmcp kafka-python

# Set email credentials (optional)
export KAFKAIQ_GMAIL_USER="your-email@gmail.com"
export KAFKAIQ_GMAIL_PASSWORD="your-app-password"

# Run the server
python kafka_mcp_server.py

python kafka_mcp_server.py

Human: Connect to my local Kafka cluster server : localhost:9092
AI → KafkaIQ: initialize_kafka_connection("localhost:9092")
KafkaIQ Response: "Successfully connected to Kafka cluster at localhost:9092"
AI → Human: "Connected to your Kafka cluster successfully!"

Human: What topics exist in my cluster?
AI → KafkaIQ: list_kafka_topics()
KafkaIQ Response: {
  "topics": ["orders", "users", "events", "payments"],
  "count": 4,
  "include_internal": false
}
AI → Human: "Your cluster has 4 topics: orders, users, events, and payments."

Human: Create a topic called 'notifications' with 5 partitions
AI → KafkaIQ: create_kafka_topic("notifications", 5, 1)
KafkaIQ Response: {
  "status": "success",
  "message": "Topic 'notifications' created successfully",
  "topic_name": "notifications",
  "num_partitions": 5,
  "replication_factor": 1
}
AI → Human: "Successfully created topic 'notifications' with 5 partitions!"

Human: How is my Kafka cluster doing?
AI → KafkaIQ: kafka_alert_summary()
KafkaIQ Response: {
  "status": "OK",
  "signals": [],
  "summary": {
    "cluster_id": "kafka-cluster-1",
    "broker_count": 3,
    "topics_count": 15,
    "under_replicated": 0,
    "offline_partitions": 0,
    "single_broker": false,
    "lag": [
      {"group": "processors", "topic": "orders", "total_lag": 50, "level": "OK"},
      {"group": "analytics", "topic": "events", "total_lag": 1250, "level": "OK"}
    ]
  }
}
AI → Human: "Your Kafka cluster is healthy! 3 brokers running, 15 topics, no offline partitions or under-replicated partitions. Consumer lag looks good across all groups."

Human: Check consumer lag for the 'analytics' group on 'events' topic
AI → KafkaIQ: get_consumer_lag("analytics", "events")
KafkaIQ Response: {
  "group_id": "analytics",
  "topic": "events",
  "total_lag": 1250,
  "partitions": [
    {"partition": 0, "consumer_offset": 10000, "high_water_mark": 10500, "lag": 500},
    {"partition": 1, "consumer_offset": 8000, "high_water_mark": 8750, "lag": 750}
  ]
}
AI → Human: "The 'analytics' group has a total lag of 1,250 messages on the 'events' topic. Partition 0 has 500 messages lag, partition 1 has 750 messages lag."

Human: Show me how partitions are distributed across brokers
AI → KafkaIQ: broker_leadership_distribution()
KafkaIQ Response: {
  "total_partitions": 45,
  "distribution": {
    "1": 15,
    "2": 15, 
    "3": 15
  },
  "by_topic": {
    "orders": {"1": [0, 3], "2": [1, 4], "3": [2, 5]},
    "events": {"1": [0, 2], "2": [1, 3], "3": [4, 5]}
  }
}
AI → Human: "Partition leadership is well-balanced! Each of your 3 brokers leads exactly 15 partitions out of 45 total. The distribution looks optimal."

Human: Are there any offline partitions?
AI → KafkaIQ: get_offline_partitions()
KafkaIQ Response: {
  "offline_count": 0,
  "by_topic": {}
}
AI → Human: "Great news! There are no offline partitions in your cluster. All partitions have active leaders."

Human: What are the resources for broker-1?
AI → KafkaIQ: get_broker_resources("broker-1")
KafkaIQ Response: {
  "broker_id": "broker-1",
  "resources": {
    "ram_gb": 32,
    "storage_gb": 1000,
    "cores": 8
  },
  "status": "active",
  "note": "Hardcoded values - integrate with monitoring system"
}
AI → Human: "Broker-1 has 32GB RAM, 1TB storage, and 8 CPU cores. Status is active."

Human: Email the cluster status to admin@company.com
AI → KafkaIQ: send_email_notification("admin@company.com", "Cluster Status: OK\nBrokers: 3\nTopics: 15\nNo issues detected", "KafkaIQ Daily Report")
KafkaIQ Response: {
  "status": "sent",
  "message": "Email sent successfully",
  "recipient": "admin@company.com",
  "subject": "KafkaIQ Daily Report",
  "timestamp": "2025-09-28T10:30:00"
}
AI → Human: "Successfully sent the cluster status report to admin@company.com!"

┌─────────────────┐    ┌──────────────────┐    ┌─────────────────┐
│   AI Assistant  │    │     KafkaIQ      │    │  Kafka Cluster  │
│                 │◄──►│   MCP Server     │◄──►│                 │
│ (Claude, GPT)   │    │   (Port 8080)    │    │  (Brokers)      │
└─────────────────┘    └──────────────────┘    └─────────────────┘
         │                        │                       │
         │                        │                       │
         ▼                        ▼                       ▼
┌─────────────────┐    ┌──────────────────┐    ┌─────────────────┐
│ Natural Language│    │  Tool Execution  │    │   Operations    │
│   Commands      │    │   & Response     │    │  & Monitoring   │
└─────────────────┘    └──────────────────┘    └─────────────────┘

# Email Configuration (Optional)
KAFKAIQ_GMAIL_USER=your-email@gmail.com
KAFKAIQ_GMAIL_PASSWORD=your-app-password

# Built-in thresholds (edit as needed)
offline_crit = 1              # Any offline partition = CRITICAL
urp_warn, urp_crit = 1, 5     # Under-replicated partitions
lag_warn, lag_crit = 10_000, 100_000  # Consumer lag thresholds
single_broker_warn = True     # Warn on single broker setup

Human: Create 5 topics for microservices: user-service, order-service, payment-service, notification-service, audit-service
AI: I'll create all 5 topics for your microservices architecture...

Human: Set up monitoring that emails me if any consumer lag exceeds 50,000 messages
AI: I'll configure automated monitoring with email alerts for high consumer lag...

# Custom topic configuration example
topic_configs = {
    "retention.ms": "604800000",  # 7 days
    "compression.type": "lz4",
    "max.message.bytes": "1048576"  # 1MB
}

Error: "Failed to connect to Kafka cluster"
Solution: Check if Kafka is running on specified port

Error: "Gmail authentication failed"
Solution: Enable 2FA and generate App Password

Error: "Consumer lag check timed out"
Solution: Increase timeout in KafkaConsumer config

Error: "Not authorized to perform operation"
Solution: Check Kafka ACLs and user permissions

# Clone and setup development environment
git clone https://github.com/yourusername/kafkaiq.git
cd kafkaiq

# Create virtual environment
python -m venv venv
source venv/bin/activate  # Linux/Mac
# or
venv\Scripts\activate     # Windows

# Install development dependencies
pip install -r requirements.txt

# Run tests
python -m pytest tests/

# Start development server
python kafka_mcp_server.py