What can you do with this server?

The Kafka MCP Enterprise Server provides governed, AI-agent-safe access to Apache Kafka through 11 tools and 6 resources, with enterprise-grade security controls. Core Capabilities: * Read cluster metadata: List or describe topics, consumer groups, and cluster details (tools: list_topics, describe_topic, describe_cluster, list_consumer_groups, describe_consumer_group). * Consume and produce messages: Read from topics with configurable limits and DLP scrubbing; write messages to topics with egress DLP scanning (tools: consume_messages, produce_message). * Manage topics: Create, alter configuration, or delete topics (tools: create_topic, alter_topic_config, delete_topic). Destructive operations (delete, ACL creation) require HMAC-signed approval tokens. * Manage ACLs: Create access control list entries (tool: create_acls). * Access resources: Retrieve topic lists, topic details, cluster info, consumer group lists, audit logs, and health status via kafka:// URIs. Security & Governance: * Fail-closed security pipeline: Enforces authentication, deny/allow lists, topic/group scoping, policy engine evaluation, taint guarding, approval gates, rate limiting, circuit breakers, validation, quarantine, identity propagation, sensitive topic gating, egress DLP, dry-run, consume clamps, byte bounds, post-execute DLP scrubbing, taint registration, and full audit with hash-chaining. * Data Loss Prevention (DLP): Detects and blocks/redacts sensitive data (e.g., emails, SSNs, credit cards, keys) on both consume and produce paths, with configurable categories. * Approval and taint tracking: Destructive actions require HMAC-signed approval tokens; tainted data is tracked to prevent information flow violations. * Resilience: Per-module circuit breakers (data_plane, control_plane), request rate limits, and rogue-agent quarantine. * Audit Trail: All decisions and actions are logged in a hash-chained ring buffer with identity and correlation IDs. * Observability: Monitor server liveness and circuit breaker states through health endpoints. The server implements JSON-RPC 2.0 for MCP methods (initialize, tools/list, tools/call, resources/list, resources/read).

Which integrations are available for this server?

Provides tools for managing Apache Kafka clusters, including listing and describing topics, consumer groups, cluster metadata, producing and consuming messages, creating and altering topics, and managing ACLs.

How do I use Kafka MCP Enterprise Server?

1. Click on "Deploy Server". 2. Wait a few minutes for the server to deploy. Once ready, it will show a "Started" state. 3. In the chat, type @ followed by the MCP server name and your instructions, e.g., "@Kafka MCP Enterprise Server List all topics in the cluster" That's it! The server will respond to your query, and you can continue using it as needed. Here is a step-by-step guide with screenshots.

Kafka MCP Enterprise Server

by vaquarkhan

Overview Schema Source Related Servers Discussions Admin

Python

Remote

The Kafka MCP Enterprise Server provides governed, AI-agent-safe access to Apache Kafka through 11 tools and 6 resources, with enterprise-grade security controls.

Core Capabilities:

Read cluster metadata: List or describe topics, consumer groups, and cluster details (tools: list_topics, describe_topic, describe_cluster, list_consumer_groups, describe_consumer_group).
Consume and produce messages: Read from topics with configurable limits and DLP scrubbing; write messages to topics with egress DLP scanning (tools: consume_messages, produce_message).
Manage topics: Create, alter configuration, or delete topics (tools: create_topic, alter_topic_config, delete_topic). Destructive operations (delete, ACL creation) require HMAC-signed approval tokens.
Manage ACLs: Create access control list entries (tool: create_acls).
Access resources: Retrieve topic lists, topic details, cluster info, consumer group lists, audit logs, and health status via kafka:// URIs.

Security & Governance:

Fail-closed security pipeline: Enforces authentication, deny/allow lists, topic/group scoping, policy engine evaluation, taint guarding, approval gates, rate limiting, circuit breakers, validation, quarantine, identity propagation, sensitive topic gating, egress DLP, dry-run, consume clamps, byte bounds, post-execute DLP scrubbing, taint registration, and full audit with hash-chaining.
Data Loss Prevention (DLP): Detects and blocks/redacts sensitive data (e.g., emails, SSNs, credit cards, keys) on both consume and produce paths, with configurable categories.
Approval and taint tracking: Destructive actions require HMAC-signed approval tokens; tainted data is tracked to prevent information flow violations.
Resilience: Per-module circuit breakers (data_plane, control_plane), request rate limits, and rogue-agent quarantine.
Audit Trail: All decisions and actions are logged in a hash-chained ring buffer with identity and correlation IDs.
Observability: Monitor server liveness and circuit breaker states through health endpoints.

The server implements JSON-RPC 2.0 for MCP methods (initialize, tools/list, tools/call, resources/list, resources/read).

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

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 `dev@kafka.apache.org`
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

Kafka MCP Enterprise - 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.

Related MCP server: mcp-confluent

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.

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.

Available Tools

11 tools

alter_topic_configD

mutate / control_plane / ALTER

consume_messagesD

read / data_plane / READ

create_aclsD

destructive / control_plane / ALTER

create_topicD

mutate / control_plane / CREATE

delete_topicC

destructive / control_plane / DELETE

describe_clusterC