Kafka

Why Kafka Through Chio

A Kafka consumer triggers agent work. An event arrives on a topic, the agent decides what to do, it calls tools in response, and it emits result events downstream. Every link in that chain is a capability boundary: the agent must be authorized to consume the inbound topic, authorized to call each tool, and authorized to produce to each outbound topic. Chio evaluates all three.

Kafka provides ACLs for topic read and write, but ACLs do not model scope, budget, guard pipelines, or signed attestation. They answer whether a principal can read a topic, not whether this specific message, with this specific content, in this specific choreography, is permitted to drive a tool call right now.

Consumer-Side Enforcement

Chio does not touch the broker. The broker stays a dumb pipe, which is important for compatibility with managed services like MSK and Confluent Cloud. Evaluation runs inside the consumer process, next to the agent process, and every poll goes through the kernel before it is handed to application code:

Two capability boundaries exist per event. First, consumption: is this agent authorized to see this topic and this message? Second, tool invocation: for each tool the agent calls in response, is the call permitted? Production is a third boundary when the agent writes result events back out. Each of the three produces a distinct receipt, and receipt references link the consumption, tool call, and production decisions.

Transactional Receipt Commit

Kafka's exactly-once semantics let Chio commit an offset and a receipt together. Either both succeed or both roll back. If the agent crashes between the tool call and the commit, the event is redelivered and the receipt envelope for the aborted attempt never reaches the receipt topic, so its absence is the signal: an attempt whose envelope is on the topic ran to commit, and one whose envelope is missing did not. There is no rolled-back flag on a receipt. The sidecar may still hold a receipt of its own for the aborted evaluation.

Install

The core distribution is chio-streaming, which pulls in chio-sdk-python, the distribution providing the chio_sdk module. Each broker client is an extra, and PEP 562 lazy imports keep an uninstalled broker's client library from loading:

# core only (Kafka middleware compiles but confluent-kafka is not installed)
pip install chio-streaming

# pick your brokers
pip install "chio-streaming[kafka]"
pip install "chio-streaming[nats]"
pip install "chio-streaming[pulsar]"
pip install "chio-streaming[eventbridge]"
pip install "chio-streaming[pubsub]"
pip install "chio-streaming[redis]"
pip install "chio-streaming[flink]"

# everything
pip install "chio-streaming[all]"

Consumer Middleware

The Kafka path is ChioConsumerMiddleware, exported at the top level of chio_streaming. It wraps a confluent-kafka consumer and producer, evaluates every polled message against a capability before your handler runs, and routes denials to a DLQRouter. Kafka is the only broker with native exactly-once, so it gets the transactional path by default: the offset commit lands atomically with the receipt publish (on allow) or the DLQ publish (on deny).

import asyncio

from chio_sdk.client import ChioClient
from chio_streaming import (
    ChioConsumerConfig,
    ChioConsumerMiddleware,
    DLQRouter,
)
from confluent_kafka import Consumer, Producer

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "research-agents",
    "enable.auto.commit": False,
    "isolation.level": "read_committed",
})

producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "transactional.id": "research-agents-tx",
    "enable.idempotence": True,
})
producer.init_transactions()


async def run() -> None:
    consumer.subscribe(["research-tasks"])

    async with ChioClient("http://127.0.0.1:9090") as chio:
        middleware = ChioConsumerMiddleware(
            consumer=consumer,
            producer=producer,
            chio_client=chio,
            dlq_router=DLQRouter(default_topic="chio-denied-events"),
            config=ChioConsumerConfig(
                capability_id="cap-research-agents",
                tool_server="kafka://prod",
                scope_map={"research-tasks": "events:consume:research-tasks"},
                receipt_topic="chio-receipts",
                transactional=True,
                max_in_flight=32,
                consumer_group_id="research-agents",
            ),
        )

        async def handle(msg, receipt):
            # Reached only on allow; receipt is the signed ChioReceipt.
            # Tool calls inside the handler are separately evaluated via
            # the standard chio SDK (chio-sdk-python / chio-fastapi / ...).
            process(msg)

        while True:
            await middleware.poll_and_process(handle)


asyncio.run(run())

The middleware resolves each message's tool name from config.scope_map (keyed on topic), evaluates it against capability_id / tool_server through the sidecar, and dispatches to your handler only on allow. A deny is routed to the DLQ topic chosen by DLQRouter. Drive the loop with await middleware.poll_and_process(handle); there is no separate .poll() / .commit() pair, and no standalone producer-wrapper class. Outbound receipt production runs through the same transactional middleware.

