What can you do with this server?

kafka-mcp is an MCP server that enables LLMs to fully manage and interact with Apache Kafka clusters across the following areas: Cluster Management * describe_cluster: Retrieve cluster metadata including brokers and controller info * describe_brokers: List all brokers with their details Topic Management * list_topics, describe_topic, create_topic, delete_topic, create_partitions (increase only) Configuration Management * describe_configs / alter_configs: View and update dynamic configs for topics, brokers, or consumer groups Consumer Group Management * list_consumer_groups, describe_consumer_group: List and inspect group state and members * get_consumer_group_offsets: Retrieve committed offsets, watermarks, and lag per partition * reset_consumer_group_offset: Reset offsets to earliest, latest, or a specific offset — with dry-run and force options * rewind_consumer_group_offset_by_timestamp: Rewind/advance offsets by timestamp — with dry-run and force options Messaging * consume_messages: Read messages from a topic with configurable offset, partition, limit, and timeout * produce_message: Publish a message to a topic with optional key and headers

Which integrations are available for this server?

Provides tools for managing and interacting with Apache Kafka clusters, enabling users to perform cluster metadata lookups, manage topics and partitions, modify configurations, monitor consumer groups, and produce or consume messages.

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

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