Kafka MCP Server

Kafka MCP Server

This project provides a Model Context Protocol (MCP) server that enables interaction with a Kafka cluster via standard tool-based interfaces. It exposes Kafka operations like producing messages, consuming, listing topics, and more — all accessible through Claude Desktop or any MCP-compatible client.

🛠 Features

• List Kafka topics
• Create and delete topics
• Produce messages

---

🚀 Quick Start

1. Clone the repository

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

2. Setup Python environment (using conda)

conda create -n kafka-mcp python=3.10 -y
conda activate kafka-mcp
pip install -r requirements.txt

3. Start the MCP server (for local dev)

mcp dev

This starts the server and opens the MCP Inspector so you can test tool prompts and responses.

---

Usage

Running the Server

python server.py

Available Tools

1. kafka_initialize_connection - Connect to Kafka using a properties file
2. kafka_list_topics - List all topics in the cluster
3. kafka_create_topic - Create a new topic
4. kafka_delete_topic - Delete an existing topic
5. kafka_get_topic_info - Get detailed information about a topic

Example Usage Flow

1. First, connect to Kafka:

   Use kafka_initialize_connection with your properties file <path>
   OR
   Initialize connection with kafka using the properties file <path>
   

2. List existing topics:

   Use kafka_list_topics to see all topics
   OR
   List all the topics
   

3. Create a new topic:

   Use kafka_create_topic with name "my-topic", partitions 3, replication_factor 1
   OR
   Create me a topic with name "my-topic"
   

4. Get topic details:

   Use kafka_get_topic_info for "my-topic"
   

Features

• Configuration Management: Loads Kafka connection details from standard properties files
• Topic Management: List, create, delete, and inspect topics
• Error Handling: Comprehensive error handling with informative messages
• Connection Management: Efficient connection pooling and cleanup
• Extensible: Easy to add more Kafka operations

Supported Kafka Operations

Current

• Topic listing
• Topic creation with custom partitions and replication factor
• Topic deletion
• Topic inspection (partitions, replicas, ISR)

Planned Features

• Message production
• Message consumption
• Consumer group management
• Offset management
• Cluster information
• Performance metrics

Troubleshooting

Common Issues

1. Connection Failed: Verify your bootstrap.servers in the properties file
2. Authentication Error: Check your SASL/SSL configuration
3. Topic Already Exists: Use kafka_get_topic_info to check existing topics
4. Permission Denied: Ensure your Kafka user has the necessary permissions

Logging

The server logs to stdout. You can adjust the logging level by modifying the logging.basicConfig(level=logging.INFO) line in the code.

Security Notes

• Store sensitive credentials securely
• Use SSL/TLS for production environments
• Implement proper authentication and authorization
• Regularly rotate credentials

🧠 Using with Claude Desktop

To integrate your Kafka MCP server with Claude Desktop:

Install it into Claude:

mcp install <file_spec> -f <optional env file>

Note: If you are seeing dependency-related issues, please configure the claud_desktop_config.json with the absolute path to the python and provide the file to run.

MCP Client Configuration

Add this to your MCP client configuration (e.g., Claude Desktop):

{
  "mcpServers": {
    "kafka": {
      "command": "python",
      "args": ["/path/to/server.py"]
    }
  }
}

{
"mcpServers": {
      "Kafka MCP Server": {
         "command": "/opt/miniconda3/envs/kafka-mcp/bin/python",
         "args": [
         "/path/to/server.py"
         ]
      }
   }
}

Claude will now detect and interact with the tools you've defined via @mcp.prompt() in your Python server.

Configuration

Create a Kafka properties file (e.g., kafka.properties) with your Kafka connection details:

# Basic Kafka connection
bootstrap.servers=localhost:9092
client.id=kafka-mcp-client

# Security configuration (if needed)
# security.protocol=SASL_SSL
# sasl.mechanism=PLAIN
# sasl.username=your-username
# sasl.password=your-password

# SSL configuration (if needed)
# ssl.ca.location=/path/to/ca-cert
# ssl.certificate.location=/path/to/client-cert
# ssl.key.location=/path/to/client-key

---

📂 Project Structure

kafka_mcp_server/
├── server.py               # Main FastMCP server defining tools
├── kafka_utils.py
├── requirements.txt
├── README.md
└── ...

---

🧩 Tool Debugging Tips

• Use mcp dev to inspect prompt behavior interactively
• Add print() or logging.debug() in your tool functions
• Use structured @prompt function signatures to ensure correct parsing

---

📦 Requirements

Add this to your requirements.txt:

kafka-python
mcp[cli]
pydantic
uv

---

Testing

Please run docker compose up to bring up kafka cluster with a single broker you can interact with using the bootstrap.server localhost:9092.

---

🔗 Resources

• MCP Protocol Docs
• Claude + MCP Guide
• mcp CLI Reference