feat(binding-mcp-kafka): implement mcp_kafka proxy binding

cloud/docker-image/pom.xml

@@ -145,6 +145,12 @@
       <version>${project.version}</version>
       <scope>runtime</scope>
     </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>binding-mcp-kafka</artifactId>
+      <version>${project.version}</version>
+      <scope>runtime</scope>
+    </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>binding-mcp-openapi</artifactId>

cloud/docker-image/src/main/docker/zpm.json.template

@@ -27,6 +27,7 @@
     "io.aklivity.zilla:binding-kafka",
     "io.aklivity.zilla:binding-mcp",
     "io.aklivity.zilla:binding-mcp-http",
+    "io.aklivity.zilla:binding-mcp-kafka",
     "io.aklivity.zilla:binding-mcp-openapi",
     "io.aklivity.zilla:binding-mqtt",
     "io.aklivity.zilla:binding-mqtt-kafka",

examples/mcp.proxy/.github/test.sh

@@ -34,6 +34,11 @@
 #      the same result as calling it directly
 #  14. the cold tool is still directly callable by name -- "cold" only ever
 #      changes what tools/list reports, never what tools/call accepts
+#  15. kafka__produce writes a record to a real, single-node KRaft Kafka
+#      broker (mcp_kafka kind:client's own generated cache_client/client/
+#      tcp_client pipeline, not the engine's test double)
+#  16. kafka__consume reads that same record back, round-tripping the exact
+#      value through the real broker
 #
 # Streamable HTTP responses arrive as Server-Sent Events; checks grep the
 # streamed body / client output rather than asserting exact-string equality.
