KIP-1318: Model Context Protocol (MCP) Server for Apache Kafka

Status

Current state: Under Discussion

Discussion thread: https://lists.apache.org/list?dev@kafka.apache.org:2026-5:1318

JIRA: KAFKA-20436 - Getting issue details... STATUS

Motivation

Background

The Model Context Protocol (MCP) is an open standard originally announced by Anthropic in November 2024 and donated to the Linux Foundation's Agentic AI Foundation in December 2025, co-founded by Anthropic, Block, and OpenAI. MCP follows a client-host-server architecture where the host application (such as Claude Desktop or VS Code) creates one MCP client per MCP server connection, with all communication using JSON-RPC 2.0 messages through stateful sessions with capability negotiation at initialization.

The protocol defines three server-side primitives:

• Tools - functions for the AI model to execute (state-changing actions)
• Resources - context and data for the model (read-only)
• Prompts - templated messages and workflows

MCP was designed to solve the "N×M integration problem" where developers previously had to build custom connectors for each combination of data source and AI tool. MCP adoption is accelerating across Claude Desktop, Claude Code, GitHub Copilot (VS Code), and Google's Agent Development Kit (ADK).

What it shows: the general MCP client-host-server model. A host application (Claude Desktop, VS Code, Google ADK) creates one MCP client per server connection. Each client talks to its MCP server over JSON-RPC 2.0 in a stateful session, negotiating capabilities at initialization. Every server exposes up to three primitives - Tools (state-changing actions), Resources (read-only context), and Prompts (templated workflows) - and wraps an external system. Why it matters: this KIP's Kafka MCP server is one such server; the host can connect to several servers at once, which is exactly why per-server isolation and least privilege matter.

Motivating Question

Should Apache Kafka provide a first-party MCP server so that AI agents can interact with Kafka clusters through the standardized MCP protocol?

Yes. Apache Kafka has a rich operational surface spanning five core APIs (Producer, Consumer, Streams, Connect, Admin) with over 100 distinct operations. Today, interacting with Kafka programmatically requires either:

1. Writing Java/Python client code against the Kafka client libraries
2. Using CLI tools (kafka-topics.sh, kafka-consumer-groups.sh, kafka-acls.sh)
3. Calling the Connect REST API directly via curl

None of these are accessible to AI agents through MCP. This means developers cannot ask their AI assistant to "create a topic with 12 partitions and 3-day retention," "show me the lag for consumer group X," or "restart the failed connector task" - operations that should be trivial in an AI-assisted workflow.

Existing Implementations and Gap Analysis

At least five open-source Kafka MCP server implementations already exist (several Python implementations), validating the demand. However, all share critical gaps:

• No ACL management - security administration is excluded from AI-assisted workflows
• No transactional produce semantics - no exactly-once guarantees
• No Kafka Streams or Share Group operations - major APIs are entirely absent
• Confluent coupling - the most feature-complete implementation (mcp-confluent) only works with Confluent Cloud REST APIs, not vanilla Apache Kafka

Additional security gaps. Beyond the functional gaps above, existing implementations also lack enterprise security controls: no separation between data-plane reads and control-plane mutations (indirect prompt injection / data-to-tool escalation), no resource-level topic scoping beyond a single broker credential, no bounded output/circuit breakers for autonomous agents, no tamper-evident audit trail, and no protection against MCP-specific attacks (tool poisoning, rug-pull, confused deputy). These are addressed in the  Security Hardening section below.

This KIP Fills All Gaps

New Capabilities

A first-party MCP server for Apache Kafka will allow:

• AI agents to create, describe, alter, and delete topics through natural language
• AI agents to produce and consume messages, including transactional exactly-once produce
• AI agents to monitor consumer group lag, reset offsets, and manage group membership
• AI agents to manage ACLs - the first MCP implementation to expose security administration
• AI agents to manage Kafka Connect connectors (create, pause, resume, restart, delete)
• AI agents to inspect cluster health, KRaft quorum status, and broker configurations
• AI agents to manage transactions (list, describe, abort hanging transactions, fence producers)
• Developers to run the same tool locally (stdio) or deploy it remotely (Streamable HTTP) with OAuth 2.1

By incorporating this feature within Apache Kafka specifically:

• More operators will have access to this feature under the Apache 2.0 license
• The community can maintain the feature, reducing dependence on vendor-specific implementations
• The server can be released alongside Kafka itself, ensuring API compatibility with each Kafka version
• Vanilla Apache Kafka deployments (not just Confluent Cloud) are supported

This KIP proposes a first-party, Apache-licensed MCP server that wraps Kafka's native Java client APIs directly, supporting both vanilla Apache Kafka and managed deployments, with zero new external dependencies.

Why a First-Party MCP Server Is Necessary

Two forces make this urgent: agentic AI needs Kafka as its operating memory, and the only alternative today - stitching in unvetted third-party MCP servers - is a security liability.

Kafka is the streaming backbone agents need. Agents do not just need "the database" (which holds state) - they need the event history: how state changed, who reacted, what was attempted, and what happened next. Operational streams (change events, telemetry, user/workflow events, tool-result events, policy/approval events) are the agent's memory substrate, coordination layer, and control record. MCP exposes callable tools and resources, Kafka carries the durable events, and stream processors (Flink, Kafka Streams) turn raw streams into stateful signals - distinct layers, not competitors.

