Kafka MCP Server

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.
Secure Connections: Connect with TLS/mTLS and SASL PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI, or OAUTHBEARER.
JMX Monitoring: Read broker and topic metrics from one or more Prometheus JMX Exporter endpoints and summarize cluster health.

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:
bash
```
uv sync
```

Configuration

Only KAFKA_BOOTSTRAP_SERVERS is required. Existing PLAINTEXT configurations remain unchanged.

Variable	Required	Description
`KAFKA_BOOTSTRAP_SERVERS`	Yes	Comma-separated broker addresses, such as `localhost:9092`.
`KAFKA_CLIENT_ID`	No	Kafka client ID. Defaults to `kafka-mcp`.
`KAFKA_SECURITY_PROTOCOL`	No	`PLAINTEXT`, `SSL`, `SASL_PLAINTEXT`, or `SASL_SSL`. librdkafka defaults to `PLAINTEXT` when omitted.
`KAFKA_CLIENT_CONFIG_JSON`	No	JSON object of additional scalar confluent-kafka/librdkafka client properties. Explicit variables and connection safety settings take precedence.

SASL

Set KAFKA_SASL_MECHANISM to PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI, or OAUTHBEARER whenever a SASL security protocol is used.

PLAIN and SCRAM use:

server.ts

export KAFKA_SECURITY_PROTOCOL=SASL_SSL
export KAFKA_SASL_MECHANISM=SCRAM-SHA-512
export KAFKA_SASL_USERNAME="$KAFKA_USERNAME"
export KAFKA_SASL_PASSWORD="$KAFKA_PASSWORD"

GSSAPI/Kerberos supports these optional variables:

Variable	librdkafka property
`KAFKA_SASL_KERBEROS_SERVICE_NAME`	`sasl.kerberos.service.name`
`KAFKA_SASL_KERBEROS_PRINCIPAL`	`sasl.kerberos.principal`
`KAFKA_SASL_KERBEROS_KEYTAB`	`sasl.kerberos.keytab`
`KAFKA_SASL_KERBEROS_KINIT_CMD`	`sasl.kerberos.kinit.cmd`
`KAFKA_SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN`	`sasl.kerberos.min.time.before.relogin`

The prebuilt confluent-kafka wheels do not include GSSAPI. Kerberos deployments outside Docker must install confluent-kafka against a librdkafka build compiled with SASL GSSAPI support; the provided Docker image builds that variant from source and includes the Kerberos runtime tools.

OAUTHBEARER supports librdkafka's built-in OIDC flow:

server.ts

export KAFKA_SECURITY_PROTOCOL=SASL_SSL
export KAFKA_SASL_MECHANISM=OAUTHBEARER
export KAFKA_SASL_OAUTHBEARER_METHOD=oidc
export KAFKA_SASL_OAUTHBEARER_CLIENT_ID="$OAUTH_CLIENT_ID"
export KAFKA_SASL_OAUTHBEARER_CLIENT_SECRET="$OAUTH_CLIENT_SECRET"
export KAFKA_SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL=https://idp.example.com/oauth/token
export KAFKA_SASL_OAUTHBEARER_SCOPE=kafka

KAFKA_SASL_OAUTHBEARER_EXTENSIONS is optional. librdkafka's default unsecured-token handler is for development and testing only; it requires both KAFKA_SASL_OAUTHBEARER_CONFIG and KAFKA_ENABLE_SASL_OAUTHBEARER_UNSECURE_JWT=true instead of the OIDC variables.

Metadata and client-assertion flows are also supported. Set KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_TYPE=azure_imds for Azure IMDS. AWS IAM uses aws_iam with KAFKA_SASL_OAUTHBEARER_CONFIG; it additionally requires confluent-kafka[oauthbearer-aws] 2.15 or newer. JWT assertion properties such as sasl.oauthbearer.grant.type and sasl.oauthbearer.assertion.private.key.file can be passed through KAFKA_CLIENT_CONFIG_JSON, so the base 2.13 client remains compatible.

