Kafka Enterprise Server

PyPI Python License

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

Area	This reference	KIP / Java track
Transport	stdio only	stdio + Streamable HTTP + OAuth
Backend	In-memory Kafka	Real Admin / Producer / Consumer
Tools	13 registered	~30+ (incl. Connect, EOS, group admin)
Resources	subset + Phase-1 lag/offsets	~20 kafka:// URIs
Rate / quarantine	per-process	optional distributed backend
Audit	in-memory hash chain	durable audit topic
Default allow-list	read/consume only (SECURE_DEFAULT_TOOLS)	same secure-by-default posture
Conformance	302/302 checks	KIP test plan

Website

Site	https://vaquarkhan.github.io/kafka-mcp-enterprise-server/
Source	site/ (index.html + local site/assets/site.css)
Deploy	.github/workflows/pages.yml publishes site/ on push to main

PyPI

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

Package	kafka-mcp-enterprise
CLI	kafka-mcp-enterprise
Import	import kafka_mcp
Optional	pip install kafka-mcp-enterprise[otel]
Publish	Tag v* → .github/workflows/publish.yml (Trusted Publishing / OIDC - see doc/publishing.md)

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

AGENTS.md	AGENTS.md - canonical instructions for every coding agent
Skills	.cursor/skills/ (Cursor) · skills/ (portable)
Guide	doc/agents-and-skills.md - how to load in each IDE

---

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.

KIP	KIP-1318: MCP Server for Apache Kafka
Jira	KAFKA-20436 - Implement KIP-1318
Discuss	[DISCUSS] KIP-1318 on [email protected]
This repo	Stdlib Python reference / conformance server (teaching, demos, security validation)
KIP production target	Java module (tools/mcp-server) wrapping native Kafka clients - see the KIP

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

Feature	Detail
JSON-RPC 2.0	Strict jsonrpc: "2.0" request/response
initialize	protocolVersion, serverInfo, capabilities
tools/list	Visible tools honor deny-list, allow-list, and readonly
tools/call	Full fail-closed pipeline + handler
resources/list	Catalog of kafka:// URIs
resources/read	Topic/cluster/group/audit/health reads
stdio transport	Newline-delimited JSON (serve_stdio / kafka-mcp-enterprise)
HTTP notes	Stateless HTTP design documented; full HTTP listener is a KIP/Java concern
Stateless approvals	HMAC tokens self-contained (no sticky session required for authz correctness)
Correlation IDs	Per-call corr_id on audit entries

B. Tools (13) - classified

Tool	kind	module	Kafka op
list_topics	read	control_plane	DESCRIBE
describe_topic	read	control_plane	DESCRIBE
describe_cluster	read	control_plane	DESCRIBE
list_consumer_groups	read	control_plane	DESCRIBE
describe_consumer_group	read	control_plane	DESCRIBE
consume_messages	read	data_plane	READ
create_topic	mutate	control_plane	CREATE
alter_topic_config	mutate	control_plane	ALTER
produce_message	mutate	data_plane	WRITE
alter_consumer_group_offsets	mutate	control_plane	ALTER
delete_topic	destructive	control_plane	DELETE
delete_consumer_group	destructive	control_plane	DELETE
create_acls	destructive	control_plane	ALTER

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

