{"id":"871a33fe-70e7-4e27-af67-163300bfca92","slug":"kafka-ops-mcp","kind":"mcp","name":"kafka-ops-mcp","summary":"Safe-by-default MCP server for Kafka: scoped read tools, redacted DLQ browsing, and replay gated by dry-run plans and single-use approval tokens. Java 21, Spring AI.","owner":"gratycodes","repoUrl":"https://github.com/gratycodes/kafka-ops-mcp","license":"MIT","licenseTier":"permissive","trust":"community","stars":0,"match":null,"fewStars":true,"format":"mcp","createdAt":"2026-10-10T17:41:34.000Z","addedAt":"2026-10-21T06:56:59.038Z","logoUrl":"https://avatars.githubusercontent.com/u/107890027?v=4","linkOnly":false,"collection":null,"skillPath":null,"categories":["cloud-infra","memory-knowledge"],"tags":["agentic-ai","ai-agents","apache-kafka","dead-letter-queue","java","kafka","llm-security","opentelemetry"],"runtime":"binary","local":true,"cloudCapable":false,"cloudCapability":"unknown","cloudReadyOnPlatform":false,"needsApiKey":false,"reviewed":false,"syncedAt":"2026-10-28T01:44:09.337Z","sample":false,"url":"https://conduits.dev/tools/kafka-ops-mcp","language":"Java","transports":["stdio"],"install":{},"requirements":{"env":[],"upstreamMayCharge":false},"tools":null,"readme":"# kafka-ops-mcp\n\n**Let an AI agent triage your Kafka dead-letter queue, without letting it break anything.**\n\nA safe-by-default [MCP](https://modelcontextprotocol.io) server for Apache Kafka. Agents get six narrow tools: list and describe topics, check consumer lag, browse a DLQ with secrets and PII redacted, and plan a replay. The one write, replaying DLQ records, only runs from a dry-run plan a human has reviewed, using a short-lived single-use approval token.\n\nJava 21 · Spring Boot 4 · Spring AI MCP server (official MCP Java SDK) · Kafka clients 4.2 · OpenTelemetry · MIT\n\n\n\n*Real run: a scripted MCP client over stdio against local Kafka 4.1 (KRaft) with synthetic records.*\n\n## Why I built it\n\nEvery payments team I have worked on has had the same 2 a.m. problem: a consumer fails, records pile up in `orders.DLT`, and someone pipes `kafka-console-consumer` into grep and then runs a hand-written replay script against production. An AI agent is genuinely good at the reading part (\"why are these 40 orders here, and which are safe to retry?\"). But most Kafka MCP servers give the agent the whole cluster, and a DLQ holds exactly what you least want in a model's context: card data, tokens in headers, and free text that may contain instructions aimed at the agent.\n\nSo this server is narrow on purpose: scoped, allowlisted, audited, and redacted before any byte leaves the process.\n\n## Quickstart\n\nRequires JDK 21 and a Kafka broker on `localhost:9092`. This is the setup used to record the demo above.\n\n```bash\n# 1. Local single-node Kafka in KRaft mode (Apache Kafka binary; any 3.x/4.x broker works)\ncurl -LO https://archive.apache.org/dist/kafka/4.1.0/kafka_2.13-4.1.0.tgz\ntar xzf kafka_2.13-4.1.0.tgz && cd kafka_2.13-4.1.0\nbin/kafka-storage.sh format --standalone -t \"$(bin/kafka-storage.sh random-uuid)\" -c config/server.properties\nbin/kafka-server-start.sh -daemon config/server.properties\ncd ..\n# (or, with Docker: docker run -d -p 9092:9092 apache/kafka:4.2.1)\n\n# 2. Build (also runs the unit and embedded-Kafka tests)\ngit clone https://github.com/Gratycodes/kafka-ops-mcp && cd kafka-ops-mcp\nmvn -B package\n\n# 3. Run over stdio (Claude Desktop, Cursor). Read-only; only orders* topics visible.\njava -jar target/kafka-ops-mcp-0.1.0-SNAPSHOT.jar --spring.profiles.active=stdio \\\n --kafka-ops.bootstrap-servers=localhost:9092 --kafka-ops.topic-allowlist='orders*'\n```\n\nAdd it to your MCP client with [`examples/mcp-client-stdio.json`](examples/mcp-client-stdio.json) (set the absolute jar path), then ask: *\"What's in orders.DLT, and which records are safe to replay?\"*\n\nTo enable replays, add the write scope: `--kafka-ops.scopes=topics:read,groups:read,dlq:read,dlq:replay` (or see [`examples/application-local.yml`](examples/application-local.yml)). For Streamable HTTP with bearer auth, see [Run it](#run-it).\n\n## Tools\n\n| Tool | Scope | What it does |\n| --- | --- | --- |\n| `list_topics` | `topics:read` | Allowed topics with partitions and replication factor. Others are counted, not named. |\n| `describe_topic` | `topics:read` | Partitions, leader, replicas, ISR, earliest/latest offsets, key configs. |\n| `consumer_group_lag` | `groups:read` | Group state, members, lag per partition. Non-allowed partitions hidden. |\n| `browse_dlq` | `dlq:read` | Recent records or from an offset, with filters. Redacted, injection-flagged, never commits. |\n| `replay_dlq_dry_run` | `dlq:read` | Builds a replay plan and (if writes are enabled) returns an approval token. Writes nothing. |\n| `replay_dlq` | `dlq:replay` | Executes a plan. Accepts only the approval token. |\n\nRead tools carry `readOnlyHint=true`; `replay_dlq` carries `destructiveHint=true`, so clients that confirm destructive calls will ask the user.\n\n## Architecture\n\n```mermaid\nflowchart LR\n Agent[\"MCP client / agent\"] -- \"stdio or Streamable HTTP
(optional bearer token)\" --> Tools\n subgraph server[\"kafka-ops-mcp\"]\n Tools[\"KafkaOpsTools
(@McpTool adapters)\"] --> Runner\n Runner[\"ToolRunner
scope check · execute_tool span · audit\"] --> Service\n Service[\"KafkaOpsService
allowlist · DLQ patterns · plans\"] --> Redactor[\"Redactor +
injection heuristics\"]\n Service --> Approvals[\"ApprovalTokenService
hashed, single-use, TTL\"]\n Service --> Gateway[\"KafkaGateway
Admin · groupless consumer · idempotent producer\"]\n Runner --> Audit[(\"audit JSONL\")]\n Runner --> OTel[(\"OpenTelemetry
(no-op unless configured)\")]\n end\n Gateway --> Kafka[(\"Kafka\")]\n```\n\n### Replay flow\n\n```mermaid\nsequenceDiagram\n participant A as Agent\n participant H as Human\n participant S as kafka-ops-mcp\n participant K as Kafka\n A->>S: replay_dlq_dry_run(orders.DLT, p0, 100..140)\n S->>K: read offsets 100..140 (no group, no commit)\n S->>S: resolve target per record, skip unsafe ones,
fingerprint each record, store plan\n S-->>A: plan + approvalToken (expires in 5 min)\n A->>H: \"41 records → orders, 2 skipped. Approve?\"\n H-->>A: yes\n A->>S: replay_dlq(approvalToken)\n S->>S: audit ATTEMPT (abort if it cannot be written)\n S->>K: re-read 100..140, compare fingerprints\n alt anything changed or vanished\n S-->>A: error, nothing written\n else unchanged\n S->>K: produce copies (acks=all, idempotent)\n S-->>A: replayed 41\n end\n```\n\n## Design decisions\n\n- **One choke point for every tool.** All calls go through `ToolRunner`, so a new tool cannot skip its scope check, trace span or audit line. Kafka and policy logic live in `KafkaOpsService`, which has no MCP dependency and is tested directly.\n- **Deny by default.** The topic allowlist starts empty; internal topics are denied even under `*`; DLQ tools also require a DLQ name pattern, so a broad allowlist does not turn into \"browse any business topic\".\n- **The token is the only argument to the write.** Offsets and targets are frozen at dry-run time, so nothing can change between human review and execution. Only the token's SHA-256 is stored; it expires, works once and is bound to the principal.\n- **Fingerprint, then re-check.** Records are re-read and compared at execution; if anything changed or aged out, nothing is written. The DLQ itself is never modified.\n- **Write-ahead audit.** If the `ATTEMPT` line cannot be written, the replay does not run.\n- **Untrusted content is structural, not heuristic.** Record content comes back as data with `contentTrust: \"untrusted\"`; no tool interprets it. Injection flags are a hint, not the guarantee.\n- **Redact before filter.** `browse_dlq` filters run on redacted text, so a filter cannot be used to probe for secrets.\n- **Observability without lock-in.** Spans follow the OpenTelemetry GenAI semantic conventions and depend only on the OTel API, so they are no-ops until you attach an SDK or agent.\n\n## Safety model\n\n**Read-only unless an operator says otherwise.** `dlq:replay` is the only write scope and is not in the default\nscope list. On a read-only instance `replay_dlq` is refused before it looks at the token, and the dry run returns\nan informational plan with no token.\n\n**Nothing is visible by default.** `topic-allowlist` is empty out of the box, which denies every topic. Internal\ntopics (`__consumer_offsets`, `__transaction_state`, ...) are denied even under `*`. DLQ tools additionally require\nthe topic to match `dlq-topic-patterns`, so a broad allowlist does not let an agent \"browse\" a business topic.\n\n**Approval tokens.** A dry run stores the plan on the server and returns a random token (`kopr_` + 192 bits).\nOnly its SHA-256 is kept. The token:\n- expires after `replay.token-ttl` (default 5 minutes),\n- works once, and is burned on a failed attempt by a different principal,\n- is bound to the principal that made the plan,\n- is the only argument `replay_dlq` accepts, so offsets and targets cannot be changed after review.\n\n**Replays are conservative.** Each record's target comes from its original-topic header (Spring Kafka's\n`kafka_dlt-original-topic`, Kafka Connect's `__connect.errors.topic`, configurable) or an explicit target. Records\nare skipped, with a reason, if the target is missing, not allowlisted, internal, or itself a DLQ. At execution\nthe records are re-read and compared against fingerprints taken at plan time (key, value and headers); if any\nchanged or were deleted by retention, nothing is written. Replayed copies keep key, value and headers, drop the\nDLQ error headers, and gain `kafka-ops-replay-of` and `kafka-ops-replay-plan` for traceability. The DLQ itself is\nnever modified.\n\n**Redaction before anything leaves.** Applied to keys, payloads, header values, error messages and the audit log:\n- headers named like `authorization`, `cookie`, `*token*`, `*secret*`, `*api-key*`, ... are dropped entirely;\n- JSON fields are matched on a normalised name, so `cardNumber`, `card_number` and `card-number` all hit\n `*cardnumber*`; defaults cover credentials, card data, bank details, email, phone, date of birth, national ids;\n- every remaining string is scanned for JWTs, bearer/basic credentials, AWS access keys, PEM private keys,\n `password=...` style pairs, emails, and 13–19 digit numbers that pass a Luhn check (so order ids survive);\n- binary payloads are not decoded, long payloads are truncated after redaction.\n\nFilters in `browse_dlq` run on the redacted text, so `contains=\"hunter2\"` cannot be used to test whether a secret\nis present.\n\n**DLQ content is data, not instructions.** Results are JSON with record content in string fields, plus\n`contentTrust: \"untrusted\"` and a notice telling the agent not to follow instructions inside records. A small set\nof heuristics flags records that look like prompt injection (`ignore previous instructions`, role overrides,\nrequests to call tools, mentions of approval tokens, exfiltration requests). The flags are a hint for the agent\nand the human reading the transcript; the actual guarantee is structural: no tool executes or interprets content,\nand a write needs a token that only a dry run can mint.\n\n**Audit log.** One JSON line per call to `KAFKA_OPS_AUDIT_FILE` (default `./kafka-ops-audit.jsonl`):\n\n```json\n{\"timestamp\":\"2026-10-10T17:39:39.49Z\",\"principal\":\"oncall-agent\",\"client\":\"example-client/1.0\",\n \"sessionId\":\"d113c6a7-...\",\"tool\":\"replay_dlq\",\"scope\":\"dlq:replay\",\"args\":{\"approvalToken\":\"[REDACTED:field]\"},\n \"outcome\":\"OK\",\"reason\":null,\"summary\":\"plan 6f0c...: completed, replayed 41\",\"durationMs\":212,\"traceId\":\"4bf9...\"}\n```\n\nArguments are redacted and results are summarised, not copied, so the audit log is not a second copy of the DLQ.\nWrites get a write-ahead `ATTEMPT` line; if that cannot be written, the replay does not run.\n\n**Tracing.** Each call is an `execute_tool {tool}` span (kind `INTERNAL`) following the OpenTelemetry GenAI\nsemantic conventions: `gen_ai.operation.name=execute_tool`, `gen_ai.tool.name`, `gen_ai.tool.type=function`,\n`gen_ai.tool.description`, `error.type` on failure, plus `mcp.method.name` and `mcp.session.id`. Tool arguments\n(`gen_ai.tool.call.arguments`, redacted) are opt-in via `kafka-ops.capture-tool-arguments`. The server only\ndepends on the OpenTelemetry API: without an SDK or the Java agent the spans are no-ops.\n\n**HTTP transport.** Binds to `127.0.0.1` by default. Set `kafka-ops.http.bearer-token-sha256` (the hex SHA-256 of\na token, not the token) to require `Authorization: Bearer `; comparison is constant-time.\n\n## Configuration\n\nEverything lives under `kafka-ops.*`. See [`examples/application-local.yml`](examples/application-local.yml):\n\n```yaml\nkafka-ops:\n bootstrap-servers: localhost:9092\n principal: oncall-agent\n scopes: [topics:read, groups:read, dlq:read] # add dlq:replay to allow replays\n topic-allowlist: [\"orders.*\", \"payments.refunds*\"]\n dlq-topic-patterns: [\"*.DLT\", \"*-dlq\"]\n replay:\n token-ttl: 5m\n max-messages: 100\n http:\n bearer-token-sha256: ${KAFKA_OPS_BEARER_TOKEN_SHA256:}\n```\n\nKafka client security settings (`security.protocol`, `sasl.*`, `ssl.*`) go under `kafka-ops.client` and are\npassed through unchanged.\n\n## Run it\n\nRequires JDK 21.\n\n```bash\nmvn -B package\n```\n\n**stdio** (Claude Desktop, Cursor, most local clients). Logs go to stderr so stdout stays clean for the protocol:\n\n```bash\njava -jar target/kafka-ops-mcp-0.1.0-SNAPSHOT.jar --spring.profiles.active=stdio \\\n --kafka-ops.bootstrap-servers=localhost:9092 --kafka-ops.topic-allowlist='orders.*'\n```\n\nClient config: [`examples/mcp-client-stdio.json`](examples/mcp-client-stdio.json).\n\n**Streamable HTTP** on `http://127.0.0.1:8080/mcp`:\n\n```bash\nexport KAFKA_OPS_BEARER_TOKEN_SHA256=$(printf '%s' \"$KAFKA_OPS_TOKEN\" | sha256sum | cut -d' ' -f1)\njava -jar target/kafka-ops-mcp-0.1.0-SNAPSHOT.jar \\\n --spring.config.additional-location=examples/application-local.yml\n```\n\nClient config: [`examples/mcp-client-http.json`](examples/mcp-client-http.json).\n\n**Traces**: attach the OpenTelemetry Java agent and point it at a collector, e.g.\n`java -javaagent:opentelemetry-javaagent.jar -Dotel.service.name=kafka-ops-mcp -jar ...`.\n\n## Tests\n\n```bash\nmvn -B verify\n```\n\n| Suite | What it covers |\n| --- | --- |\n| `ToolPolicyTest`, `RedactorTest`, `InjectionHeuristicsTest`, `ApprovalTokenServiceTest` | Scopes, allowlist, globs, redaction rules, Luhn, token expiry/reuse/principal binding |\n| `ToolRunnerTest` | Audit outcomes, write-ahead audit, span names and GenAI attributes (in-memory exporter) |\n| `EmbeddedKafkaScenariosTest` | 17 end-to-end scenarios on an in-JVM KRaft broker: lag, browse, filters, dry run, replay, skips, stale plans |\n| `PromptInjectionTest` | Malicious instructions, JSON breakout, forged tokens and hostile headers planted in DLQ records |\n| `McpServerEndToEndTest` | Boots the app and drives it with the official MCP Java client over Streamable HTTP, including bearer auth |\n| `ContainerKafkaScenariosTest` | The same 17 scenarios on `apache/kafka:4.2.1` via Testcontainers. Skipped without Docker; CI fails if it was skipped. |\n\nAll test data is synthetic.\n\n## How it compares\n\nChecked against each project's README in October 2026. Corrections welcome.\n\n| | Scope of tools | Write controls | DLQ inspect / replay |\n| --- | --- | --- | --- |\n| [confluentinc/mcp-confluent](https://github.com/confluentinc/mcp-confluent) (TypeScript) | Very broad: Kafka, Flink SQL, Connect, Schema Registry, Tableflow, Confluent Cloud | Per-connection `read_only` flag; HTTP auth with OAuth for Confluent Cloud | No dedicated DLQ tools |\n| [tuannvm/kafka-mcp-server](https://github.com/tuannvm/kafka-mcp-server) (Go) | Produce/consume, topics, groups, cluster health | OAuth 2.1 on the HTTP transport; I did not find a read-only mode in its README | No |\n| [denizgursoy/kafka-mcp](https://github.com/denizgursoy/kafka-mcp) (Go) | Broad debugging toolset across clusters | Read-only endpoints, per-tool disabling, previews and `confirm` on writes, audit logging | Has message search; no DLQ replay workflow that I found |\n| [conduktor/kafka-mcp](https://github.com/conduktor/kafka-mcp) | Ships inside Conduktor Console; fleet-wide ops | Inherits Console RBAC and audit log | Record lookup; part of a commercial product |\n| **kafka-ops-mcp** | Six tools, DLQ-focused | Scopes, allowlist, plan + single-use token, write-ahead audit | Yes: redacted browse, fingerprinted plans, guarded replay |\n\nIf you want a general Kafka assistant, use one of the first three. This project is for the narrower case where an\nagent should help triage and replay a DLQ and you want the write path to be boring and reviewable.\n\n## Limitations\n\n- **Scopes are per server instance, not per user.** Run separate instances (read-only for agents in general, a\n replay-enabled one for on-call) rather than expecting per-caller permissions. The bearer token authenticates\n the client; it does not carry scopes. OAuth 2.1 / per-user identity is not implemented.\n- **Approval tokens live in memory.** A restart drops pending approvals (deliberately), and the server is not\n designed to run as several replicas behind a load balancer.\n- **The \"human approval\" step is a protocol, not enforcement.** In the default setup the token goes back to the\n agent; an agent that ignores instructions could call `replay_dlq` without asking. The guarantees are: no replay\n without a dry run, no change between review and execution, no writes on read-only instances. Clients that\n confirm destructive tools (`destructiveHint`) close the gap; delivering the token out-of-band is future work.\n- **Replay is not transactional.** Records are sent in order with an idempotent producer and the result reports\n partial progress, but a failure halfway leaves the earlier records replayed.\n- **Redaction is pattern matching.** It covers common shapes, not every possible secret or PII format, and does\n not decode Avro/Protobuf (binary payloads are hidden rather than shown). Add your own field names.\n- **Injection flags are heuristics** and will miss things. They are not the safety mechanism; the tool design is.\n- **Single partition per plan**, up to `replay.max-messages` offsets. Large backfills belong in a proper job.\n- Not tested against Confluent Cloud, MSK IAM auth, or clusters with ACLs; client settings pass through, but the\n tests use plaintext brokers.\n\n## License\n\nMIT, see [LICENSE](LICENSE).\n"}