TLS and mTLS

The following variables map directly to librdkafka TLS settings:

Variable	Description
`KAFKA_SSL_CA_LOCATION`	CA certificate path. The system CA store is used when omitted.
`KAFKA_SSL_CERTIFICATE_LOCATION`	Client certificate path for mTLS.
`KAFKA_SSL_KEY_LOCATION`	Client private key path for mTLS.
`KAFKA_SSL_KEY_PASSWORD`	Optional private key password.
`KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM`	Hostname verification algorithm, usually `https`.
`KAFKA_SSL_CRL_LOCATION`	Certificate revocation list path.

The client certificate and key must be configured together. Less common librdkafka properties, including PKCS#12 keystores and cipher controls, can be supplied through KAFKA_CLIENT_CONFIG_JSON:

server.ts

export KAFKA_SECURITY_PROTOCOL=SSL
export KAFKA_CLIENT_CONFIG_JSON='{"ssl.keystore.location":"/run/secrets/client.p12","ssl.keystore.password":"secret"}'

JMX Exporter

JMX monitoring is optional and does not affect the Kafka tools. Run Prometheus JMX Exporter as a Java agent on each broker, then provide its HTTP /metrics endpoints. Keeping Java RMI disabled avoids exposing Kafka's unauthenticated remote JMX port.

server.ts

export KAFKA_JMX_EXPORTER_ENDPOINTS='{
  "broker-1": "http://broker-1:7071/metrics",
  "broker-2": "http://broker-2:7071/metrics",
  "broker-3": "http://broker-3:7071/metrics"
}'

The value may also be a JSON array, a comma-separated URL list, or a comma-separated name=url list. For HTTPS or an authenticated reverse proxy, use:

Variable	Description
`KAFKA_JMX_TIMEOUT_SECONDS`	Per-request timeout. Defaults to `5`.
`KAFKA_JMX_VERIFY_SSL`	Verify HTTPS certificates. Defaults to `true`.
`KAFKA_JMX_CA_LOCATION`	Custom CA certificate path.
`KAFKA_JMX_CLIENT_CERTIFICATE_LOCATION`	HTTPS client certificate path.
`KAFKA_JMX_CLIENT_KEY_LOCATION`	HTTPS client key path.
`KAFKA_JMX_CLIENT_KEY_PASSWORD`	Optional HTTPS client key password.
`KAFKA_JMX_USERNAME` / `KAFKA_JMX_PASSWORD`	HTTP Basic authentication.
`KAFKA_JMX_BEARER_TOKEN`	HTTP Bearer authentication.
`KAFKA_JMX_HEADERS_JSON`	Additional HTTP headers as a JSON object of strings.

Basic and Bearer authentication are mutually exclusive. JMX tools return a structured not_configured response when no exporter endpoints are set.

Usage

Running the Server

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

Using uv (Recommended)

server.ts

export KAFKA_BOOTSTRAP_SERVERS=localhost:9092
uv run kafka-mcp

Using Docker

The image supports TLS, PLAIN, SCRAM, OAUTHBEARER/OIDC, and GSSAPI. Its multi-stage build compiles the Python client against a checksum-verified librdkafka release with Kerberos support.

Build the Docker image:
Terminal
```
docker build -t kafka-mcp .
```
Run the container:
Terminal
```
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):

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):

server.ts

# 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.
Metrics	`describe_cluster_health`	Summarize replication, ISR, controller, log-directory, and broker reachability signals.
	`get_broker_metrics`	Read curated broker metrics or filtered raw exporter samples.
	`get_topic_metrics`	Read metrics carrying the requested topic label.
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

Code

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

Troubleshooting

Connection Refused: Ensure KAFKA_BOOTSTRAP_SERVERS is correct and reachable.
SASL Authentication Failed: Verify the security protocol, mechanism, credentials, and CA trust path together.
JMX Status Is Unknown: Ensure the exporter rules include Kafka broker/controller MBeans. Raw samples remain available through get_broker_metrics(include_raw=true).