Skip to main content

Kafka as MCP Tools

See Streaming Data as Tools for why this pattern is worth reaching for. This tutorial covers the runnable steps.

Two capabilities, each documented on its own, combine into one useful pattern:

  • Kafka Gateway's http-kafka binding exposes a Kafka topic as a REST API with no application code (see Get Started and REST over Kafka).
  • AI Gateway's mcp-http · proxy binding exposes any HTTP API as MCP tools with no MCP server needed upstream (see Virtual Server).

Chain them and a Kafka topic becomes a set of MCP tools an AI agent can call directly: mcp-http · proxy in front, an http-kafka-backed REST API behind it, Kafka behind that.

This tutorial keeps the tool surface small on purpose: two tools, create_item and list_items, mapped onto the POST /items and GET /items routes from REST over Kafka. See the mcp-http proxy reference for the full set of options, including options.resources for exposing a Kafka-backed GET /items/{id} lookup as an MCP resource instead of a tool.

Prerequisites

The Kafka-Backed REST API

This tutorial reuses the exact http-kafka configuration from REST over Kafka, unmodified, as the rest-api service. It maps GET, POST, PUT, and DELETE on /items directly onto Kafka produce and fetch operations against the items-snapshots topic.

rest-api.yaml

Download rest-api.yaml

name: REST-example
bindings:
  # Proxy service entrypoint
  north_tcp_server:
    type: tcp
    kind: server
    options:
      host: 0.0.0.0
      port: 7114
    exit: north_http_server
  north_http_server:
    type: http
    kind: server
    routes:
      - when:
          - headers:
              :scheme: http
              :authority: localhost:7114
        exit: north_http_kafka_mapping

  # Proxy REST endpoints to a Kafka topic
  north_http_kafka_mapping:
    type: http-kafka
    kind: proxy
    routes:
      - when:
          - method: POST
            path: /items
        exit: north_kafka_cache_client
        with:
          capability: produce
          topic: items-snapshots
          key: ${idempotencyKey}
      - when:
          - method: GET
            path: /items
        exit: north_kafka_cache_client
        with:
          capability: fetch
          topic: items-snapshots
          merge:
            content-type: application/json
      - when:
          - method: GET
            path: /items/{id}
        exit: north_kafka_cache_client
        with:
          capability: fetch
          topic: items-snapshots
          filters:
            - key: ${params.id}
      - when:
          - method: PUT
            path: /items/{id}
        exit: north_kafka_cache_client
        with:
          capability: produce
          topic: items-snapshots
          key: ${params.id}
      - when:
          - method: DELETE
            path: /items/{id}
        exit: north_kafka_cache_client
        with:
          capability: produce
          topic: items-snapshots
          key: ${params.id}

  # Kafka sync layer
  north_kafka_cache_client:
    type: kafka
    kind: cache_client
    exit: south_kafka_cache_server
  south_kafka_cache_server:
    type: kafka
    kind: cache_server
    options:
      bootstrap:
        - items-snapshots
    exit: south_kafka_client

  # Connect to local Kafka
  south_kafka_client:
    type: kafka
    kind: client
    options:
      servers:
        - ${{env.KAFKA_BOOTSTRAP_SERVER}}
    exit: south_kafka_tcp_client
  south_kafka_tcp_client:
    type: tcp
    kind: client

telemetry:
  exporters:
    stdout_logs_exporter:
      type: stdout

See REST over Kafka for a full walkthrough of each route in this file. In this stack, rest-api isn't called by AI agents directly; it's an internal upstream that only mcp-gateway and your own testing reach.

Add the mcp-http Proxy

A second Zilla instance, mcp-gateway, terminates the AI agent's Streamable HTTP connection and routes tool calls by toolkit, same as any other AI Gateway deployment. The only new piece is the exit: instead of an mcp · client connecting to an upstream MCP server, an mcp-http · proxy binding names two tools and maps each one to an upstream request against the REST API.

Server

Terminates the inbound Streamable HTTP connection from the AI agent at /mcp, same as Get Started.

north_tcp_server:
  type: tcp
  kind: server
  options:
    host: 0.0.0.0
    port: 7114
  routes:
    - when:
        - port: 7114
      exit: north_http_server

north_http_server:
  type: http
  kind: server
  options:
    access-control:
      policy: cross-origin
  routes:
    - when:
        - headers:
            ":path": /mcp
      exit: north_mcp_server

north_mcp_server:
  type: mcp
  kind: server
  exit: north_mcp_proxy
  telemetry:
    metrics:
      - mcp.*
    attributes:
      method: ${mcp.method}
      tool: ${mcp.tool}
      outcome: ${mcp.outcome}

rest-api:7114 is rest-api's internal address on the Compose network; it's the same host and port an HTTP client inside the stack would use to reach it.

Full mcp-gateway.yaml config

Download mcp-gateway.yaml

#region catalog
catalogs:
  items_catalog:
    type: inline
    options:
      subjects:
        create_item_params:
          schema: |
            {
              "type": "object",
              "properties": {
                "idempotencyKey": {
                  "type": "string"
                },
                "greeting": {
                  "type": "string"
                }
              },
              "required": [
                "idempotencyKey",
                "greeting"
              ]
            }
        list_items_params:
          schema: |
            {
              "type": "object"
            }
        list_items_result:
          schema: |
            {
              "type": "array",
              "items": {
                "type": "object"
              }
            }
#endregion catalog

