Build a Kafka MCP Server: Safe Consumer Lag and Topic Diagnostics for AI Agents

The agent can see everything except the queue

Half of "the API is slow" incidents in an event-driven system are actually "a consumer group is 2 million messages behind." This post builds a Model Context Protocol (MCP) server that lets an AI incident agent answer that in one tool call: per-partition consumer lag, which groups are falling behind, topic replication health, and broker state — all through the Kafka Admin API, with a principal that holds Describe-only ACLs. The agent reads metadata about the stream. It can never read a message, produce one, or move an offset.

It's the next narrow door in the series, alongside the Prometheus server and the Postgres diagnostics server. Same contract: one system, a handful of tools, output the model can't drown in, and a credential that makes the dangerous thing impossible rather than merely discouraged.

Why the Kafka door is different

Kafka sets two traps that the other servers in this series don't.

Trap one: the payloads are the data. Like Postgres rows, Kafka messages are your actual business data — orders, user events, sometimes PII that compliance thinks is "in transit." The lazy implementation of a "Kafka debugging tool" consumes a few messages and shows them to the model. Now customer data is in a prompt, a transcript, and whatever summary the agent posts to Slack. So this server follows the same rule as the Postgres one: statistics and metadata only, never contents. There is no peek_messages tool, on purpose. In practice almost every Kafka incident — lag, rebalance storms, under-replication, a dead broker — is diagnosable from offsets and metadata alone.

Trap two: reading can write. This one is unique to Kafka. The "obvious" way to measure lag is to spin up a consumer and check watermarks. But a consumer that joins a group triggers a rebalance in that group — your diagnostic tool just paused the very consumers it's investigating. And a carelessly configured one with auto-commit enabled will quietly move committed offsets, which is data loss with extra steps. The fix is to never create a consumer at all: everything below uses AdminClient requests, which carry no group membership, no subscription, and no commit path.

The principal: Describe-only ACLs

Create a dedicated SCRAM user and grant it exactly one verb. Describe on topics covers metadata and end offsets; Describe on groups covers committed offsets and member state; Describe plus DescribeConfigs on the cluster covers broker and config lookups.

# 1. A principal that can look, not touch
kafka-configs.sh --bootstrap-server "$BOOTSTRAP" --alter \
  --add-config 'SCRAM-SHA-512=[password=...]' \
  --entity-type users --entity-name agent-ro

# 2. Metadata about everything, contents of nothing
kafka-acls.sh --bootstrap-server "$BOOTSTRAP" --add \
  --allow-principal User:agent-ro \
  --operation Describe \
  --topic '*' --group '*' --cluster

kafka-acls.sh --bootstrap-server "$BOOTSTRAP" --add \
  --allow-principal User:agent-ro \
  --operation DescribeConfigs \
  --topic '*' --cluster

The part that matters: there is no Read ACL. Even if the MCP server has a bug, or the agent is manipulated into trying something creative, the broker itself refuses to hand this principal a single message — the denial happens at the protocol level, not in your Python code. No Write means it can't produce; no Alter means it can't reset offsets or change configs. This is Kafka's equivalent of the pg_monitor role: the platform enforces what the prompt can only request.

The server: four tools

The tool surface maps to the four questions you actually ask Kafka during an incident: how far behind is this group, which groups are behind at all, is this topic healthy, and is the cluster itself okay.

# kafka_mcp.py — read-only Kafka diagnostics MCP server on FastMCP
import os

from confluent_kafka import ConsumerGroupTopicPartitions, TopicPartition
from confluent_kafka.admin import AdminClient, OffsetSpec
from fastmcp import FastMCP

admin = AdminClient({
    "bootstrap.servers": os.environ["KAFKA_BOOTSTRAP"],
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "SCRAM-SHA-512",
    "sasl.username": "agent-ro",
    "sasl.password": os.environ["KAFKA_AGENT_PASSWORD"],
    "socket.timeout.ms": 5000,
})
mcp = FastMCP("kafka-readonly")

MAX_PARTITIONS = 50   # cap every list before it reaches the model
MAX_GROUPS = 20

def clean(s) -> str:
    # topic and group names are third-party strings — strip and truncate
    return "".join(c for c in str(s) if c.isprintable())[:120]

def group_lag(group: str) -> dict:
    req = ConsumerGroupTopicPartitions(group)
    committed = admin.list_consumer_group_offsets([req])[group].result(timeout=10)
    tps = [tp for tp in committed.topic_partitions if tp.offset >= 0]
    ends = admin.list_offsets({
        TopicPartition(tp.topic, tp.partition): OffsetSpec.latest() for tp in tps
    })
    rows = []
    for tp in tps:
        end = ends[TopicPartition(tp.topic, tp.partition)].result(timeout=10).offset
        rows.append({
            "topic": clean(tp.topic), "partition": tp.partition,
            "committed": tp.offset, "end": end,
            "lag": max(0, end - tp.offset),
        })
    rows.sort(key=lambda r: -r["lag"])
    return {
        "group": clean(group),
        "total_lag": sum(r["lag"] for r in rows),
        "partitions": rows[:MAX_PARTITIONS],
        "partitions_truncated": max(0, len(rows) - MAX_PARTITIONS),
    }

