Kafka MCP Enterprise Server

Scope: This is a stdio, in-memory conformance reference for the KIP-1318 security control model. Real brokers, Streamable HTTP, OAuth 2.1, EOS/fencing, Connect tooling, distributed rate/breaker state, and durable audit topics are specified in the KIP (Java production track under KAFKA-20436) and are not implemented here. Full matrix: doc/kip-alignment.md.

Implemented vs Spec-only

Website

PyPI

pip install kafka-mcp-enterprise
echo {"jsonrpc":"2.0","id":1,"method":"tools/list"} | kafka-mcp-enterprise

Agents & skills (Cursor, Kiro, ChatGPT, Gemini, Copilot, …)

Python conformance reference for KIP-1318: a Model Context Protocol (MCP) control plane for agent access to Kafka. The KIP's production target is Java. This package validates the fail-closed security model over stdio against an in-memory backend - it is not a drop-in for the forthcoming first-party Java MCP server.

Scope clarity: The Apache Kafka project tracks the official implementation under KAFKA-20436. This repository is an independent, zero-dependency reference that encodes the enterprise control plane, error model, and conformance tests so designs can be validated before or alongside the Java work. It is not a drop-in replacement for the forthcoming first-party Java MCP server.

Why this exists

AI agents need governed Kafka access-not ad-hoc scripts, unbounded consumes, or shared "god" principals. KIP-1318 proposes a standalone MCP process (stdio / HTTP in the spec) that exposes tools and kafka:// resources without changing the Kafka wire protocol. This reference implements the fail-closed security control plane over stdio against an in-memory backend (see matrix above).

Broker ACLs remain authoritative. Guardrails here complement them; they never replace them.

What’s included

Documentation product image (not a web UI). Checklist of what this Python stdio reference implements: fail-closed pipeline, DLP, 13 tools, kafka:// resources, tests, and examples.

How it runs

flowchart LR
  Agent["Agent / IDE"]
  MCP["kafka-mcp-enterprise<br/>stdio JSON-RPC + pipeline"]
  Mem["InMemoryKafka<br/>conformance backend"]
  Agent -->|"MCP tools/call"| MCP
  MCP -->|"tools / resources"| Mem

This repo: Python stdlib reference (in-memory Kafka for tests). KIP production track: Java wrapping real Admin/Producer/Consumer (KAFKA-20436), not shipped here.

Fail-closed pipeline (overview)

flowchart TB
  A["1 Auth"] --> B["2 Deny-list"] --> C["3 Allow / readonly"]
  C --> D["4 Scope"] --> E["5 Policy"] --> F["6 Taint / IFC"]
  F --> G["7 Approval"] --> H["8 Rate limit"] --> I["9 Breaker execute"]

Full detail: doc/security-controls.md · architecture: doc/architecture.md.

Complete features, security controls & guardrails

Nothing below is optional marketing - every item is implemented in kafka_mcp/ and covered by the 302/302 conformance suite and/or demos/examples unless noted as a documented reference gap.

A. MCP protocol & surface

B. Tools (13) - classified

Shipped default exposes the six read tools only (SECURE_DEFAULT_TOOLS).

C. Resources (kafka://)

D. Fail-closed security pipeline (exact order)

Every tools/call - first denial wins:

Pre / around execute (also enforced):

E. Data-protection guardrails (DLP)

F. Approval, taint & IFC

G. Scoping, exposure & identity

H. Resilience & blast-radius controls

I. Consume semantics (Direct Partition Assignment)

J. Backend surface (in-memory Kafka)

create_topic, delete_topic, list_topics, describe_topic, alter_topic_config, produce, consume, list_groups, describe_group, group_lag, create_acls, list_acls, describe_cluster, principal ACLs, rebalance counter, dependency hooks.

K. Audit

L. Error codes (complete - 15)

M. Configuration surface (32 fields)

bootstrap_servers, transport, tools_allowed, tools_denied, readonly, allowed_topic_prefixes, allowed_group_prefixes, taint_guard_enabled, approval_required_tools, dryrun_tools, audit_topic, policy_engine, circuit_breaker_enabled, dependency_timeout_ms, rate_requests_per_second, rate_admin_requests_per_second, oauth_expected_audience, oauth_expected_issuer, approval_signing_secret, redaction_enabled, dlp_mode, dlp_block_categories, scrub_all_outputs, redact_sensitive_configs, sensitive_topic_patterns, max_value_bytes, max_output_bytes, max_destructive_per_minute, ifc_strict, hard_max_records, hard_max_bytes, identity_propagation.

Full defaults: doc/configuration.md.

N. Quality, packaging & agent DX

O. Documented reference gaps (intentional)

HTTP full server · real brokers · HTTP policy URL client · durable audit_topic publish · wall-clock dependency_timeout_ms · some approval tool names reserved but not all registered · production language = Java (this package is the Python reference). See doc/kip-alignment.md.

Quick start

Requires Python 3.8+. Core has no third-party packages.

# From source
python run_tests.py
python demo_end_to_end.py
python examples/01_sre_readonly_triage/run.py
echo {"jsonrpc":"2.0","id":1,"method":"tools/list"} | python serve_stdio.py

See the PyPI section at the top for pip install, or doc/publishing.md to publish a release.

Documentation & examples

Repository layout

flowchart TB
  root["kafka-mcp-enterprise-server"]
  pkg["kafka_mcp/<br/>reference server"]
  tests["tests/<br/>302-check suite"]
  docs["doc/<br/>guides + diagrams"]
  ex["examples/<br/>6 scenarios"]
  entry["serve_stdio.py · run_tests.py · demo_end_to_end.py"]
  root --> pkg
  root --> tests
  root --> docs
  root --> ex
  root --> entry

Engineering standards

This reference aims at production-grade practice even while staying a teaching implementation:

Honest limitations (by design)

• Taint / IFC is best-effort - defeatable by data laundering; do not treat as complete mediation.
• In-memory Kafka - validates control logic; not a broker client.
• stdio-first - HTTP is specified in the KIP; this reference documents notes, full HTTP is a Java/production concern.
• Secure-by-default - shipped tools_allowed is read/consume only (SECURE_DEFAULT_TOOLS); enable mutate/destructive explicitly.

Note: Python reference vs Java production code

This repository is a Python reference implementation for validating the KIP-1318 security control model (stdio + in-memory backend + conformance tests). It is not written in Java and is not the official Apache Kafka MCP server.

Items listed as intentional gaps (Streamable HTTP / OAuth, live Admin-Producer-Consumer, EOS/fencing, Connect tools, distributed rate/audit state, JMX, and remaining tools/resources) are specified for the real implementation tracked under KAFKA-20436. Those belong in the actual Java production module (tools/mcp-server) and will be added there - not claimed as complete in this Python reference.

Full matrix: doc/kip-alignment.md.

Related links

• KIP-1318 (wiki): https://cwiki.apache.org/confluence/display/KAFKA/KIP-1318%3A+Model+Context+Protocol+%28MCP%29+Server+for+Apache+Kafka
• Jira: https://issues.apache.org/jira/browse/KAFKA-20436
• MCP specification: https://modelcontextprotocol.io/

License & affiliation

Apache Kafka, KIP-1318, and KAFKA-20436 are trademarks / projects of the Apache Software Foundation. This repository is a community reference aligned with that proposal; it is not the official ASF deliverable unless and until merged under the Kafka project.