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.
- **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:
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:
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:
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.