Without MCP, Kafka is not reachable by agents. Today a Kafka cluster is accessible only through client code, CLI tools, or the Connect REST API - none of which an AI agent speaks. MCP is the standard interface agents already use across Claude, Copilot, and Google ADK. A first-party MCP server is the prerequisite that makes Apache Kafka agentic-AI-ready.

Relying on third-party MCP servers is a real security risk. The broader MCP ecosystem's security posture is weak: an audit of 17 popular MCP servers found an average score of about 34/100 with none declaring permissions; researchers reported exploitable issues across roughly 7,000 public MCP servers spanning 150M+ downloads; the first malicious MCP package appeared in September 2025 and exfiltrated data undetected for about two weeks; and multiple CVEs were disclosed in a widely used official Git MCP server. Tool poisoning, rug-pulls, confused-deputy, token-passthrough, and indirect prompt injection are demonstrated, not theoretical. Wiring a Kafka cluster's control and data planes to an unvetted external MCP server means trusting that third party with the crown jewels of the data platform.

Therefore Kafka must ship its own MCP server. Only a first-party, Apache-licensed server can wrap the native clients with zero new dependencies, stay API-compatible via the Kafka release cycle, bake in the security hardening the ecosystem lacks (see Security Hardening), and run on vanilla Apache Kafka - not just one vendor's cloud. Sources: Kafka for Agentic AI (AutoMQ), Apache Kafka docs; third-party statistics are ecosystem findings as of their publication dates.

Public Interfaces

This KIP introduces no changes to the Kafka protocol, public APIs, client behaviors, or broker metrics.

It adds a new standalone module (tools/mcp-server) that packages a JSON-RPC 2.0 server exposing Kafka operations as MCP Tools (state-changing actions) and MCP Resources (read-only data). The server is a separate process - it does not run inside the broker.

New Configuration Properties

The MCP server is configured via command-line arguments and/or a properties file:

All standard Kafka client properties (security.protocol, sasl.*, ssl.*) are passed through directly to the underlying Admin, KafkaProducer, and KafkaConsumer instances.

 Security configuration properties. The following properties add the enterprise hardening controls described in the Security Hardening section.

New MCP Tools (State-Changing Operations)

Topic Management:

Message Operations:

Consumer Group Management:

ACL Management:

Cluster Operations:

Transaction Management:

Connect Operations (requires mcp.connect.url):

Tool classification. Every tool is classified as read, mutate, or destructive-mutate. Read: consume_messages and all resources. Mutate includes writes such as produce_message. Destructive-mutate (e.g., delete_topic, create_acls, alter_broker_config) defaults into mcp.approval.required.tools. mcp.readonly=true registers read tools only and disables all writes, produce included.

The taint guard gates read-value → destructive-mutate chaining specifically; ordinary read tools and single-purpose calls are unaffected.

New MCP Resources (Read-Only Data)

Resource URIs use the kafka:// scheme as internal MCP identifiers. This is a custom scheme used only within the MCP protocol for resource discovery and is not registered with IANA. It is never exposed on the network - MCP clients resolve resources by calling the resources/read JSON-RPC method with the URI as a parameter. This follows the same pattern used by other MCP servers (e.g., github://, postgres://).

Note -Resource URIs use the kafka:// scheme as internal MCP identifiers, resolved via the resources/read JSON-RPC method (never exposed on the network).

 Audit resource. kafka://audit/recent exposes the most recent audit records (read-only, subject to the same authorization) so an operator or agent can review recent actions and correlation IDs.

Proposed Changes

Architecture

The MCP server is a standalone Java process that wraps three existing Kafka client interfaces. It does not run inside the broker.

What it shows: the three tiers and where the MCP server sits. The AI host runs an MCP client that speaks JSON-RPC 2.0 to a standalone Kafka MCP server (a separate process, never inside the broker). Every request crosses the transport, then the security pipeline, before reaching one of three decoupled modules - data-plane (produce/consume via KafkaProducer/KafkaConsumer), control-plane (topics/ACLs/cluster via Admin), and ecosystem (Kafka Connect via REST). 

Why it matters: the agent only ever touches Kafka through the guarded server, and a failure in one module (for example a hung Connect endpoint) cannot stall the others.
 

Architecture & Flow Diagrams 

What it shows: the fixed order every tools/call passes through. The first stage that denies stops the request and returns a specific JSON-RPC error code (shown on each "no" branch). The order is deterministic: authentication, deny-list, allow-list/read-only, resource scope, external policy engine, taint guard, approval-token check, rate limit, then execution wrapped in a per-dependency circuit breaker.

 Why it matters: authorization is predictable and auditable - every rejection maps to a known code, and no Kafka call happens until all gates pass. "Fail-closed" means any error or timeout in a gate denies rather than allows.

What it shows: indirect prompt injection and how it is stopped. An agent consumes a topic that contains a hidden instruction planted by an attacker. The server redacts PII from the records and marks the returned values Untrusted. When the (now-hijacked) agent tries to call delete_topic using a value that came from that message, the taint guard detects that the argument derives from Untrusted data and denies it (-32040) unless a signed, out-of-band human approval token is presented. 