bindings:
  #region mcp_server
  north_tcp_server:
    type: tcp
    kind: server
    options:
      host: 0.0.0.0
      port: 7114
    routes:
      - when:
          - port: 7114
        exit: north_http_server

  north_http_server:
    type: http
    kind: server
    options:
      access-control:
        policy: cross-origin
    routes:
      - when:
          - headers:
              ":path": /mcp
        exit: north_mcp_server

  north_mcp_server:
    type: mcp
    kind: server
    exit: north_mcp_proxy
    telemetry:
      metrics:
        - mcp.*
      attributes:
        method: ${mcp.method}
        tool: ${mcp.tool}
        outcome: ${mcp.outcome}
  #endregion mcp_server

  #region mcp-http-proxy
  north_mcp_proxy:
    type: mcp
    kind: proxy
    routes:
      - exit: items_http_proxy
        when:
          - toolkit: items

  items_http_proxy:
    type: mcp-http
    kind: proxy
    options:
      tools:
        create_item:
          description: Create a new item on the items-snapshots Kafka topic.
          schemas:
            input:
              model: json
              catalog:
                items_catalog:
                  - subject: create_item_params
                    version: latest
        list_items:
          description: List every item on the items-snapshots Kafka topic.
          schemas:
            input:
              model: json
              catalog:
                items_catalog:
                  - subject: list_items_params
                    version: latest
            output:
              model: json
              catalog:
                items_catalog:
                  - subject: list_items_result
                    version: latest
    routes:
      - when:
          - tool: create_item
        exit: sys:http_client
        with:
          headers:
            ":method": POST
            ":scheme": http
            ":authority": rest-api:7114
            ":path": /items
            idempotency-key: ${args.idempotencyKey}
          body:
            template:
              greeting: ${args.greeting}
      - when:
          - tool: list_items
        exit: sys:http_client
        with:
          headers:
            ":method": GET
            ":scheme": http
            ":authority": rest-api:7114
            ":path": /items
  #endregion mcp-http-proxy

telemetry:
  metrics:
    - mcp.initialize
    - mcp.initialize.duration
    - mcp.tools.list
    - mcp.tools.list.duration
    - mcp.tools.call
    - mcp.tools.call.duration
  exporters:
    prometheus_exporter:
      type: prometheus
      options:
        endpoints:
          - scheme: http
            port: 7190
            path: /metrics

Run the Stack

Create rest-api.yaml, mcp-gateway.yaml, and docker-compose.yaml in the same directory, then start the stack:

docker-compose.yaml

Download docker-compose.yaml

version: '3'
services:

  rest-api:
    image: ghcr.io/aklivity/zilla:latest
    pull_policy: always
    depends_on:
      - kafka
    ports:
      - 7115:7114
    environment:
      KAFKA_BOOTSTRAP_SERVER: "kafka:29092"
    volumes:
      - ./rest-api.yaml:/etc/zilla/zilla.yaml
    command: start -v -e

  mcp-gateway:
    image: ghcr.io/aklivity/zilla:latest
    pull_policy: always
    depends_on:
      - rest-api
    ports:
      - 7114:7114
      - 7190:7190
    volumes:
      - ./mcp-gateway.yaml:/etc/zilla/zilla.yaml
    command: start -v -e

  kafka:
    image: bitnami/kafka:3.5
    hostname: kafka
    ports:
      - 9092:9092
      - 29092:9092
    environment:
      ALLOW_PLAINTEXT_LISTENER: "yes"
      KAFKA_CFG_NODE_ID: "1"
      KAFKA_CFG_BROKER_ID: "1"
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: "1@127.0.0.1:9093"
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: "CLIENT:PLAINTEXT,INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT"
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_CFG_LOG_DIRS: "/tmp/logs"
      KAFKA_CFG_PROCESS_ROLES: "broker,controller"
      KAFKA_CFG_LISTENERS: "CLIENT://:9092,INTERNAL://:29092,CONTROLLER://:9093"
      KAFKA_CFG_INTER_BROKER_LISTENER_NAME: "INTERNAL"
      KAFKA_CFG_ADVERTISED_LISTENERS: "CLIENT://localhost:9092,INTERNAL://kafka:29092"
      KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "true"

  kafka-init:
    image: bitnami/kafka:3.5
    command:
      - "/bin/bash"
      - "-c"
      -  |
        /opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server kafka:29092 --create --if-not-exists --topic items-snapshots
    depends_on:
      - kafka
    init: true
docker-compose up --detach

This starts Kafka, the items-snapshots topic, the rest-api Zilla instance, and the mcp-gateway Zilla instance in front of it.

Try It

Confirm the REST API works directly, the same as in REST over Kafka, but on port 7115 since mcp-gateway now owns port 7114:

curl -X POST http://localhost:7115/items -H 'Content-Type: application/json' -H 'Idempotency-Key: 1234' -d '{"greeting":"Hello, world"}'
curl http://localhost:7115/items
[{"greeting":"Hello, world"}]

Confirm the MCP gateway is up and exporting metrics:

curl http://localhost:7190/metrics

Point an MCP client at http://localhost:7114/mcp. The items toolkit is available alongside any other upstream you route to the same proxy, and calling create_item or list_items produces and fetches messages on the items-snapshots Kafka topic through rest-api, the same operations you just exercised with curl.

Remove the Running Containers

docker-compose down

Next Steps

  • Walk through MCP Gateway Setup for the same mcp-http · proxy pattern against a plain REST API, and Secure MCP with OAuth to guard this entrypoint with a jwt guard.
  • Read Virtual Server for how mcp · proxy merges listings across any number of upstreams, MCP or HTTP.
  • See Monitoring and Observability for dimensioning these metrics by toolkit and outcome.
  • Explore other HTTP-to-Kafka examples in the Zilla examples directory: http.kafka.async, http.kafka.cache, http.kafka.oneway, and http.kafka.sync.