feat(binding-mcp-kafka): implement mcp_kafka proxy binding#1714
Open
jfallows wants to merge 24 commits into
Open
feat(binding-mcp-kafka): implement mcp_kafka proxy binding#1714jfallows wants to merge 24 commits into
jfallows wants to merge 24 commits into
Conversation
jfallows
force-pushed
the
feature/1671-mcp-kafka
branch
2 times, most recently
from
781e8df to
3a5357f
Compare
jfallows
force-pushed
the
feature/1671-mcp-kafka
branch
4 times, most recently
from
bd42fa0 to
e728f7a
Compare
…upOffsets APIs; extend KafkaMetaDataEx and KafkaDescribeBeginEx - #1698: extend KafkaMetaDataEx with KafkaPartitionMetadata (replicas, isr arrays) and replicationFactor - #1699: add KafkaResourceType enum to KafkaDescribeBeginEx, rename topic->name, replace KafkaConfig with KafkaConfigDetail (isDefault, isSensitive) - #1700: add ListGroups (API key 16) support with KafkaListGroupsRequestBeginEx/ResponseBeginEx - #1701: add DescribeGroups (API key 15) support with KafkaDescribeGroupsRequestBeginEx/ResponseBeginEx - #1702: add AlterConsumerGroupOffsets (API key 53) support with request/response begin extensions Implements #1698, #1699, #1700, #1701, #1702 https://claude.ai/code/session_0174raBeXFTgt98bp4DTyRDm
Adds a new mcp_kafka proxy binding that bridges MCP (Model Context Protocol) streams to Apache Kafka, exposing Kafka broker operations as intrinsic MCP tools. Supports produce, consume, list_topics, describe_topic, create_topic, delete_topic, list_consumer_groups, describe_consumer_group, reset_offsets, list_brokers, describe_cluster, cluster_overview, describe_configs, and alter_configs tools. https://claude.ai/code/session_0174raBeXFTgt98bp4DTyRDm
…lterConsumerGroupOffsets change KafkaPartitionMetadata.replicas/isr declared as int32[], but the flyweight generator cannot use a primitive int as an array element type; switch to varuint32[] matching the convention used elsewhere (e.g. mqtt.idl subscriptionIds). KafkaResourceType's underlying type changed from uint8 to int8, breaking the one int-based valueOf() call site in KafkaClientAlterConfigsFactory. KafkaDescribeBeginEx.topic was renamed to .name; KafkaFunctions (spec builder DSL) and its test were not updated to match. KafkaCacheClientMetaFactory still referenced the old KafkaPartitionFW type for KafkaMetaDataEx.partitions, which is now KafkaPartitionMetadataFW. Also fixes several import-order checkstyle violations introduced by the original change.
… shapes
The module was written before the real mcp binding's McpBeginEx grew into a
union (lifecycle/toolsList/toolsCall/...) and before several kafka.idl
request shapes settled; it never compiled against that reality.
Add the missing scaffolding (COPYRIGHT/LICENSE/NOTICE/mvnw) required of every
new module. Point runtime/binding-mcp-kafka at only binding-mcp-kafka.spec
(provided) rather than the production binding-mcp/binding-kafka jars, and
have binding-mcp-kafka.spec depend on binding-mcp.spec + binding-kafka.spec
so their scopes propagate transitively as provided -- matching
binding-mcp-http's dependency shape rather than binding-mcp-http's own
(redundant) explicit spec deps. Delete the local mcp.idl stub, which
redefined a trivial McpBeginEx{kind} conflicting with the real union type;
the flyweight-maven-plugin's existing scopeNames (core internal mcp kafka)
now regenerates the real mcp + kafka scopes locally.
Rewrite McpKafkaProxyFactory to dispatch on the real McpBeginEx union kind:
lifecycle replies with a session handshake, toolsList replies with the
static list of intrinsic tools, and toolsCall resolves a route by tool name
and opens the Kafka-side proxy, building the begin extension against the
current kafka.idl request shapes.
Known gap: tool call arguments (topic name, payload, etc.) are not yet
parsed from the MCP request body, so every call currently targets an empty
resource name -- follow-up work, likely requiring new kafka.idl shapes.
No .rpt spec coverage was added; this is not yet test-first per the
project's discipline.
…s in KafkaCacheClientMetaFactory KafkaCacheClientMetaFactory.onMetaFanoutMemberOpened built a KafkaMetaDataEx.partitionsItem without setting the now-required replicationFactor (on the parent) or replicas/isr (on each partition), which are mandatory fields on the flyweight builder with no default value. This asserted at runtime whenever a meta fanout member reopened after the fanout was already established (CI: CacheFetchIT.shouldReceiveMessageKey). Same root cause as the fixes already applied elsewhere for this kafka.idl change; this call site was missed because it isn't one of the files the original commit touched.
binding-mqtt-kafka regenerates its own local copy of the kafka scope (same convention as binding-mcp-kafka), so it picked up the KafkaMetaDataEx.partitions type change from KafkaPartitionFW to KafkaPartitionMetadataFW along with the rest of the reactor. Only .partitionId() was used, so this is a type-only fix. CI: Analyze (java) autobuild failed to compile MqttKafkaSessionFactory.java.
…nMetadataFW types kafka.idl renamed KafkaConfigFW -> KafkaConfigDetailFW and KafkaPartitionFW -> KafkaPartitionMetadataFW, and KafkaDescribeBeginEx's topic field was renamed to name.
…Ex shapes kafka.idl now emits a leading replicationFactor before the partitions array in KafkaMetaDataEx, variable-length replicas/isr varuint32 arrays per partition, a leading resourceType byte in KafkaDescribeBeginEx, and trailing isDefault/isSensitive bytes on each KafkaConfigDetail. Update the Lua dissector to match and regenerate golden.lua and the affected IT fixtures.
…streams
Parse MCP tools/call arguments for produce/consume and wire them into
the existing merged(PRODUCE_ONLY/FETCH_ONLY) KafkaBeginEx shape, mirroring
how http-kafka/grpc-kafka target Kafka's typed app stream instead of raw
bytes. Produce completion is signaled by local flow-control draining then
the Kafka reply END (not WINDOW/ack, which only reflects local backpressure,
not broker acceptance) — mirrors HttpKafkaProxyFactory's no-reply produce
path. Consume accumulates fetched records up to a client-supplied limit
(clamped to 100) and stops the merged fetch stream once satisfied, or on a
Signaler-armed timeout derived from McpToolsCallBeginEx.timeout. Both paths
report results as MCP tool-call JSON envelopes ({"content":...,"isError":...}).
Also fixes a latent bug in KafkaProxy's stream construction: it opened the
downstream Kafka stream via newStream() before ever writing a real BEGIN
frame into the shared writeBuffer, so the call could observe stale bytes
left over from an unrelated stream. newKafkaStream() now builds the BEGIN
first and passes that buffer region through.
Adds test-first K3PO coverage under specs/binding-mcp-kafka.spec (dual
mcp/kafka script trees per specs/AGENTS.md convention) plus a runtime IT
and unit tests; jacoco threshold raised to reflect actual coverage.
Sibling cross-protocol proxies (binding-http-kafka, binding-mcp-http, binding-mqtt-kafka) were already declared as runtime dependencies here and listed in zpm.json.template; binding-mcp-kafka was missing from both, so it built and tested correctly but was never shipped in the packaged Zilla image.
Generates the internal kafka cache_client -> cache_server -> client -> tcp_client pipeline from options.servers/options.authorization, so mcp_kafka can own its own Kafka connection instead of requiring a hand-wired kafka binding via kind:proxy + exit. Modeled on AsyncapiClientGenerator's composite-generation pattern. - McpKafkaClientGenerator builds the 4-binding chain and resolves the generated kafka_client0's binding id - McpKafkaClientFactory (extends McpKafkaProxyFactory) attaches/detaches the composite and patches each route's exit id post-generation - McpKafkaOptionsConfig/Builder/Adapter expose servers + authorization, reusing KafkaServerConfig/KafkaAuthorizationConfig from binding-kafka - Schema patch splits into kind:proxy (unchanged) and kind:client (requires options.servers, optional options.authorization) branches - Unit tests cover the generator's chain shape, options propagation, and the factory's route-id-patching/attach-detach lifecycle; SchemaTest covers the new kind:client validation paths Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013ymTGGvFZY8S4M3EyP9L5r
Adds McpKafkaClientIT, exercising the full attach -> generate -> route-id-patch -> dispatch path against the same merged-stream .rpt scripts the existing kind:proxy IT already uses. Testability is via a new zilla.binding.mcp.kafka.cache.client.exit configuration property (McpKafkaConfiguration.cacheClientExit()), mirroring McpOpenapiCompositeGenerator's httpClientExit precedent: when unset (production default), the composite's exit resolves to the generated kafka_cache_client0 binding as before; when set, it bypasses the whole generated pipeline and resolves the override string directly via BindingConfig.resolveId, letting a test point mcp0 straight at an external k3po stub the same way a hand-wired kind:proxy config would. This also fixes a real bug the new IT caught: the composite's exit was resolving to kafka_client0 (a real kind:CLIENT kafka binding, which only understands broker-facing protocol, not the caller-facing merged extension) instead of kafka_cache_client0 (the binding that actually accepts the same merged KafkaBeginEx/DataEx shape McpKafkaProxyFactory builds) - mcp0 would never have successfully dispatched a real produce or consume call in production before this fix. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013ymTGGvFZY8S4M3EyP9L5r
Routes can now restrict which topics a produce/consume call is allowed to target via a topic matcher on when conditions (an allowlist: list multiple when entries with different topic values under one route), composable with the existing guard:/guarded route mechanism for per-caller/role restrictions - mirroring GrpcKafkaRouteConfig's established when+guarded composition pattern. Since the topic isn't known until the tool-call arguments are parsed (it arrives on DATA frames after BEGIN), route resolution for produce/consume is now deferred: newStream() only pre-checks that some route matches the tool name (preserving the existing no-matching-route behavior), and the real topic-aware + authorization-aware resolve happens once arguments are parsed. A topic that doesn't match any route's allowlist, or a route whose guard denies the caller, gets a plain RESET - distinct from the existing invalid-params RESET used for malformed argument JSON. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013ymTGGvFZY8S4M3EyP9L5r
Establishes a topics/model surface on McpKafkaOptionsConfig, wired into the generated cache_client0/cache_server0 bindings so kind:client gets schema-registry validation for produce/consume just like kind:proxy's hand-wired kafka cache pipeline.
…n produce rejection Decodes the KafkaResetExFW error code on a rejected produce stream and distinguishes INVALID_RECORD (schema/model validation failure) from a generic produce failure in the MCP tool result text.
Reads more naturally than the bare topic name, e.g. "orders topic" instead of just "orders".
…e too Matches the failure message wording for consistency.
…structuredContent Each consumed record's key, headers, and value are now captured (buffered, not yet streamed) and returned as an MCP structuredContent array alongside a short "Consumed N messages from topic X" content summary, matching the MCP spec's backward-compatible content+structuredContent shape and the in-repo precedent from binding-mcp's search_tools result. Real incremental streaming (one MCP DATA frame per Kafka record) is tracked separately.
…erator Injects tls_client0 between kafka_client0 and tcp_client0 whenever options.authorization is configured (SASL requires TLS regardless of mechanism) or a vault: is referenced on the binding. Relies on zilla#1576's VaultHandler wildcard resolution (initKeys()/initSigners()/initTrust(cacerts)) and binding-tls's own no-explicit-options auto-wildcard behavior, so no new mcp-kafka options surface is needed — the generated tls_client0 has empty options and an optional vault reference. Lifts the blanket "vault": false restriction in the schema patch for kind:client only; kind:proxy still rejects vault.
…crementally Replace whole-body buffering with common-json's JsonPipeline/JsonSink machinery, matching binding-mcp-http's established pattern: - McpKafkaArguments: terminal JsonSink that captures produce/consume scalar arguments (topic/key/value/partition/offset/limit) as they parse, driven through a BufferPool-backed decode slot with real window replenishment instead of a single upfront credit grant (fixes a latent bug where an args body larger than one window would stall forever). - McpKafkaConsumeResult: synthesizes the consume tool result envelope directly against a JsonSink-wrapped generator, one Kafka record at a time, resumable across encode-slot boundaries via SUSPENDED/resume, instead of buffering the whole JsonArrayBuilder result before one unbounded DATA send. - Reorders the consume envelope (structuredContent, with messages before count, ahead of content) since the summary text and count both depend on the final tally, which can't be known until the last record streams out. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013ymTGGvFZY8S4M3EyP9L5r
withEncodeSlot(Supplier<Status>) allocated a capturing lambda per call on the consume hot path with no precedent elsewhere in the codebase. Replace it with acquireEncodeSlot(), matching McpHttpProxyFactory's established boolean-check-then-inline-wrap pattern exactly, and rename two helpers to match this file's own conventions: invalidParamsResetEx -> buildXxx (as used by every other value-constructing helper here), onArgsCaptured -> completeArgs (the on* prefix is reserved for handlers of a received wire frame elsewhere in this file; this one only takes a traceId). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013ymTGGvFZY8S4M3EyP9L5r
…a broker Wires a new south_mcp_kafka_client (mcp_kafka kind:client) into the existing toolkit-aggregation example, guarded by kafka:tools like every other toolkit here. Its routes restrict produce/consume to the orders topic via the tool+topic allow-list form. Adds the same real, single-node KRaft-mode Kafka broker + one-shot topic-init service already used by the *.kafka.* examples elsewhere in this repo (apache/kafka:4.1.1), rather than the engine's type:test double specs/ITs use. Also extends tools-list-client to print a tool result's structuredContent (when present) alongside its text content, so test.sh and the README can verify the exact value that round-trips through produce -> consume against a real broker, not just a record count. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013ymTGGvFZY8S4M3EyP9L5r
CI showed real kafka__produce/consume calls succeeding against the live broker, but the full-scope assert_full_token check failed because both tools were missing from north_mcp_proxy's eager tools/list allow-list -- every other toolkit's key tools were already there.
north_mcp_proxy's tools/list cache hydrates by resolving routes against options.cache.authorization's shared credential, upstream of the eager.match filter added in c261ba9. south_mcp_kafka_client's route (guarded authn_jwt: [kafka:tools]) was never authorized for hydration because the shared JWT's scope lacked kafka:tools, so kafka__produce/ kafka__consume never entered the cache regardless of eager.match. Re-signed the same test JWT (same iss/aud/kid/exp, private.pem) with kafka:tools appended to scope.
jfallows
force-pushed
the
feature/1671-mcp-kafka
branch
from
e728f7a to
6a582dd
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Closes #2117 (sub-issue of #1671, which tracks the full
mcp_kafkatool surface)Adds a new
mcp_kafkaproxy binding that bridges MCP (Model Context Protocol) streams to Apache Kafka, implementing the data-plane tools:producemergedstream withPRODUCE_ONLYcapabilityconsumemergedstream withFETCH_ONLYcapabilityThe remaining tools from the original design (topic administration, cluster/broker overview, topic metadata, consumer group management, configs, and the
kafka://{topic}/messagesresource) are tracked as separate sub-issues of #1671, to land in follow-up PRs.New modules
runtime/binding-mcp-kafka/— proxy binding implementationspecs/binding-mcp-kafka.spec/— spec module with MCP IDL and JSON schema patchImplementation notes
McpKafkaProxyFactoryfollows the standard flyweight-on-factory pattern (all*RO/*RWfields on the factory, non-static inner stream classes)McpKafkaStateprovides bitmask helpers for initial/reply open/close lifecyclemcp_kafka(underscore convention for multi-word binding types)when[].tool+when[].topic(AND semantics — a route is scoped to exactly one topic per tool, not a bare tool-name dispatch), viaMcpKafkaConditionConfigAdapterkind: clientcomposite generator (McpKafkaClientGenerator), including TLS support for the generated Kafka clientcommon-json'sJsonPipeline/JsonSink, mirroringbinding-mcp-http's established patternexamples/mcp.proxy,kafka__produce/kafka__consumeround trip through the live gateway in CI)module-info.java); allinternal.*packages unexported; SPI registered via bothMETA-INF/services/andprovides...with...in module-infohttps://claude.ai/code/session_0174raBeXFTgt98bp4DTyRDm