Why it matters: data an agent reads can never silently become a destructive control-plane action - the exact failure mode that makes naive Kafka MCP servers dangerous. (The taint guard is best-effort; the approval gate, read-only mode, scoping, and least-privilege ACLs are the backstops.)

What it shows: the two ways consume_messages reads. With no groupId (the common ephemeral agent case) the server uses assign() - a transient, unnamed consumer that never joins the group coordinator, so connecting and disconnecting causes no rebalance and leaves no orphaned consumer group. With an explicit groupId it uses the group path with static membership (group.instance.id) and a longer session.timeout.ms to absorb LLM reasoning latency.

Why it matters: dozens of short-lived agent consumers using normal subscribe() would trigger continuous rebalance storms that degrade the whole cluster; Direct Partition Assignment eliminates that.

MCP Primitives Mapping

MCP defines three server-side primitives. This KIP uses two:

Tool Implementations - Kafka API Mapping

Each MCP Tool maps to a specific method on Kafka's Java client APIs. Below is the exact mapping with source file references verified against the Kafka codebase.

Representative mappings:

• create_topic → Admin.createTopics(Collection<NewTopic>); omitted partitions/replicationFactor fall back to broker defaults via the Optional constructor.
• delete_topic → Admin.deleteTopics(...); alter_topic_config → Admin.incrementalAlterConfigs(...) with SET; create_partitions → Admin.createPartitions(...); delete_records → Admin.deleteRecords(...).
• produce_message → KafkaProducer.send(ProducerRecord); produce_transactional → initTransactions()/beginTransaction()/send()/commitTransaction() with abortTransaction() on failure (exactly-once, absent from all existing implementations).

consume_messages → short-lived KafkaConsumer per request; enable.auto.commit=false; max.poll.records bounded by maxMessages (default 10, hard-capped by mcp.hard.max.records).

• Direct Partition Assignment. When no groupId is supplied (the common ephemeral case), the consumer uses assign() rather than subscribe(): it is a transient, unnamed consumer that does not join the group coordinator, so its connect/terminate causes no rebalance and leaves no orphaned group. When a groupId is explicitly supplied, the group path is used (with group.instance.id static membership and extended session.timeout.ms recommended to absorb LLM latency). This eliminates ephemeral-consumer rebalance storms.

• Consumer group, ACL, cluster, and transaction tools map to the corresponding Admin methods (deleteConsumerGroups, alterConsumerGroupOffsets, createAcls, deleteAcls, electLeaders, alterPartitionReassignments, abortTransaction, fenceProducers, etc.).
• Connect tools call the Kafka Connect REST API (they do not use the Java client).