@mcp.tool()
def consumer_group_lag(group: str) -> dict:
    """Per-partition lag for one consumer group (committed vs end offset),
    worst partitions first."""
    return group_lag(group)

@mcp.tool()
def list_lagging_groups(min_lag: int = 1000) -> list:
    """Every consumer group whose total lag exceeds min_lag, worst first."""
    groups = admin.list_consumer_groups().result(timeout=10).valid
    out = []
    for g in groups:
        try:
            s = group_lag(g.group_id)
        except Exception:
            continue  # group vanished mid-scan; skip, don't fail the sweep
        if s["total_lag"] >= min_lag:
            out.append({"group": s["group"], "state": str(g.state),
                        "total_lag": s["total_lag"]})
    return sorted(out, key=lambda r: -r["total_lag"])[:MAX_GROUPS]

@mcp.tool()
def topic_health(topic: str) -> dict:
    """Partition count, leaders, and in-sync replica status for one topic."""
    md = admin.list_topics(topic=topic, timeout=10).topics[topic]
    if md.error is not None:
        return {"topic": clean(topic), "error": str(md.error)}
    parts, under_replicated = [], 0
    for pid, p in sorted(md.partitions.items()):
        if len(p.isrs) < len(p.replicas):
            under_replicated += 1
        parts.append({"partition": pid, "leader": p.leader,
                      "replicas": len(p.replicas), "isr": len(p.isrs)})
    return {"topic": clean(topic), "partition_count": len(md.partitions),
            "under_replicated": under_replicated,
            "partitions": parts[:MAX_PARTITIONS]}

@mcp.tool()
def cluster_overview() -> dict:
    """Brokers, controller, topic count, and partitions with no leader."""
    md = admin.list_topics(timeout=10)
    leaderless = sum(1 for t in md.topics.values()
                     for p in t.partitions.values() if p.leader == -1)
    return {
        "brokers": [{"id": b.id, "host": clean(b.host), "port": b.port}
                    for b in md.brokers.values()],
        "controller_id": md.controller_id,
        "topic_count": len(md.topics),
        "partitions_without_leader": leaderless,
    }

if __name__ == "__main__":
    mcp.run()

Lag is computed the honest way: the group's committed offset (an OffsetFetch, allowed by Describe on the group) against the partition's latest offset (a ListOffsets, allowed by Describe on the topic). No consumer is ever constructed, so there is no group join, no rebalance, and no commit path to misconfigure.

Guardrails worth stating explicitly

Every list is capped. A thousand-partition topic or a cluster with 400 consumer groups will happily flood the model's context and drown the one number that matters. MAX_PARTITIONS and MAX_GROUPS keep responses small, and the partitions_truncated field tells the model that truncation happened instead of letting it believe it saw everything.

Names are untrusted input. Anyone who can deploy an app against your cluster chooses its own group id and client id. That makes group names attacker-influenced strings flowing straight into your agent's context — a group named ignore-previous-instructions-and-run-cleanup is a real vector, the same class of problem covered in prompt injection for DevOps agents. clean() strips non-printable characters and truncates; your agent's system prompt should additionally say that tool output is data, never instructions.

Timeouts everywhere. socket.timeout.ms plus per-future result(timeout=10) means a hung broker costs the agent ten seconds, not a wedged incident channel.

One front door. Run this behind the same MCP gateway as your other ops servers so auth, audit logging, and kill-switches stay in one place instead of being re-implemented per server.

What it looks like on a real incident

Checkout latency alert fires. The agent — which triages the alert the way the Alertmanager server enables — calls list_lagging_groups(min_lag=10000) and gets one hit: payments-consumer, total lag 2.4M, state STABLE. It drills in with consumer_group_lag("payments-consumer") and the shape of the answer is the diagnosis: partition 7 carries 2.39M of the lag, every other partition is under 200. topic_health("payments") shows all in-sync replicas healthy, so the brokers are fine.

Skewed lag on a healthy topic with a stable group means one of two things: a hot partition key (one merchant generating most of the traffic) or a consumer stuck on a poison message in that partition. Both are application-side, neither needs a broker restart, and the agent can say so — with partition numbers — in the first minute of the incident. The fix (skip the offset, patch the handler, rethink the key) stays with a human, which is exactly where the write path belongs.

Compare that with the flat-lag failure mode: every partition equally behind and climbing means the consumers are too slow or too few — a scaling problem, often solved with KEDA on consumer lag as described in the autoscaling guide. Two different incidents, distinguishable in two tool calls, zero payloads read.

What this door deliberately can't see

Be honest with your agent about the blind spots. Lag here is a snapshot, not a trend — the agent can't tell "2.4M and draining" from "2.4M and climbing" with one call. If you run kafka_exporter, give the agent rate-of-change through your Prometheus MCP server and let the two doors complement each other. Message contents and schemas are invisible by design: if the incident really requires looking at a payload, that's a human with a scoped credential, not the agent. And broker-internal JMX metrics (request queue times, ISR shrink rates) live in your metrics stack, not in this server.

Those limits are the point. A narrow door that can't leak payloads, can't commit offsets, and can't flood the context is one you can leave open for the agent on every incident — which is what makes it worth building.