C. Resources (kafka://)

URI	Purpose
kafka://topics	List topics
kafka://topics/{name}	Describe topic
kafka://topics/{name}/offsets	Earliest/latest offsets per partition
kafka://cluster	Cluster id + brokers
kafka://groups	Consumer groups
kafka://groups/{id}	Describe a consumer group
kafka://groups/{id}/offsets	Committed offsets
kafka://groups/{id}/lag	Per-topic/partition lag (KIP Phase 1)
kafka://audit/recent	Recent audit entries
kafka://health	Liveness + per-module circuit breaker state

D. Fail-closed security pipeline (exact order)

Every tools/call - first denial wins:

Step	Control	Denial code
1	Bearer auth - audience / issuer validation (off until configured)	-32001 UNAUTHORIZED
2	Deny-list (tools_denied)	-32044 POLICY_DENIED
3	Allow-list (tools_allowed) + readonly (blocks all non-read, including produce)	-32044
4	Topic prefix scope + group prefix scope	-32041 SCOPE_VIOLATION
5	Policy engine - callable; deny or exception → fail-closed	-32044
6	Taint guard / IFC - destructive tools; optional ifc_strict; approval bypasses	-32040 TAINT_VIOLATION
7	Approval gate - HMAC signed TTL token (_approval_token)	-32042 APPROVAL_REQUIRED
8	Rate limit - general vs admin/control-plane buckets	-32029 RATE_LIMITED
9	Execute via per-module circuit breaker + dependency check	-32043 DEPENDENCY_UNAVAILABLE

Pre / around execute (also enforced):

Guardrail	Behavior	Code
Input validation	Identifier charset; max_value_bytes on produce values	-32046 VALIDATION_FAILED
Rogue-agent kill-switch	Per-identity destructive burst → quarantine	-32047 QUARANTINED
Identity propagation	Optional per-principal broker ACL check before execute	-32044
Sensitive-topic gating	Pattern match on consume → requires approval	-32042
Egress DLP	Block secret categories on produce	-32045 SENSITIVE_DATA_BLOCKED
Dry-run tools	dryrun_tools returns plan without mutation	-
Consume clamp	maxMessages capped by hard_max_records	-
Byte bounds	hard_max_bytes trims consume payload; max_output_bytes truncates scrubbed output	truncation tags
Post-execute DLP scrub	Redact/scrub whole result tree	-
Taint registration	Consumed values registered into session taint set	-
Audit	ALLOW/DENY recorded (params truncated, hash-chained)	-

E. Data-protection guardrails (DLP)

Capability	Detail
Modes	redact | block | off (dlp_mode)
Default block categories	private_key, aws_access_key, jwt
Detectors (10)	email, ssn, credit_card (Luhn-validated), phone, ipv4, aws_access_key, private_key, jwt, iban, secret_assignment
Consume path	Redact PII in records; block-mode can refuse
Produce path	Egress scan → -32045
Sensitive configs	Mask password/secret-like keys on describe (redact_sensitive_configs)
Scrub all outputs	Walk entire JSON result (scrub_all_outputs)
Legacy interceptor	interceptor.redact_record kept for compatibility; DLP is primary

F. Approval, taint & IFC

Capability	Detail
HMAC approval tokens	mint / verify; TTL (default 300s); tool-bound
Forged / expired tokens	Rejected → -32042
Default approval-required tools	delete_topic, delete_records, create_acls, delete_acls, alter_partition_reassignments, alter_broker_config
Taint guard	Best-effort substring match of session tainted values into destructive args
ifc_strict	After untrusted read, blocks destructive/control-plane without approval
Honesty	Taint is defeatable by laundering; least-privilege broker ACLs are load-bearing

G. Scoping, exposure & identity

Capability	Detail
Tool allow-list / deny-list	tools_allowed, tools_denied
Secure-by-default guidance	Shipped default is read/consume only; expand tools_allowed for mutate/destructive
Readonly mode	Disables create/produce/alter/delete/ACLs
Topic prefixes	allowed_topic_prefixes
Group prefixes	allowed_group_prefixes
Identity propagation	identity_propagation + in-memory per-principal ACLs (set_principal_acl / authorize)
Session identity	session["identity"] for audit, quarantine, ACL principal

H. Resilience & blast-radius controls

Capability	Detail
Circuit breakers	Per module: data_plane, control_plane, ecosystem
Breaker isolation	Control-plane open does not take down data-plane consume/produce
Dependency failure hook	_inject_dependency_failure / _fail_module → -32043
Rate limits	rate_requests_per_second + rate_admin_requests_per_second
Quarantine	max_destructive_per_minute per identity
Health resource	Breaker states on kafka://health

I. Consume semantics (Direct Partition Assignment)

Mode	Behavior
No groupId	assignment=direct, no consumer group, no rebalance
With groupId	Classic group path; register offsets; rebalance counter increments

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

Capability	Detail
Ring buffer	Recent entries (maxlen=1000)
Hash chaining	Tamper-resistant best-effort chain
Param truncation	Long params truncated (>64 chars)
Decisions	ALLOW / DENY with identity, tool, corr_id
Resource	kafka://audit/recent
audit_topic	Config name present; durable Kafka mirror is a documented reference gap

L. Error codes (complete - 15)

Code	Constant	Meaning
-32700	PARSE_ERROR	JSON parse error
-32600	INVALID_REQUEST	Invalid request
-32601	METHOD_NOT_FOUND	Unknown method/tool
-32602	INVALID_PARAMS	Invalid params / structured Kafka errors
-32603	INTERNAL_ERROR	Internal error
-32001	UNAUTHORIZED	Bad/missing bearer
-32029	RATE_LIMITED	Rate limited
-32040	TAINT_VIOLATION	Tainted value into destructive tool
-32041	SCOPE_VIOLATION	Topic/group out of scope
-32042	APPROVAL_REQUIRED	Destructive/sensitive needs approval
-32043	DEPENDENCY_UNAVAILABLE	Circuit breaker open / dependency down
-32044	POLICY_DENIED	Deny/allow/readonly/policy/ACL propagation
-32045	SENSITIVE_DATA_BLOCKED	Egress / DLP block
-32046	VALIDATION_FAILED	Malformed identifier / oversized value
-32047	QUARANTINED	Rogue-agent kill-switch

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

Feature	Detail
Conformance suite	302/302 checks (incl. KIP Test Plan matrix + coverage gap-fill)
Line coverage	100% of kafka_mcp/ via python run_coverage.py (optional coverage package)
Smoke + demo	test_kafka_mcp.py (16), demo_end_to_end.py (22 steps, all security codes)
Examples	Six folders with real-world data/ fixtures
PyPI	kafka-mcp-enterprise · CLI kafka-mcp-enterprise
Stdlib-only core	No hard third-party deps
Optional OTel	pip install …[otel] - not required (doc/observability.md)
AGENTS.md + skills	Cursor / Kiro / Copilot / ChatGPT / Gemini (doc/agents-and-skills.md)

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

Resource	Description
doc/	End-to-end guides: getting started, architecture, security, config, tools, errors, testing
doc/kip-alignment.md	Feature matrix vs KIP-1318 - what is implemented vs intentional reference gaps
doc/publishing.md	PyPI package kafka-mcp-enterprise
doc/observability.md	OpenTelemetry: optional, not required
doc/agents-and-skills.md	AGENTS.md + skills for all IDEs
examples/	Six folder-based scenarios (run.py + real-world data/ fixtures)

---

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:

Practice	How it shows up
Fail-closed	First denial wins; no execute-then-check paths
Least privilege	Prefix scopes, allow/deny lists, readonly, approval for destructive ops
Defense in depth	MCP controls + explicit honesty that broker ACLs are load-bearing
Bounded blast radius	Hard record/byte caps, rate limits, per-plane breakers, quarantine
Observable denials	Stable error codes, correlation IDs, audit ALLOW/DENY
Testability	Deterministic in-memory backend; security + integration coverage
Zero dependency debt	Python stdlib only - easy to audit and run in CI
Clear product boundary	Official Kafka delivery tracked on KAFKA-20436 (Java)

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.