Consumer lag (kafka://groups/{id}/lag) is computed by combining listConsumerGroupOffsets (committed) with listOffsets (end offsets): lag = endOffset - committedOffset per partition.

Transport Mechanisms

stdio - the AI client launches the server as a subprocess; JSON-RPC 2.0 over stdin/stdout. Default for local development.

Streamable HTTP - the server runs as an independent HTTP service; JSON-RPC over HTTP POST with optional SSE streaming. Suited for remote and production deployments.

Stateless HTTP deployment. In HTTP mode the server is stateless: each request is self-contained and independently authenticated, and the server holds no cross-request session state. MCP capability negotiation still occurs per connection, but correctness never depends on server-held session memory. This removes the session-hijacking and cross-request state-poisoning surface and allows horizontal scaling behind any load balancer (Kubernetes HPA, Fargate, Cloud Run, Container Apps) without sticky sessions. When multiple replicas run, set mcp.ratelimit.backend=distributed so rate limits hold cluster-wide. stdio remains a local single-session transport.

Security Architecture

Layer 1: Kafka Broker Authentication. The server connects to Kafka as a regular client. All standard Kafka authentication mechanisms are supported via standard client configuration properties (security.protocol, sasl.mechanism, sasl.jaas.config, ssl.*), passed directly to Admin.create(props), new KafkaProducer(props), and new KafkaConsumer(props).

Layer 2: MCP Protocol Authentication (Streamable HTTP only). For remote deployments, the server implements OAuth 2.1 with PKCE per the MCP specification. The MCP server acts as an OAuth resource server, validating bearer tokens on each request. stdio transport needs no MCP-layer authentication because the client launches the server as a local subprocess.

Authorization. The MCP server inherits Kafka's ACL-based authorization. If the configured Kafka user lacks permission, the operation fails with an authorization exception surfaced as a structured JSON-RPC error. Operators should configure least-privilege ACLs.

Integration with KIP-1298 (Per-Resource-Type Authorization). When available, the MCP server's Kafka credentials can be restricted to specific resource types, providing defense-in-depth beyond tool-level restrictions.

Tool Allow-List. The server supports mcp.tools.allowed and mcp.tools.denied. Denied tools are never registered in capability negotiation, so the agent cannot discover or invoke them regardless of prompt content.

 Security Hardening

This section adds enterprise controls mapped to the OWASP Top 10 for LLM Applications (2025), the OWASP Top 10 for Agentic Applications (2026, ASI01-ASI10), and MCP-protocol-specific attacks. Controls are opt-in, defense-in-depth mechanisms that complement, never replace, broker-side ACLs; the server implements no authorization model of its own and can only do what its Kafka identity permits. The load-bearing backstops remain broker-side ACLs and a least-privilege Kafka identity. Guidance-only patterns are called out as such, and the Limitations subsection states what the server cannot address.

Secure-by-default posture (devil's-advocate fix)

The server ships locked down. In the default profile, only read/describe and non-destructive tools are registered; destructive and cluster-admin tools (delete_topic, delete_records, create_acls, delete_acls, alter_broker_config, alter_partition_reassignments, elect_leaders, fence_producers, abort_transaction) are NOT exposed unless an operator explicitly adds them to mcp.tools.allowed. Even when enabled they require an approval token by default. Pure-observability agents should run mcp.readonly=true. This inverts the usual "everything on" default so granting an agent dangerous capability is always a deliberate, auditable act.

Control evaluation order

Every request passes through a fixed pipeline; the first stage that denies stops the request (fail-closed): (1) authentication (OAuth 2.1 bearer on HTTP, audience/issuer validated); (2) tool deny-list; (3) tool allow-list / mcp.readonly; (4) resource scope (mcp.allowed.topic.prefixes, mcp.allowed.group.prefixes); (5) external policy engine, if configured, fail-closed; (6) taint guard; (7) approval-token check; (8) rate limit; (9) execute, through the dependency circuit breaker.

Context-tool isolation (taint guard) - best-effort

The most severe risk for a data-plane + control-plane server is indirect prompt injection via data-to-tool escalation (OWASP LLM01, ASI01/ASI02): a malicious record read from a topic steers the agent into a mutating tool. When mcp.taint.guard.enabled=true (default), values returned by reads are matched (normalized exact/substring) against arguments to destructive-mutate tools; a match requires a valid approval token, else JSON-RPC error -32040 (taint violation).

This is best-effort, not a complete defense: an LLM can launder a value (paraphrase, re-encode, partially quote) to evade string matching. It raises the cost of the naive attack and creates an audit signal; the real backstops are the approval gate, mcp.readonly, resource scoping, and least-privilege ACLs. To stay consistent with the stateless HTTP design, taint is evaluated within a single request, and across a multi-step plan only via a client-supplied signed provenance token (the server holds no cross-request taint state). It makes the client-side Dual-LLM and Plan-Then-Execute patterns enforceable (see Security Guidance) but does not replace them.

Resource scoping (topic and group prefixes)

mcp.allowed.topic.prefixes and mcp.allowed.group.prefixes constrain topic- and group-scoped tools/resources to configured prefixes (e.g., agent., sandbox.); out-of-scope targets are rejected before any Kafka call with error -32041 (scope violation). This bounds topic and group operations only. Cluster/ACL/broker mutations (create_acls, alter_broker_config, alter_partition_reassignments, elect_leaders, alter_client_quotas) are not prefix-scopable and are governed by mcp.readonly, the tool allow/deny list, and, authoritatively, broker-side ACLs on the server's Kafka identity. Scoping is complementary to, not a replacement for, broker ACLs.

Approval gate and dry-run (stateless)

Tools in mcp.approval.required.tools require an approval token before execution; without it the server returns -32042 (approval required). The token is a short-lived, signed grant minted out-of-band by the host or an external approval service after a human confirms, then presented by the client on the follow-up call. The MCP server verifies the token signature and stores nothing (stateless); it renders no UI and the human interaction never happens on the chat surface (ASI09). This is defense-in-depth beneath the host's own tool-consent prompt. Tools in mcp.dryrun.tools return the intended change and blast radius without executing (ASI10).

Circuit breakers and fail-fast timeouts

When mcp.circuit.breaker.enabled=true, each external dependency (Kafka Admin, Connect, Schema Registry) has its own circuit breaker and a mcp.dependency.timeout.ms fail-fast timeout. A hung or failing dependency trips only its own breaker and returns -32043 (dependency unavailable); other tool categories continue serving. This prevents one slow ecosystem endpoint from stalling the whole server (ASI08). Tools are grouped into decoupled modules - data-plane (produce/consume), control-plane (admin/ACL/cluster), and ecosystem (Connect/Burrow/Cruise Control) - each with an independent breaker, so degradation in one module never stalls the others. Per-module health is exposed read-only at kafka://health.

Bounded output and rate limiting

Consume operations are bounded by maxMessages (default 10) with per-message and cumulative byte/size caps and truncation, preventing context-window flooding (OWASP LLM10). Server-side per-replica rate limits (requests/sec, produce/consume bytes/sec, Admin requests/sec) plus broker-side client quotas form a dual ceiling; the broker quotas are already cluster-wide and need no new infrastructure. Cross-replica shared rate limiting is optional (mcp.ratelimit.backend=distributed) and requires an external store (see Further Work); it is not needed for correctness because broker quotas bound aggregate load.

Tamper-resistant audit trail

Every request is recorded with caller identity, tool, parameters, policy decision, result, and a correlation id. When mcp.audit.topic is set, records are appended to a Kafka topic that MUST be configured append-only for the MCP identity (deny Delete/Alter on that topic), with a local durable fallback if the topic is unavailable. This is tamper-resistant, not tamper-proof; an external WORM store/SIEM is the authoritative record. Recent entries are exposed read-only via kafka://audit/recent (ASI10, compliance).

External policy engine hook

When mcp.policy.engine.url is set, the server evaluates each tool call against an external policy engine (OPA/Rego or Cedar) before execution, bounded by mcp.dependency.timeout.ms. On deny, timeout, or error it fails closed with -32044 (policy denied). This keeps authorization declarative and auditable.

MCP-protocol attack defenses

• Tool poisoning / rug-pull: the first-party tool set is compiled and static by construction (no dynamic tool descriptions from untrusted input), which is the primary defense. The optional mcp.tools.manifest.signature additionally covers operator-added interceptors/config, verified against the release-signing key.
• Confused deputy / token passthrough: on HTTP transport, bearer tokens are validated against mcp.oauth.expected.audience/mcp.oauth.expected.issuer (OAuth 2.1 / RFC 8707) and never blind-forwarded. Note the server acts on Kafka with a single shared identity; that identity therefore must be least-privilege. Per-caller-to-Kafka-credential mapping (true downstream scoping) is Further Work.
• Session hijacking: removed by the stateless HTTP design; TLS is required for HTTP transport.
• Capability over-claim: broker ACLs are the source of truth; the MCP layer cannot grant more than its Kafka identity holds.
• Bidirectional sampling injection: MCP sampling is disabled by default; if enabled, origin is authenticated and content filtering applies.

Limitations and threat-model boundaries

• The server cannot fully prevent prompt injection; the taint guard is best-effort and defeatable by data laundering. Treat it as one layer.
• Model/agent-layer risks (misinformation, memory poisoning, goal drift, rogue-agent behavior) originate in the client/host and cannot be solved by a Kafka MCP server. The server contributes audit, bounded blast radius, and hard denies only.
• With a single shared Kafka identity, the server is only as constrained as that identity's ACLs. Least-privilege configuration is mandatory, not optional.
• These controls are defense-in-depth. None is a guarantee; they reduce blast radius and add detectability.

OWASP coverage summary (F = full, P = partial/defense-in-depth)

OWASP item	Coverage	Control in this KIP
LLM01 Prompt Injection / ASI01 Goal Hijack	P	Best-effort taint guard, approval gate, tool allow/deny, least privilege
LLM02 Sensitive Info Disclosure	F	McpRecordInterceptor redaction + topic scoping
LLM06 Excessive Agency / ASI03 Identity Abuse	F	Read-only mode, allow/deny, least-privilege identity, KIP-1298
LLM10 Unbounded Consumption	F	Bounded consume + per-replica limits + broker quotas
ASI02 Tool Misuse	P	Taint guard + approval gate on read-to-destructive chaining
ASI08 Cascading Failures	F	Per-dependency circuit breakers + timeouts
ASI09 Human-Agent Trust	P	Out-of-band approval token, not via chat surface (host-dependent)
ASI10 Rogue Agents	P	Audit trail, dry-run, read-only, deny-list kill
MCP tool poisoning / rug-pull	F	Static compiled tool set (+ optional signed manifest)
MCP confused deputy	P	Audience/issuer validation; least-privilege shared identity (per-caller mapping = Further Work)

Rate Limiting and Noisy Neighbor Controls

Level 1: MCP server-side rate limiting enforces configurable request rate limits before any Kafka API call:

Requests exceeding the limit receive JSON-RPC error -32029 (rate limited) and a Retry-After header (HTTP transport).

Level 2: Kafka client quotas - the MCP server's Kafka user should be configured with broker-side quotas (consumer_byte_rate, producer_byte_rate, request_percentage) as a hard ceiling.

Level 1 limits are per-replica by default. Across multiple stateless HTTP replicas, the broker-side client quotas (Level 2) are the cluster-wide ceiling and require no extra infrastructure. An optional mcp.ratelimit.backend=distributed can enforce Level 1 cluster-wide via an external shared store, but this adds a dependency and is not required for correctness (see Further Work).

Data Governance and PII Masking

The server provides an extensible McpRecordInterceptor interface for payload redaction. Implementations can regex-mask credit card numbers, SSNs, and emails; strip JSON fields; exclude entire topics; or integrate with external classification services. Interceptors run inside the server process before the payload is serialized into the JSON-RPC response, so the LLM never sees unredacted data. Configured via mcp.interceptor.classes (comma-separated, applied in order).

Data-protection guardrails. The interceptor is one layer of a broader, defense-in-depth data-protection set (validated end-to-end in the reference implementation):

• DLP everywhere, not just consume. A built-in DLP pass runs over consumed records AND over every tool/resource response, so no response leaks secrets or PII. Detectors: email, SSN, credit card (Luhn-validated), phone, IPv4, IBAN, AWS access keys, PEM private keys, JWTs, and password=/secret=/token= assignments. Modes: redact (mask), block (drop/deny), off.
• Sensitive config masking. Response values under sensitive keys (*.password, sasl.jaas.config, *.token, ssl.key*, credential*) are masked before serialization, closing the describe_configs/describe_topic secret-leak path (OWASP LLM02/LLM07).
• Egress / exfiltration control. produce_message values are scanned; producing a secret/PII outward is blocked with -32045, preventing an agent from exfiltrating data it read from a sensitive topic by writing it elsewhere.
• Sensitive-topic gating. Topics matching mcp.sensitive.topic.patterns require an approval token to consume and are always processed in block-mode DLP.
• Input validation and size caps. Topic/group identifiers are charset/length validated (rejecting injection-y names, -32046); produce values are capped by mcp.max.value.bytes; response size is capped by mcp.max.output.bytes (context-flooding defense, OWASP LLM10).
• Rogue-agent kill-switch. An identity exceeding mcp.max.destructive.per.minute is quarantined (-32047), bounding damage from a hijacked or misaligned agent (ASI10).

New  properties: 

mcp.dlp.mode, mcp.dlp.block.categories, mcp.scrub.all.outputs, mcp.redact.sensitive.configs, mcp.sensitive.topic.patterns, mcp.max.value.bytes, mcp.max.output.bytes, mcp.max.destructive.per.minute.

These are pattern-based and defense-in-depth: they catch well-formed secrets/PII and bound blast radius, but for regulated data they should be paired with schema/classification services and data contracts. Least-privilege broker ACLs remain the authoritative control.

Module Structure

tools/
└── mcp-server/
    ├── build.gradle
    └── src/
        ├── main/java/org/apache/kafka/tools/mcp/
        │   ├── KafkaMcpServer.java           # Entry point, transport selection
        │   ├── McpRequestHandler.java         # JSON-RPC dispatch
        │   ├── tools/                          # TopicTools, MessageTools, GroupTools,
        │   │                                   #   AclTools, ClusterTools, TransactionTools, ConnectTools
        │   ├── resources/                      # Topic/Group/Cluster/Acl/Transaction/Connect resources
        │   ├── security/                       # TaintGuard, ApprovalGate, TopicScope,
        │   │                                   #  PolicyEngineClient, ToolManifestVerifier, AuditSink
        │   ├── resilience/                      # CircuitBreaker, DependencyTimeout, RateLimiter (local/distributed)
        │   └── transport/                      # StdioTransport, HttpTransport (stateless)
        └── test/java/org/apache/kafka/tools/mcp/

The security/ and resilience/ packages hold the hardening components. They wrap the existing tool/resource dispatch so controls apply uniformly to every tool.

Dependencies

Zero new external dependencies. The optional policy-engine hook and distributed rate-limit backend are called over HTTP/standard interfaces and require no new bundled dependency.

Observability

JMX metrics under kafka.mcp: request-total, request-error-total, request-latency-avg, request-latency-p99, active-consumers, and per-tool {toolName}-total / {toolName}-error-total.

Security metrics. Add: taint-violation-total, topic-scope-violation-total, approval-required-total, policy-denied-total, circuit-breaker-open (per dependency), and audit-write-error-total. Emit OpenTelemetry (OTLP) alongside JMX so any cloud backend can ingest them.

Graceful Shutdown

On SIGTERM/SIGINT: stop accepting new requests, wait for in-flight requests (30s), flush and close the singleton KafkaProducer, close Admin/KafkaConsumer (10s), stop the HTTP server, exit. Short-lived consumers close within their request handler (try-with-resources).

Phased Rollout

Hardening in each phase. The taint guard, topic-prefix allowlist, read-only mode, bounded consume, and audit trail ship in Phase 1 (they are core safety, not add-ons). Approval gate, circuit breakers, and signed tool manifests ship in Phase 2 alongside ACLs/Connect. Policy-engine hook and distributed rate limiting ship in Phase 3.

JSON-RPC error codes

Security Guidance (deployment patterns)

These are recommended client/deployment patterns, not server features. The server-side taint guard and approval gate exist to make them enforceable.

• Dual-LLM pattern: run a privileged LLM that plans and calls tools, and a separate quarantined LLM that processes untrusted topic payloads without tool access. Return only constrained, validated outputs (e.g., typed booleans/numbers) from the quarantined model back to the orchestrator.
• Plan-Then-Execute: have the agent commit to a fixed tool plan before reading any untrusted data, so tool outputs cannot inject new mutating calls. Pairs naturally with the approval gate as a human checkpoint.
• Least agency: grant only the tools and topic prefixes the task needs; prefer mcp.readonly=true unless mutation is required.

 Recommended Implementation Language

KIP-1318 proposes Java as the official implementation language, wrapping Kafka's native Java client APIs with zero new external dependencies. Java is the only language with the canonical, most feature-complete Kafka client maintained in the Apache Kafka project, and the only one supporting Kafka Streams and Connect. The MCP SDK ecosystem publishes official SDKs organized into maintenance tiers (see the MCP SDK tiers page): Tier 1 currently includes TypeScript, Python, C#, and Go; Tier 2 includes Java (maintained with Spring AI) and Rust. There is a deliberate tension - Java has the strongest Kafka client but a Tier 2 MCP SDK, while Go has a Tier 1 MCP SDK but a Kafka client (franz-go) without Streams/Connect - which this KIP resolves in favor of Kafka feature completeness.

Language	MCP SDK Tier	Kafka client completeness	Best suited for
Java/Kotlin	Tier 2	Canonical: transactions, Admin, Streams, Connect	Enterprise, KIP-1318 compliance
Go	Tier 1	franz-go: transactions, Admin, groups (no Streams/Connect)	Lightweight high-performance deployments
Python	Tier 1	confluent-kafka-python: transactions, Admin	Prototyping, data-science
TypeScript	Tier 1	KafkaJS unmaintained; Confluent JS wraps librdkafka	Broad MCP community, limited Kafka depth

Concurrency: Java 21+ virtual threads let each message be handled on its own virtual thread; the earlier synchronized-block pinning limitation was resolved in Java 24 (JEP 491, March 2025). Go's goroutines provide preemptive M:N scheduling. TypeScript's single-threaded event loop constrains concurrent Kafka processing. 

Streaming Backbone, Retention, and Cost

Agentic AI changes the shape of Kafka traffic: more producers emit context, more consumers subscribe, and audit/tool-outcome topics become long-retention, high-fanout streams by design (agents replay history for evaluation, simulation, audit, and recovery). Retention should be sized as a first-class requirement in the initial cost estimate, not treated as after-the-fact cleanup. The storage backend is a deployment choice: keep latency-critical control paths on broker-local/NVMe storage, and consider tiered/object-storage backends (Kafka tiered storage, Amazon MSK, or object-storage-native Kafka such as AutoMQ) for the high-volume audit, telemetry, and replay streams the MCP server produces. This is backend-agnostic: the MCP server speaks the Kafka protocol and runs unchanged on Apache Kafka, MSK, Confluent, Redpanda, or object-storage-native distributions. (Vendor-neutral note; the choice is workload-specific, not ideological.)

Natural extensions after the core is stable: MCP Prompts (templated workflows), multi-cluster support via cluster aliases, Schema Registry integration (read-only resources), AI-driven operations (lag remediation, DLQ analysis), binary message support via pluggable serializers (Avro/Protobuf/JSON Schema), and Kafka Streams topology visualization.

Additional security-track further work: per-caller-to-Kafka-credential mapping (true downstream scoping, closing the single-shared-identity confused-deputy gap); distributed rate-limit backend (external shared store for cluster-wide Level 1 limits, an optional new dependency); SPIFFE/SVID workload identity for the MCP server itself; pluggable cloud IAM credential providers (AWS MSK IAM, Azure Entra, GCP IAM) via SASL OAUTHBEARER callback handlers; secrets-manager integration (AWS Secrets Manager, Azure Key Vault, GCP Secret Manager, HashiCorp Vault); and a field-level DLP/classification interceptor.

Relationships to Other Open KIPs

Because the MCP server exposes Kafka's operational surface to AI agents, it will naturally benefit from and interact with other features:

• KIP-1150: Diskless Topics (Accepted) - The MCP server's create_topic tool can expose Diskless topic type configuration once KIP-1150's sub-KIPs (KIP-1163, KIP-1164) are implemented, allowing AI agents to provision cost-optimized topics on hyperscaler clouds.
• KIP-1279: Cluster Mirroring - The MCP server can expose mirror link status and management operations once KIP-1279 adds the relevant Admin API methods, enabling AI-assisted DR management.
• KIP-848: The Next Generation of the Consumer Rebalance Protocol - The MCP server's consumer group resources already use Admin.listGroups() which returns all group types including the new consumer group protocol. No changes needed.
• KIP-932: Queues for Kafka (Share Groups) - The MCP server already includes kafka://share-groups/{id} resources wrapping Admin.describeShareGroups() and Admin.listShareGroupOffsets().

Compatibility, Deprecation, and Migration Plan

This KIP adds a new standalone tool module. It does not modify any existing Kafka code, protocol, or public API.

• Backward compatibility: Not applicable - entirely new module.
• Deprecation: None.
• Migration: None - opt-in tool.

The MCP server targets Kafka 4.0+ (KRaft mode). It works with any Kafka cluster that the standard Java clients can connect to. The server uses only the public Admin, KafkaProducer, and KafkaConsumer APIs, so it is forward-compatible with future Kafka versions as long as these APIs remain stable (they are part of Kafka's public interface contract).

Test Plan

Unit tests with mocked Admin/KafkaProducer/KafkaConsumer for each tool and resource (topic creation, produce/transactional ordering, ACL binding construction, lag computation, Connect HTTP calls).

Integration tests using kafka.test.ClusterTestExtensions: start embedded cluster, start MCP server over stdio, send JSON-RPC requests, verify Kafka state changes and structured error handling.

Security tests. Add conformance tests mapped to each control: taint-guard blocks a read value passed to delete_topic (-32040); topic-scope rejects an out-of-prefix target (-32041); approval gate blocks a destructive tool until confirmed (-32042); circuit breaker opens on a stubbed hung Connect endpoint without affecting Admin tools (-32043); policy-engine deny closes a call (-32044); rate limiter returns -32029; interceptor redacts a seeded PII payload; tool-manifest signature mismatch refuses startup; HTTP token with wrong audience/issuer is rejected; policy-engine timeout fails closed. Tests cover the server-enforceable OWASP/ASI items and MCP-specific attacks only; model/agent-layer risks (misinformation, memory poisoning, goal drift) are out of the server's scope and are noted as boundaries rather than tested here

Reference validation. A runnable reference implementation that mirrors this architecture and the full security evaluation pipeline validates the design end-to-end with 72 automated checks across six categories (functional/tool-level, security conformance, data-protection guardrails, the report mechanisms IFC/identity-propagation/bounded-buffers/Direct-Partition-Assignment/decoupled-modules, resources, and stdio integration). Production is Java per this KIP; the reference is a separate teaching/validation artifact.

 Reference Implementation

A complete, runnable reference implementation is available so reviewers can exercise the design and the security model end-to-end before the production Java module lands.

• Repository:https://github.com/vaquarkhan/kafka-mcp-enterprise-server
• Language / scope: Python, standard library only (zero third-party deps). A teaching/validation artifact, not the shipped server. The production implementation proposed by this KIP is Java wrapping the native Admin/KafkaProducer/KafkaConsumer APIs (see Recommended Implementation Language). The reference mirrors the same architecture, tool/resource surface, JSON-RPC methods, security pipeline, and every error code, using an in-memory Kafka backend so it runs without a cluster.
• Validates: core tools, kafka:// resources, the fail-closed pipeline, DLP + egress control, Direct Partition Assignment, decoupled-module circuit breakers, identity propagation, and each error code.
• Automated tests: 72 checks across six categories (functional, security conformance, data-protection guardrails, report mechanisms, resources, stdio integration) - all passing via python run_tests.py.
• Worked examples: six persona-based runnable scenarios (SRE read-only triage, multi-tenant isolation, change-window approvals, PII/secret egress guard, blast-radius resilience, identity + sensitive topics).
• Try it: git clone https://github.com/vaquarkhan/kafka-mcp-enterprise-server && cd kafka-mcp-enterprise-server-kip-1318 && python run_tests.py

The canonical implementation will be a Java module (tools/mcp-server) contributed as a PR against apache/kafka and linked from KAFKA-20436. The Python reference is for evaluation during discussion and is not part of the Kafka release.

pip install kafka-mcp-enterprise-kip1318
kafka-mcp-enterprise          # stdio JSON-RPC server
pip install "kafka-mcp-enterprise-kip1318[otel]"

Documentation Plan

New MCP Server documentation page (config, tools, resources, transports, security setup, examples with Claude Desktop/VS Code/Google ADK); updates to CONTRIBUTING.md and the Kafka Tools list; inline Javadoc.

Add a "Security Hardening and Deployment" page covering the taint guard, topic-prefix isolation, approval/dry-run, audit trail, policy-engine hook, multi-cloud/on-prem deployment (MSK/Confluent/Redpanda/Strimzi), secrets sourcing, and the OWASP coverage matrix.

Rejected Alternatives

• Embedding MCP in the broker - would couple AI tool concerns with broker stability and increase attack surface; existing CLI tools are standalone client processes for the same reason.
• Using Confluent's mcp-confluent - tied to Confluent Cloud REST APIs; not vanilla Kafka; not Apache-licensed.
• Python/Go implementation - Kafka's client libraries are Java-native; a Java server avoids version skew and integrates with the Gradle build.
• Extending the broker with a REST API - brokers have no REST API; MCP is the protocol agents already speak.
• Adding MCP Prompts in Phase 1 - deferred to reduce scope.
• Multi-cluster in Phase 1 - deferred to a follow-up KIP.

Rejected security alternatives.

• Rely solely on broker ACLs for agent scoping - rejected: ACLs are correct but not an agent-facing, easily-configured scratchpad boundary, and they do not stop data-to-tool escalation within the granted scope. The taint guard + topic-prefix allowlist are complementary, not replacements.
• Blind OAuth token passthrough to Kafka/Connect - rejected: creates a confused-deputy risk; the server validates audience/issuer and uses a scoped downstream credential.
• Dynamic/remote tool descriptions - rejected: enables tool poisoning; tool manifests are static, signed, and version-pinned.
• Stateful HTTP sessions - rejected for HTTP mode: adds a session-hijacking surface and blocks horizontal scaling; the server is stateless on HTTP.

FAQ

Q: Why not extend Kafka Connect's REST API instead? Connect's REST API manages connectors, not Kafka itself; it cannot create topics, manage ACLs, produce/consume, or inspect groups.

Q: Should this be a separate Apache project? No - value comes from tight coupling with Kafka's client APIs and release cycle.

Q: Maintenance burden as MCP evolves? Small MCP surface (JSON-RPC dispatch, tool/resource registration, capability negotiation); the bulk is Kafka API wrappers that change rarely.

Q: Should we wait for MCP to stabilize? Core primitives and transports are stable; waiting cedes the ecosystem to vendor-specific implementations.

Q: Binary/non-string payloads? Phase 1 uses String serializers; binary can be base64-encoded; Avro/Protobuf/JSON Schema via pluggable serializers is Further Work.

Security FAQ. Q: How does this stop a poisoned Kafka message from making the agent delete a topic? The taint guard marks values read from topics as tainted and refuses to pass them as arguments to mutating tools without the human-approval gate; destructive tools also require out-of-band approval. Q: Can I restrict an agent to a sandbox? Yes - set mcp.allowed.topic.prefixes=agent. and mcp.readonly=true (or a narrow tool allow-list). Requests outside the prefix are rejected before any Kafka call. Q: Does the security add-on require new dependencies? No. Controls use existing libraries; the optional policy engine and distributed rate-limit backend are called over standard interfaces. Q: Is the server safe to scale across clouds? Yes - HTTP mode is stateless; run N replicas behind a load balancer with mcp.ratelimit.backend=distributed. It runs on any Kafka the Java client can reach (Apache, MSK, Confluent, Redpanda, Strimzi) and on-prem/air-gapped.