Kafka MCP Server

A comprehensive Model Context Protocol (MCP) server for Apache Kafka operations, enabling seamless integration with Claude Desktop and other MCP clients.

🚀 Features

• Cluster Management: Monitor cluster health, broker information, and metadata
• Topic Operations: Create, list, describe, and delete Kafka topics
• Message Operations: Send and consume messages with flexible configuration
• Consumer Group Management: List and describe consumer groups
• Real-time Monitoring: Get topic metrics, offsets, and performance data
• Health Checks: Comprehensive cluster health monitoring
• Secure Configuration: Support for SASL, SSL, and various authentication methods

📋 Prerequisites

• Python 3.8+
• Apache Kafka cluster (local or remote)
• pip package manager

🛠️ Installation

1. Clone the repository:

   git clone https://github.com/aswinayyolath/kafka-mcp-server.git
   cd kafka-mcp-server
   

2. Create a virtual environment:

   python -m venv venv
   source venv/bin/activate  # On Windows: venv\Scripts\activate
   

3. Install dependencies:

   pip install mcp kafka-python
   

4. Configure environment variables:

   cp .env.template .env
   # Edit .env with your Kafka configuration
   

⚙️ Configuration

Environment Variables

Create a .env file based on .env.template:

# Basic Kafka Configuration
KAFKA_BOOTSTRAP_SERVERS=localhost:9092
KAFKA_SECURITY_PROTOCOL=PLAINTEXT

# SASL Configuration (if needed)
KAFKA_SASL_MECHANISM=PLAIN
KAFKA_SASL_USERNAME=your_username
KAFKA_SASL_PASSWORD=your_password

# SSL Configuration (if needed)
KAFKA_SSL_CAFILE=/path/to/ca.pem
KAFKA_SSL_CERTFILE=/path/to/cert.pem
KAFKA_SSL_KEYFILE=/path/to/key.pem

Docker Setup (Optional)

Start a local Kafka cluster using Docker:

docker-compose up -d

This will start Kafka on localhost:9092.

🧪 Testing

Run the comprehensive test suite to validate your setup:

# Set environment variables
export KAFKA_BOOTSTRAP_SERVERS=localhost:9092
export KAFKA_SECURITY_PROTOCOL=PLAINTEXT

# Run tests
python test_kafka_mcp.py

The test suite validates:

• Kafka connectivity
• MCP server startup
• All available tools
• Topic operations
• Message operations
• Resources and prompts

🔧 Usage

Standalone Server

Run the MCP server directly:

python kafka_mcp_server.py

Claude Desktop Integration

Add to your Claude Desktop configuration (claude_desktop_config.json):

{
  "mcpServers": {
    "kafka": {
      "command": "python",
      "args": ["/path/to/kafka_mcp_server.py"],
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "localhost:9092",
        "KAFKA_SECURITY_PROTOCOL": "PLAINTEXT"
      }
    }
  }
}

🛠️ Available Tools

Cluster Operations

• get_cluster_info - Get comprehensive cluster information
• health_check - Perform cluster health check

Topic Operations

• list_topics - List all topics in the cluster
• create_topic - Create a new topic
• describe_topic - Get detailed topic information
• delete_topic - Delete a topic
• get_topic_metrics - Get comprehensive topic metrics
• get_topic_offsets - Get partition offsets

Message Operations

• send_message - Send a single message
• send_batch_messages - Send multiple messages
• consume_messages - Consume messages from a topic

Consumer Group Operations

• list_consumer_groups - List all consumer groups
• describe_consumer_group - Get detailed consumer group information

📚 Resources

The server provides MCP resources for easy access to cluster information:

• kafka://cluster/info - Real-time cluster information
• kafka://topics/list - Current topics list

🎯 Prompts

Built-in prompts for common scenarios:

• kafka_monitoring_prompt - Generate monitoring and troubleshooting guidance
• kafka_troubleshooting_prompt - Get help with specific issues

💡 Example Usage with Claude

Once integrated with Claude Desktop, you can ask:

• "Can you check the health of my Kafka cluster?"
• "List all topics and their partition counts"
• "Create a topic called 'user-events' with 3 partitions"
• "Send a test message to the user-events topic"
• "Show me the last 10 messages from user-events"
• "What consumer groups are active?"

🔒 Security

Authentication

The server supports various Kafka authentication methods:

• PLAINTEXT: No authentication (development only)
• SASL_PLAINTEXT: SASL authentication over plain connection
• SASL_SSL: SASL authentication over SSL
• SSL: SSL client certificate authentication

Best Practices

1. Never commit .env files - Use .env.template for examples
2. Use SSL in production - Always encrypt connections to production clusters
3. Limit permissions - Use dedicated service accounts with minimal required permissions
4. Monitor access - Log and monitor MCP server usage

🐛 Troubleshooting

Common Issues

1. Connection refused:

   • Verify Kafka is running
   • Check KAFKA_BOOTSTRAP_SERVERS configuration
   • Ensure network connectivity

2. Authentication failures:

   • Verify SASL credentials
   • Check SSL certificate paths
   • Validate security protocol settings

3. Topic not found:

   • Ensure topic exists
   • Check topic name spelling
   • Verify permissions

Debug Mode

Enable debug logging by setting:

export PYTHONPATH=.
python -c "import logging; logging.basicConfig(level=logging.DEBUG)"
python kafka_mcp_server.py

🤝 Contributing

1. Fork the repository
2. Create a feature branch: git checkout -b feature-name
3. Make your changes
4. Run tests: python test_kafka_mcp.py
5. Commit changes: git commit -am 'Add feature'
6. Push to branch: git push origin feature-name
7. Submit a pull request

📄 License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.

🙏 Acknowledgments

• Model Context Protocol - For the MCP specification
• Apache Kafka - For the distributed streaming platform
• kafka-python - For the Python Kafka client

📞 Support

For issues and questions:

1. Check the troubleshooting section
2. Run the test suite to validate your setup
3. Review Kafka and MCP documentation
4. Open an issue with detailed error information

---

Happy Kafka streaming with MCP! 🎉