wklee610/kafka-mcp

---

An MCP server implementation for Kafka, allowing LLMs to interact with and manage Kafka clusters.

Features

• Cluster Management: View broker details describe_cluster, describe_brokers.
• Topic Management: List list_topics, create create_topic, delete delete_topic, describe describe_topic, and increase partitions create_partitions.
• Configuration Management: View describe_configs and modify alter_configs dynamic configs for topics, brokers, and groups.
• Consumer Groups: List list_consumer_groups, describe describe_consumer_group, and securely manage offsets with reset_consumer_group_offset and rewind_consumer_group_offset_by_timestamp. Advanced tools include state validation, dry runs, and execution audit logging.
• Messaging: Consume messages consume_messages (from beginning, latest, or specific offsets) and produce messages produce_message.

Prerequisites

• Python 3.10+
• uv package manager (recommended)
• A running Kafka cluster (e.g., local Docker, Confluent Cloud, etc.)

Installation

1. Clone the repository.
2. Install dependencies:

   uv sync
   

Configuration

The server requires the KAFKA_BOOTSTRAP_SERVERS environment variable.

• KAFKA_BOOTSTRAP_SERVERS: Comma-separated list of broker urls (e.g., localhost:9092).
• KAFKA_CLIENT_ID: (Optional) Client ID for connection (default: kafka-mcp).

Usage

Running the Server

You can run the server directly using uv or python, or use Docker.

Using uv (Recommended)

export KAFKA_BOOTSTRAP_SERVERS=localhost:9092
uv run kafka-mcp

Using Docker

1. Build the Docker image:

   docker build -t kafka-mcp .
   

2. Run the container:

   docker run -i --rm -e KAFKA_BOOTSTRAP_SERVERS=host.docker.internal:9092 kafka-mcp
   

   (Note: Use host.docker.internal instead of localhost if your Kafka cluster is running on the host machine.)

Claude Desktop Configuration

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

{
  "mcpServers": {
    "kafka": {
      "command": "<uv PATH>",
      "args": [
        "--directory",
        "<kafka-mcp PATH>",
        "run",
        "kafka-mcp"
      ],
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "localhost:9092"
      }
    }
  }
}

Debugging / Development

To verify that the server can start and connect to your Kafka cluster (ensure your Kafka is running first):

# Set your bootstrap server
export KAFKA_BOOTSTRAP_SERVERS=localhost:9092

# Run a quick check
uv run python -c "from src.kafka_mcp import main; print('Imports successful')"

Available Tools

Category	Tool Name	Description
Cluster	describe_cluster	Get cluster metadata (controller, brokers).
	describe_brokers	List all brokers.
Topics	list_topics	List all available topics.
	describe_topic	Get detailed info (partitions, replicas) for a topic.
	create_topic	Create a new topic with partitions/replication factor.
	delete_topic	Delete a topic.
	create_partitions	Increase partitions for a topic.
Configs	describe_configs	View dynamic configs for topic/broker/group.
	alter_configs	Update dynamic configs.
Consumers	list_consumer_groups	List all active consumer groups.
	describe_consumer_group	Get members and state of a group.
	get_consumer_group_offsets	Get committed offset, high/low watermarks, and calculate total lag for a topic.
	reset_consumer_group_offset	Safely change consumer group offsets to earliest, latest, or a specific offset.
	rewind_consumer_group_offset_by_timestamp	Rewind/advance consumer group offsets securely based on a timestamp.
Messages	consume_messages	Consume messages from a topic (supports offsets, limits).
	produce_message	Send a message to a topic.

Project Structure

src/kafka_mcp/
├── configs/       # Configuration handling
├── connections/   # Kafka client factories (singleton)
├── tools/         # Tool implementations
│   ├── admin.py     # Topic & Config management
│   ├── cluster.py   # Cluster metadata
│   ├── consumer.py  # Consumer group & message consumption
│   └── producer.py  # Message production
└── main.py        # Entry point & MCP tool registration

Troubleshooting

• Connection Refused: Ensure KAFKA_BOOTSTRAP_SERVERS is correct and reachable.

TODO

• SASL
• JMX