Kafka

You can run a FastMCP-based server that connects to Kafka and exposes MCP tools for managing brokers, topics, consumers, and retention. This server lets you inspect topics, stream messages, manage users locally, and monitor cluster metadata, all through a consistent MCP interface.

How to use

To use the Kafka MCP Server, connect your MCP client to the default HTTP endpoint and start the local server if you plan to run it directly. You can perform common Kafka operations through dedicated MCP tools, such as listing topics, creating topics with specific configurations, tailing messages, inspecting consumer groups, and retrieving cluster information.

How to install

Prerequisites you need before installation:

• Python 3.10+
• A Python environment manager (uv is recommended)

Install and run locally using the following steps. Start by creating a virtual environment and activating it, then install dependencies and run the server.

uv python install 3.13
uv venv .venv --python 3.13
source .venv/bin/activate
uv sync
uv run python -m app.main

Additional configuration and usage notes

The server reads its settings from FastMCP ServerSettings. The default host and port are 0.0.0.0 and 8000, respectively. You can override these using environment variables FASTMCP_SERVER_HOST and FASTMCP_SERVER_PORT.

Smoke tests and clients can use the following environment variables to customize behavior and testing endpoints: MCP_URL defaults to http://localhost:8000/mcp, KAFKA_BOOTSTRAP_SERVERS defaults to localhost:9092, KAFKA_TEST_TOPIC defaults to mcp_smoke, and KAFKA_TEST_GROUP defaults to mcp_smoke_group.

Configuration

Kafka connection payloads require a connection object with these fields: bootstrap_servers, security_protocol (default PLAINTEXT), optional sasl_mechanism, sasl_username, sasl_password, ssl_cafile, ssl_certfile, ssl_keyfile, and oauth_token when using OAUTHBEARER.

Security and storage considerations

Kafka users can be stored locally for quick reference and management. User entries live in a local store and can be upserted or deleted through the MCP tools dedicated to user management.

Examples of basic MCP usage patterns

Connect to the MCP endpoint from your client and perform operations such as listing topics, inspecting topic configurations, creating topics with optional configs, tailing messages, and inspecting consumer group lag. Use the provided tools to gather broker metadata and manage local Kafka users.

Tools and endpoints overview

The server exposes a collection of MCP tools that you can call via the MCP client session. Each tool accepts a structured payload containing a Kafka connection and the specific parameters for the operation.

Troubleshooting

If you cannot reach the MCP endpoint, verify that the server is running and listening on the configured host and port. Check environment variable overrides and ensure your Kafka broker is accessible from the server environment. Review Kafka connection parameters for correctness, especially security settings when using SASL or TLS.

Available tools

health

Return a simple status object indicating the server is healthy.

list_topics

List topics with partition and replication details for the connected Kafka cluster.

create_topic

Create a new topic with optional configurations such as partitions, replication factor, and topic configs.

delete_topic

Delete an existing topic by name.

topic_configs

Fetch the configuration values for a given topic.

topic_retention

Return the retention policy value (retention.ms) for a topic.

tail_messages

Read the most recent messages from a topic up to a specified limit.

live_messages

Collect a short live stream of messages from a topic with configurable duration.

list_consumer_groups

List consumer groups with their state and member count.

consumer_group_lag

Compute lag per partition for a specific consumer group.

cluster_info

Return broker and controller metadata for the connected cluster.

list_kafka_users

List locally stored user entries used for authentication against Kafka.

upsert_kafka_user

Create or update a local user entry with specified SASL or authentication details.

delete_kafka_user

Delete a local user entry by username.