Transactional and Best-Effort Modes

ChioConsumerConfig.transactional selects the commit strategy. There is no separate processor class; the same ChioConsumerMiddleware drives both modes.

Side effects your handler performs (HTTP, database writes) are not part of the Kafka transaction; use an outbox if they must be atomic with the offset. A cross-cluster DLQ is not atomic either, so keep the DLQ co-located. The sidecar RPC is also outside the transaction: on abort the sidecar may still have recorded a receipt, but the receipt envelope only appears on the receipt topic when the transaction commits.

Shared Primitives

Every broker middleware is built on the same primitives in chio_streaming.core, chio_streaming.receipt, and chio_streaming.dlq, plus the error types in chio_streaming.errors. Integrating any broker means building on this API:

The Flink path guarantees the same bytes: the receipt side output equals build_envelope(...).value exactly and the DLQ side output equals DLQRouter.build_record(...).value exactly, so a single downstream consumer can audit ingress across every broker regardless of source.

Schema Registry as a Guard Input

Schema Registry governs what the data looks like. Chio governs what agents do with the data. The two compose, but not by having the guard call the registry: a guard is a WASM component whose only host imports are log, get_config, get_time_unix_secs and fetch_blob (wit/chio-guard/world.wit:22-27), so it has no network egress at all. Resolve the schema outside the sandbox and hand the guard the answer through config:

import json

from guard import Guard as BaseGuard
from guard.imports.types import GuardRequest, Verdict_Allow, Verdict_Deny

from chio_guard import get_config


class Guard(BaseGuard):
    """Require data:pii:read when the event carries a PII-tagged field.

    evaluate is synchronous and takes a GuardRequest: no topic field, no
    has_scope helper. The PII field list for each topic is resolved from
    Schema Registry by the deployment and passed in as guard config.
    """

    def evaluate(self, request: GuardRequest) -> Verdict_Allow | Verdict_Deny:
        arguments = json.loads(request.arguments or "{}")
        topic = arguments.get("topic", "")

        pii_by_topic = json.loads(get_config("pii_fields_by_topic") or "{}")
        pii_fields = [
            field for field in pii_by_topic.get(topic, [])
            if field in arguments.get("event", {})
        ]

        if pii_fields and "data:pii:read" not in request.scopes:
            return Verdict_Deny(
                f"event carries PII fields {sorted(pii_fields)}; "
                "requires data:pii:read scope"
            )
        return Verdict_Allow()

The shapes above are the real ones: evaluate is def, not async def (wit/chio-guard/world.wit:41, sdks/guard/chio-guard-py/examples/tool-gate/app.py:26), the verdict types are Verdict_Allow and Verdict_Deny(reason), and the granted scopes arrive as a plain list[str] on request.scopes (sdks/guard/chio-guard-py/src/chio_guard/types.py:23-51).

Dead Letter Queue as a Security Signal

In a traditional Kafka system, the DLQ is a place for messages whose processing crashed. In a chio-governed system it is something else: a feed of messages an agent was not authorized to process. That shift turns the DLQ from an error channel into a security channel. High DLQ volume is not a bug, it is evidence that an agent is trying to do things it does not have capabilities for.

Traditional DLQ:
  Event -> Consumer -> Processing failed -> DLQ
  Meaning: "We tried and couldn't."

chio-governed DLQ:
  Event -> Consumer -> chio denied -> DLQ + signed denial receipt
  Meaning: "We were not authorized to process this."

The DLQ becomes a security signal:
  - High DLQ volume indicates unauthorized action attempts
  - Repeating denial patterns surface misconfigured capabilities or attacks
  - Receipt-enriched DLQ is an auditable proof of enforcement

DLQ routing is handled by DLQRouter, which the middleware calls on every deny. It picks the destination topic (an exact topic_map match, then default_topic) and builds a self-describing DLQRecord: the denial reason, the guard, the receipt id and full receipt, the originating topic / partition / offset, and the original value.

from chio_streaming import DLQRouter

router = DLQRouter(
    default_topic="chio-denied-events",
    topic_map={"orders": "chio-denied-orders"},
    include_original_value=True,   # embed original bytes as utf8 or hex
)

