Kafka MCP Server

!Python !License !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

Clone the repository.
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

Build the Docker image:

   docker build -t kafka-mcp .

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

kafka-mcp