@@ -66,7 +71,7 @@ encode_jwt() {
 JWT_NONE=""
 JWT_URLELICIT=$(encode_jwt "urlelicit:authorize")
 JWT_PARTIAL=$(encode_jwt "github:tools petstore:tools")
-JWT_FULL=$(encode_jwt "urlelicit:authorize github:tools github:pr:write petstore:tools pets:write")
+JWT_FULL=$(encode_jwt "urlelicit:authorize github:tools github:pr:write petstore:tools pets:write kafka:tools")
 
 # WHEN: a url-elicitation-capable client initializes against the gateway
 # THEN: the gateway negotiates protocol version 2025-11-25 in the response
@@ -139,6 +144,7 @@ assert_no_token() {
     ! echo "$TOOLS_NONE" | grep -q '^urlelicit__' &&
     ! echo "$TOOLS_NONE" | grep -q '^github__' &&
     ! echo "$TOOLS_NONE" | grep -q '^petstore__' &&
+    ! echo "$TOOLS_NONE" | grep -q '^kafka__' &&
     ! echo "$TOOLS_NONE" | grep -q 'petstore+' &&
     ! echo "$TOOLS_NONE" | grep -q 'github+'
 }
@@ -148,6 +154,7 @@ if echo "$TOOLS_NONE" | grep -q '^everything__' &&
     ! echo "$TOOLS_NONE" | grep -q '^urlelicit__' &&
     ! echo "$TOOLS_NONE" | grep -q '^github__' &&
     ! echo "$TOOLS_NONE" | grep -q '^petstore__' &&
+    ! echo "$TOOLS_NONE" | grep -q '^kafka__' &&
     ! echo "$TOOLS_NONE" | grep -q 'petstore+' &&
     ! echo "$TOOLS_NONE" | grep -q 'github+'; then
   echo "✅ no token: only the ungated everything toolkit is listed"
@@ -173,7 +180,8 @@ assert_partial_token() {
     echo "$TOOLS_PARTIAL" | grep -q '^template:github+pr://{owner}/{repo}/{number}$' &&
     ! echo "$TOOLS_PARTIAL" | grep -q '^petstore__create_pet$' &&
     ! echo "$TOOLS_PARTIAL" | grep -q '^github__create_pr$' &&
-    ! echo "$TOOLS_PARTIAL" | grep -q '^urlelicit__'
+    ! echo "$TOOLS_PARTIAL" | grep -q '^urlelicit__' &&
+    ! echo "$TOOLS_PARTIAL" | grep -q '^kafka__'
 }
 retry_until 5 3 assert_partial_token
 echo "TOOLS_PARTIAL=$TOOLS_PARTIAL"
@@ -184,7 +192,8 @@ if echo "$TOOLS_PARTIAL" | grep -q '^petstore__list_pets$' &&
     echo "$TOOLS_PARTIAL" | grep -q '^template:github+pr://{owner}/{repo}/{number}$' &&
     ! echo "$TOOLS_PARTIAL" | grep -q '^petstore__create_pet$' &&
     ! echo "$TOOLS_PARTIAL" | grep -q '^github__create_pr$' &&
-    ! echo "$TOOLS_PARTIAL" | grep -q '^urlelicit__'; then
+    ! echo "$TOOLS_PARTIAL" | grep -q '^urlelicit__' &&
+    ! echo "$TOOLS_PARTIAL" | grep -q '^kafka__'; then
   echo "✅ toolkit-only scope: sees list_pets, search_pets, and all three read-only resources, but not create_pet or create_pr"
 else
   echo "❌ toolkit-only scope did not layer as expected"
@@ -203,7 +212,9 @@ assert_full_token() {
     echo "$TOOLS_FULL" | grep -q '^petstore__create_pet$' &&
     echo "$TOOLS_FULL" | grep -q '^resource:petstore+/pets/featured$' &&
     echo "$TOOLS_FULL" | grep -q '^template:petstore+/pets/{petId}$' &&
-    echo "$TOOLS_FULL" | grep -q '^template:github+pr://{owner}/{repo}/{number}$'
+    echo "$TOOLS_FULL" | grep -q '^template:github+pr://{owner}/{repo}/{number}$' &&
+    echo "$TOOLS_FULL" | grep -q '^kafka__produce$' &&
+    echo "$TOOLS_FULL" | grep -q '^kafka__consume$'
 }
 retry_until 5 3 assert_full_token
 echo "TOOLS_FULL=$TOOLS_FULL"
@@ -215,7 +226,9 @@ if echo "$TOOLS_FULL" | grep -q '^everything__' &&
     echo "$TOOLS_FULL" | grep -q '^petstore__create_pet$' &&
     echo "$TOOLS_FULL" | grep -q '^resource:petstore+/pets/featured$' &&
     echo "$TOOLS_FULL" | grep -q '^template:petstore+/pets/{petId}$' &&
-    echo "$TOOLS_FULL" | grep -q '^template:github+pr://{owner}/{repo}/{number}$'; then
+    echo "$TOOLS_FULL" | grep -q '^template:github+pr://{owner}/{repo}/{number}$' &&
+    echo "$TOOLS_FULL" | grep -q '^kafka__produce$' &&
+    echo "$TOOLS_FULL" | grep -q '^kafka__consume$'; then
   echo "✅ full scope: every toolkit's tools and resources are listed"
 else
   echo "❌ full scope did not unlock every toolkit"
@@ -434,4 +447,48 @@ else
   EXIT=1
 fi
 
+# WHEN: a kafka:tools-scoped caller calls kafka__produce
+# THEN: the record reaches the real, single-node KRaft Kafka broker started by
+#       this example -- not the engine's `type: test` double specs/ITs use --
+#       proving mcp_kafka's kind:client composite generator (kafka_cache_client
+#       -> kafka_client -> tcp_client) talks to an actual broker end to end
+call_kafka_produce() {
+  KAFKA_PRODUCE_OUT=$(docker compose run --rm --no-deps \
+      -e JWT_TOKEN="$JWT_FULL" \
+      -e MCP_URL="http://zilla:$PORT/mcp" \
+      -e CALL_TOOL="kafka__produce" \
+      -e CALL_ARGS='{"topic":"orders","value":"hello from mcp-kafka"}' \
+      tools-list-client 2>&1)
+  echo "$KAFKA_PRODUCE_OUT" | grep -q 'Produced record to orders topic'
+}
+retry_until 10 3 call_kafka_produce
+echo "KAFKA_PRODUCE_OUT=$KAFKA_PRODUCE_OUT"
+if echo "$KAFKA_PRODUCE_OUT" | grep -q 'Produced record to orders topic'; then
+  echo "✅ kafka__produce wrote a record to the real Kafka broker"
+else
+  echo "❌ kafka__produce did not succeed against the real broker"
+  EXIT=1
+fi
+
+# WHEN: that same caller calls kafka__consume for the same topic
+# THEN: the exact value produced above comes back in structuredContent.messages
+#       -- round-tripping through the real broker, not just proving a count
+call_kafka_consume() {
+  KAFKA_CONSUME_OUT=$(docker compose run --rm --no-deps \
+      -e JWT_TOKEN="$JWT_FULL" \
+      -e MCP_URL="http://zilla:$PORT/mcp" \
+      -e CALL_TOOL="kafka__consume" \
+      -e CALL_ARGS='{"topic":"orders","limit":1}' \
+      tools-list-client 2>&1)
+  echo "$KAFKA_CONSUME_OUT" | grep -q 'hello from mcp-kafka'
+}
+retry_until 10 3 call_kafka_consume
+echo "KAFKA_CONSUME_OUT=$KAFKA_CONSUME_OUT"
+if echo "$KAFKA_CONSUME_OUT" | grep -q 'hello from mcp-kafka'; then
+  echo "✅ kafka__consume read the produced record back from the real Kafka broker"
+else
+  echo "❌ kafka__consume did not read the produced record back"
+  EXIT=1
+fi
+
 exit $EXIT

examples/mcp.proxy/README.md

@@ -10,25 +10,25 @@ synthesized `zilla__search_tools` keyword-search tool instead of crowding out
 every `tools/list` response.
 
 ```text
-                                     ┌────────────────────────────────── Zilla ───────────────────────────────────┐
-                                     │  tcp(7114) → http → mcp(server, jwt) → mcp(proxy, guarded routes)           │
-client ── Authorization: Bearer ───►│                                     │                                       │
-                                     │            ┌────────────┬──────────┴───────────┬───────────────┐          │
-                                     │            ▼            ▼                      ▼               ▼          │
-                                     │    mcp(client)     mcp(client)          mcp_http(proxy)   mcp_openapi(client)
-                                     │    everything       urlelicit           github toolkit     petstore toolkit
-                                     │       │                │                      │                   │        │
-                                     │       ▼                ▼                      ▼                   ▼        │
-                                     │     http →           http →              http(client) →      http(client) →
-                                     │     tcp               tcp                 tcp                   tcp        │
-                                     └───────┼────────────────┼──────────────────────┼───────────────────┼───────┘
-                                             ▼                ▼                      ▼                   ▼
-                                      everything:3001   urlelicit:3003          ghapi:4001         petstore:4002
-                                      (reference server) (url-mode elicitation) (mock GitHub API)  (mock REST API,
-                                                                                                     OpenAPI-described)
+                                     ┌───────────────────────────────────── Zilla ─────────────────────────────────────┐
+                                     │  tcp(7114) → http → mcp(server, jwt) → mcp(proxy, guarded routes)                │
+client ── Authorization: Bearer ───►│                                     │                                            │
+                                     │            ┌────────────┬──────────┴───────────┬───────────────┬───────────┐  │
+                                     │            ▼            ▼                      ▼               ▼           ▼  │
+                                     │    mcp(client)     mcp(client)          mcp_http(proxy)   mcp_openapi(client)  mcp_kafka(client)
+                                     │    everything       urlelicit           github toolkit     petstore toolkit   kafka toolkit
+                                     │       │                │                      │                   │             │      │
+                                     │       ▼                ▼                      ▼                   ▼             ▼      │
+                                     │     http →           http →              http(client) →      http(client) →  kafka_cache_client →
+                                     │     tcp               tcp                 tcp                   tcp             kafka_client → tcp
+                                     └───────┼────────────────┼──────────────────────┼───────────────────┼─────────────┼──────┘
+                                             ▼                ▼                      ▼                   ▼             ▼
+                                      everything:3001   urlelicit:3003          ghapi:4001         petstore:4002   kafka.examples.dev:9092
+                                      (reference server) (url-mode elicitation) (mock GitHub API)  (mock REST API,  (real, single-node
+                                                                                                     OpenAPI-described) KRaft Kafka broker)
 ```
 
-This one configuration exercises all five `mcp*` binding kinds:
+This one configuration exercises all six `mcp*` binding kinds:
 
 | Binding | Kind | Role in this example |
 | --- | --- | --- |
@@ -37,6 +37,7 @@ This one configuration exercises all five `mcp*` binding kinds:
 | `mcp` | `client` | Talks to an upstream server that is itself MCP (`everything`, `urlelicit`); `urlelicit` also forwards the caller's own JWT upstream |
 | `mcp_http` | `proxy` | Synthesizes MCP tools from hand-authored config, backed by a plain REST API (`github` toolkit) |
 | `mcp_openapi` | `client` | Synthesizes MCP tools from an OpenAPI document, backed by a plain REST API (`petstore` toolkit) |
+| `mcp_kafka` | `client` | Exposes Kafka broker operations (`produce`/`consume`) as intrinsic MCP tools, generating its own `kafka_cache_client → kafka_client → tcp_client` pipeline against a real broker (`kafka` toolkit) |
 
 ## Authorization model
 
@@ -53,6 +54,7 @@ the pipeline, each demonstrating a different mechanism:
 | `mcp_http(proxy)` route for `create_pr` | a second, tool-specific `routes[].guarded`, layered under the `mcp_http` binding's base guarded route | `github:tools` **and** `github:pr:write` |
 | `mcp(proxy)` route for `petstore` toolkit | `routes[].guarded` on the toolkit route | `petstore:tools` |
 | `mcp_openapi(client)` operation `create_pet` | the OpenAPI document's own `security` requirement, mapped to `authn_jwt` via `options.specs.petstore.security` | `petstore:tools` **and** `pets:write` |
+| `mcp(proxy)` route for `kafka` toolkit | `routes[].guarded` on the toolkit route | `kafka:tools` |
 
 `list_pets`, `list_featured_pets`, and `get_pet` declare no OpenAPI `security`
 of their own, so they need only the toolkit-level `petstore:tools` scope --
@@ -78,11 +80,13 @@ exist.
 docker compose up -d
 ```
 
-This starts Zilla plus four locally-reachable upstream services: a Node
+This starts Zilla plus five locally-reachable upstream services: a Node
 `everything` reference MCP server on `:3001`, a minimal `urlelicit` MCP server
-on `:3003` demonstrating url-mode elicitation, and two plain REST mocks --
+on `:3003` demonstrating url-mode elicitation, two plain REST mocks --
 `ghapi` on `:4001` (subset of the GitHub API) and `petstore` on `:4002`
-(a small Petstore API, described by an inline OpenAPI document).
+(a small Petstore API, described by an inline OpenAPI document) -- and a real,
+single-node KRaft-mode Kafka broker on `:9092` (`kafka.examples.dev`), with an
+`orders` topic created by the one-shot `kafka-init` service.
 
 ## Verify
 
@@ -114,7 +118,7 @@ export JWT_TOKEN=$(docker compose run --rm jwt-cli encode \
     --aud "https://api.example.com" \
     --exp=+1d \
     --no-iat \
-    --payload "scope=urlelicit:authorize github:tools github:pr:write petstore:tools pets:write" \
+    --payload "scope=urlelicit:authorize github:tools github:pr:write petstore:tools pets:write kafka:tools" \
     --secret @/private.pem | tr -d '\r\n')
 ```
 
@@ -146,7 +150,7 @@ export JWT_TOKEN=$(docker compose run --rm jwt-cli encode \
     --alg "RS256" --kid "example" \
     --iss "https://auth.example.com" --aud "https://api.example.com" \
     --exp=+1d --no-iat \
-    --payload "scope=urlelicit:authorize github:tools github:pr:write petstore:tools pets:write" \
+    --payload "scope=urlelicit:authorize github:tools github:pr:write petstore:tools pets:write kafka:tools" \
     --secret @/private.pem | tr -d '\r\n')
 docker compose run --rm -e JWT_TOKEN="$JWT_TOKEN" tools-list-client
 ```
@@ -348,6 +352,53 @@ docker compose run --rm -e JWT_TOKEN="$JWT_TOKEN" \
 `petstore`'s log line reads `search_pets query: {"tag":"cat"}` -- the caller
 said `category`, the request said `tag`.
 
+### Produce and consume through a real Kafka broker
+
+`south_mcp_kafka_client` is an `mcp_kafka` `kind: client` binding -- unlike
+every other toolkit in this example, it does not proxy to a separate MCP or
+REST server. It generates its own `kafka_cache_client → kafka_client →
+tcp_client` pipeline directly from `options.servers`, talking to the real,
+single-node KRaft-mode Kafka broker this example starts
+(`kafka.examples.dev:9092`, the `kafka` compose service; the `orders` topic
+is created by the one-shot `kafka-init` service, though
+`KAFKA_AUTO_CREATE_TOPICS_ENABLE` would create it anyway on first produce).
+
+Both routes below restrict their tool to exactly the `orders` topic --
+`tool` and `topic` together in one `when` form an allow-list, not just a
+dispatch by tool name; a `produce`/`consume` call naming any other topic has
+no matching route and is rejected.
+
+Produce a record with the full-scope `$JWT_TOKEN` from above:
+
+```bash
+docker compose run --rm -e JWT_TOKEN="$JWT_TOKEN" \
+    -e CALL_TOOL=kafka__produce \
+    -e CALL_ARGS='{"topic":"orders","value":"hello from mcp-kafka"}' \
+    tools-list-client
+# Produced record to orders topic
+```
+
+Then consume it back -- `limit` bounds how many records one call returns
+(1-100, default 10); with nothing else produced to `orders` yet, the earliest
+record is exactly the one just written:
+
+```bash
+docker compose run --rm -e JWT_TOKEN="$JWT_TOKEN" \
+    -e CALL_TOOL=kafka__consume \
+    -e CALL_ARGS='{"topic":"orders","limit":1}' \
+    tools-list-client
+# Consumed 1 messages from topic orders
+# {"topic":"orders","messages":[{"key":null,"headers":[],"value":"hello from mcp-kafka"}],"count":1}
+```
+
+The second line is the tool result's raw `structuredContent` -- each message
+carries its Kafka `key` (`null` here, since `produce` was not given one),
+`headers`, and `value`. Both `produce` and `consume` stream incrementally
+rather than buffering the whole request or result in memory: arguments are
+parsed as they arrive, and consumed records are written to the reply as each
+one comes off the broker, so neither tool is bounded by how large a single
+value or a full result set happens to be.
+
 ### Trigger a form elicitation round-trip
 
 Call the `everything` server's elicitation-demo tool through the gateway (no
@@ -452,6 +503,7 @@ docker compose down -v
 
 - [Zilla docs -- `mcp` bindings](https://docs.aklivity.io/zilla/latest/reference/config/bindings/mcp/README.html)
 - [Zilla docs -- `mcp-http` binding](https://docs.aklivity.io/zilla/latest/reference/config/bindings/mcp-http/README.html)
+- [Zilla docs -- `mcp-kafka` binding](https://docs.aklivity.io/zilla/latest/reference/config/bindings/mcp-kafka/README.html)
 - [Zilla docs -- `jwt` guard](https://docs.aklivity.io/zilla/latest/reference/config/guards/jwt.html)
 - [MCP -- Streamable HTTP transport](https://modelcontextprotocol.io/docs/concepts/transports)
 - [MCP -- elicitation](https://modelcontextprotocol.io/specification/2025-11-25/client/elicitation)