# The middleware builds one DLQRecord per denied event. Its payload,
# shown expanded (on the wire it is canonical JSON: keys sorted, no
# whitespace):
# {
#   "version":        "chio-streaming/dlq/v1",
#   "request_id":     "...",
#   "verdict":        "deny",
#   "reason":         "...",
#   "guard":          "...",
#   "receipt_id":     "...",
#   "receipt":        { ...full ChioReceipt... },
#   "source":         {"topic": "orders", "partition": 3, "offset": 4021},
#   "original_value": {"utf8": "..."}
# }
# Headers on the DLQ record:
#   X-Chio-Receipt, X-Chio-Verdict, X-Chio-Deny-Guard, X-Chio-Deny-Reason

Load the DLQ into a data warehouse and denial patterns become queryable: group by reason, scope, and agent identity over a rolling window, and threshold alerts fire when a specific agent starts attempting actions outside its scope.

Receipt Correlation Across Choreography

In a choreography without a coordinator, correlate receipts to build a cross-service view. The convention is that each event a handler produces carries the receipt id in the same X-Chio-Receipt header the middleware uses on its own records, and the verdict in X-Chio-Verdict. Your handler sets those: the middleware writes the receipt envelope and the DLQ record and does not touch what the handler produces (chio_streaming/middleware.py:385-439). The receipt envelope on the receipt topic is keyed on the request_id the receipt is associated with. A downstream consumer reads the inbound header, correlates it against the receipt store, and records receipts for its resulting tool calls. The header and request_id provide the correlation lineage.

Receipts themselves are inspected out of band through the CLI receipt commands, chio receipt (list, explain, audit, and the checkpoint subcommands), or by querying the receipt store directly.

Supported Brokers

Kafka is the reference implementation. The consumer-side model below uses the same evaluation steps, receipt envelope, and DLQ handling; broker-specific acknowledgment and ordering semantics differ. Install the selected broker package with the matching extra, for example pip install "chio-streaming[nats]"; PEP 562 lazy imports keep an unused broker's client library from loading.

Flink is the one non-transactional engine: it uses side outputs instead of driving transactions itself, because Flink provides exactly-once processing through aligned checkpoints and 2PC sinks. The Python operator requires apache-flink>=2.2.0,<3 and produces wire-identical build_envelope / DLQRouter.build_record output.

The Flink Operator on the JVM

A Flink job is often a JVM job, so the operator exists twice: chio_streaming.flink in Python and world.chio.flink in Kotlin, under sdks/jvm/chio-streaming-flink. The class names, the configuration keys, and the side-output tag names are the same in both. Cross-language byte parity is pinned by the @Tag("parity") vector tests in chio-sdk-jvm, which compare against hand-computed Python vectors (sdks/jvm/chio-sdk-jvm/README.md:105-109); each language additionally runs its own end-to-end suite against a real Kafka source on a Redpanda broker. The JVM README claims its Flink integration test asserts byte equality across the two languages (sdks/jvm/chio-streaming-flink/README.md:141-142); it does not, and cannot as written, because the two suites build receipts from different fixtures. Java 21 is the floor, which is Flink 2.2's own minimum.

# OutputTag instances must be reused between emission and collection,
# so they are module-level singletons here. The operator constructs
# matching tags internally via the same name + type information.
RECEIPT_TAG = OutputTag(RECEIPT_TAG_NAME, Types.PICKLED_BYTE_ARRAY())
DLQ_TAG = OutputTag(DLQ_TAG_NAME, Types.PICKLED_BYTE_ARRAY())

config = ChioFlinkConfig(
    capability_id="cap-fraud-scoring",
    tool_server="flink://fraud-job",
    client_factory=build_chio_client,
    dlq_router_factory=build_dlq_router,
    scope_map={"transactions": "events:consume:transactions"},
    receipt_topic="chio-fraud-receipts",
    max_in_flight=64,
    on_sidecar_error="deny",
    subject_extractor=lambda _event: "transactions",
)

evaluated = AsyncDataStream.unordered_wait(
    transactions,
    ChioAsyncEvaluateFunction(config),
    Time.milliseconds(10_000),
    128,
    Types.PICKLED_BYTE_ARRAY(),
)

split = evaluated.process(ChioVerdictSplitFunction())
receipts = split.get_side_output(RECEIPT_TAG)
dlq = split.get_side_output(DLQ_TAG)

