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 will be redelivered and the receipt for the aborted attempt is flagged as rolled back, so auditors can distinguish an invocation that ran from a transient failure.

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. Integrating any broker means changing 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: a chio guard can read the schema for a topic and identify sensitive fields, then require a stronger scope when those fields are present:

class PiiFilterGuard:
    async def evaluate(self, context):
        schema = await schema_registry.get_schema(context.topic)
        pii_fields = [f for f in schema.fields if f.has_tag("pii")]

        if pii_fields and not context.has_scope("data:pii:read"):
            return Deny(
                f"Event contains PII fields {pii_fields}, "
                f"requires data:pii:read scope"
            )
        return Allow()

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 canonical
# JSON payload:
# {
#   "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. Each produced event includes the receipt id in itsX-Chio-Receipt header and the verdict in X-Chio-Verdict; 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 trust receipt (list, explain, format, and the checkpoint-verification 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 PyFlink side outputs instead of driving transactions itself, because Flink provides exactly-once processing through aligned checkpoints and 2PC sinks. It requires apache-flink>=2.2.0,<3.0 and produces wire-identical build_envelope / DLQRouter.build_record output.

Package Layout

sdks/python/chio-streaming/
  pyproject.toml             # deps: chio-sdk-python; 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