Kafka MCP Server

Kafka MCP Server

Overview

FastMCP-based MCP server that connects to Kafka and exposes MCP tools for common broker, topic, consumer, and retention operations.

Features

• List/create/delete topics and inspect topic configs.
• Tail recent messages or collect a short live stream.
• List consumer groups and compute group lag.
• Inspect cluster broker metadata.
• Store and manage named Kafka users locally.

Requirements

• Python 3.10+ (Makefile defaults to 3.13).
• uv (recommended) or any Python environment manager.
• A Kafka broker (local via Podman or your own cluster).

Quick start (uv)

1. Create and activate a virtual environment:
   • uv python install 3.13
   • uv venv .venv --python 3.13
   • source .venv/bin/activate
2. Install dependencies:
   • uv sync
3. Run the server:
   • uv run python -m app.main

Default MCP SSE endpoint: http://localhost:8000/mcp

Makefile shortcuts

• Start server: make start
• Start Kafka (Podman): make kafka-start
• Stop Kafka: make kafka-stop
• Kafka logs: make kafka-logs
• Run smoke test: make test-smoke

Configuration

Server settings come from FastMCP ServerSettings. The defaults are:

• Host: 0.0.0.0
• Port: 8000

FastMCP supports environment overrides such as:

• FASTMCP_SERVER_HOST
• FASTMCP_SERVER_PORT

Smoke tests and clients can use:

• MCP_URL (default http://localhost:8000/mcp)
• KAFKA_BOOTSTRAP_SERVERS (default localhost:9092)
• KAFKA_TEST_TOPIC (default mcp_smoke)
• KAFKA_TEST_GROUP (default mcp_smoke_group)

Kafka connection payload

Most tools require a Kafka connection object with the following fields:

• bootstrap_servers (required)
• security_protocol (default PLAINTEXT)
• sasl_mechanism, sasl_username, sasl_password (optional)
• ssl_cafile, ssl_certfile, ssl_keyfile (optional)
• oauth_token (required when sasl_mechanism is OAUTHBEARER)

MCP tool catalog

Each tool name below is the MCP command. Parameter shapes match the schemas in app/schemas.py.

• health - Returns {"status":"ok"}.
  • Params: none
• list_topics - Lists topics with partition and replication info.
  • Params: connection
• create_topic - Creates a topic with optional configs.
  • Params: connection, payload (name, num_partitions, replication_factor, configs)
• delete_topic - Deletes a topic by name.
  • Params: connection, name
• topic_configs - Fetches topic configuration values.
  • Params: connection, name
• topic_retention - Returns retention.ms for a topic.
  • Params: connection, name
• tail_messages - Reads the most recent messages for a topic.
  • Params: connection, name, payload (limit)
• live_messages - Collects a short live stream of messages.
  • Params: connection, name, payload (max_messages, duration_seconds, poll_interval_ms)
• list_consumer_groups - Lists consumer groups with state and member count.
  • Params: connection
• consumer_group_lag - Computes lag per partition for a group.
  • Params: connection, group_id
• cluster_info - Returns broker and controller metadata.
  • Params: connection
• list_kafka_users - Lists locally stored user entries.
  • Params: none
• upsert_kafka_user - Creates or updates a local user entry.
  • Params: user (username, sasl_mechanism, note)
• delete_kafka_user - Deletes a local user entry.
  • Params: username

MCP command examples (Python)

from mcp import ClientSession
from mcp.client.sse import sse_client

MCP_URL = "http://localhost:8000/mcp"

connection = {
      "bootstrap_servers": "localhost:9092",
      "security_protocol": "PLAINTEXT",
      "sasl_mechanism": None,
      "sasl_username": None,
      "sasl_password": None,
      "ssl_cafile": None,
      "ssl_certfile": None,
      "ssl_keyfile": None,
      "oauth_token": None,
}

async with sse_client(MCP_URL) as (read, write):
      async with ClientSession(read, write) as session:
            await session.initialize()

            await session.call_tool("list_topics", {"connection": connection})

            await session.call_tool(
                  "create_topic",
                  {
                        "connection": connection,
                        "payload": {
                              "name": "example-topic",
                              "num_partitions": 1,
                              "replication_factor": 1,
                              "configs": {},
                        },
                  },
            )

Local storage

Kafka user entries are stored at app/data/users.json.

Testing

Smoke test all MCP tools end-to-end:

1. Start Kafka: make kafka-start
2. Start the server: make start
3. In a new terminal: make test-smoke

LocalAI (optional)

LocalAI is not required for Kafka MCP usage, but this repo includes helper targets for running models locally:

1. Install LocalAI and a model: make localai-install && make localai-model
2. Start LocalAI: make localai-start