val config =
    ChioFlinkConfig
        .builder<Transaction>()
        .capabilityId("cap-fraud")
        .toolServer("flink://fraud-job")
        .subjectExtractor { "transactions" }
        .clientFactory { ChioClient("http://127.0.0.1:9090") }
        .dlqRouterFactory { DlqRouter(defaultTopic = "chio-fraud-dlq") }
        .scopeMap(mapOf("transactions" to "events:consume:transactions"))
        .receiptTopic("chio-fraud-receipts")
        .maxInFlight(64)
        .onSidecarError(SidecarErrorBehaviour.DENY)
        .build()

val evaluated =
    AsyncDataStream.unorderedWait(
        transactions,
        ChioAsyncEvaluateFunction(config),
        10_000L,
        TimeUnit.MILLISECONDS,
        128,
    )

val split = evaluated.process(ChioVerdictSplitFunction<Transaction>())
val receiptTag = OutputTag(ChioOutputTags.RECEIPT_TAG_NAME, Types.PRIMITIVE_ARRAY(Types.BYTE))
val dlqTag = OutputTag(ChioOutputTags.DLQ_TAG_NAME, Types.PRIMITIVE_ARRAY(Types.BYTE))

The two tabs stop at different points. The Python one ends with the side outputs collected; the Kotlin one ends with the tags constructed, and the README's next two lines (sdks/jvm/chio-streaming-flink/README.md:97-98) collect and sink them. Nothing differs between the languages here: the tag names are the same two strings in both.

Fail-closed behaviour is named rather than implied on the JVM side: SidecarErrorBehaviour.DENY synthesises a deny receipt carrying the chio-streaming/synthetic-deny/v1 marker, routes it to the DLQ, and keeps the pipeline flowing; SidecarErrorBehaviour.RAISE propagates a ChioStreamingError so Flink restarts the task and the source rewinds. Only ChioError subtypes count as sidecar failures; any other exception propagates unchanged, matching chio_streaming.core.evaluate_with_chio.

Package Layout

sdks/python/chio-streaming/
  pyproject.toml             # deps: chio-sdk-python, chio-adapter-base, pydantic; per-broker extras
  src/chio_streaming/
    __init__.py              # top-level exports + PEP 562 lazy broker imports
    core.py                  # ChioClientLike, BaseProcessingOutcome
    receipt.py               # ReceiptEnvelope, build_envelope, ENVELOPE_VERSION
    dlq.py                   # DLQRouter, DLQRecord
    errors.py                # ChioStreamingError, ChioStreamingConfigError
    middleware.py            # Kafka: ChioConsumerMiddleware, ChioConsumerConfig
    nats.py                  # ChioNatsMiddleware / build_nats_middleware
    pulsar.py                # ChioPulsarMiddleware / build_pulsar_middleware
    eventbridge.py           # ChioEventBridgeHandler / build_eventbridge_handler
    pubsub.py                # ChioPubSubMiddleware / build_pubsub_middleware
    redis_streams.py         # ChioRedisStreamsMiddleware / build_redis_streams_middleware
    flink.py                 # ChioAsyncEvaluateFunction, ChioVerdictSplitFunction

Open Questions

• Broker-level enforcement. This design evaluates at the consumer, not the broker. Should Chio ship a Kafka interceptor plugin or NATS authorization callout that evaluates at the broker level? Pro: earlier enforcement. Con: broker coupling, latency on the hot path.
• Compacted topics. Kafka compacted topics retain the latest value per key. If a capability is revoked after an event is compacted, should the agent still be able to consume the compacted event based on the original attestation?
• Multi-cluster streaming. MirrorMaker and Confluent Cluster Linking replicate events across clusters. Should receipts replicate with the events, or should each cluster maintain its own receipt log with cross-cluster federation?
• Backpressure. If Chio denies a high volume of events, the DLQ can become the bottleneck. Should the consumer apply backpressure to the source topic, or should a high denial rate trigger a consumer group circuit breaker?
• Event replay. Consumers can reset offsets and replay events. Should Chio re-evaluate capabilities on replay, since they may have changed, or honor the original evaluation recorded in the receipt log?

Next Steps

• AWS Lambda · run chio as a Lambda extension alongside serverless tool servers
• Temporal · orchestrated workflows that complement choreographed streams
• Budgets · per-group, per-consumer spending envelopes for streaming agents