Kafka集群管理服务器

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

1. Clone the repository.
2. Install dependencies:
   复制

   uv sync
   

Configuration

Only KAFKA_BOOTSTRAP_SERVERS is required. Existing PLAINTEXT configurations remain unchanged.

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:

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:

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:

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:

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:

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.

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:

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)

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.

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

Project Structure

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