Kafka

A consumer that drives agent work from a Kafka topic hands each polled record to the kernel first. Your handler runs only for records the kernel admits, and every decision leaves a signed receipt beside the offset it settles.

The check lives in the consumer process, between poll() and your handler, as ChioConsumerMiddleware from the chio-streaming package. It calls the Chio sidecar at 127.0.0.1:9090, sends refused records to a dead-letter topic, and with transactional=True commits the receipt and the offset in one Kafka transaction. The broker (MSK, Confluent Cloud, or self-hosted) and its ACLs stay in place.

Notice what each layer decides. Broker ACLs decide which principal may read a topic. The kernel decides whether this record, under this capability, may drive this handler, and it checks scope and guards before signing the result. Tool calls and result events the handler makes afterwards are separate governed calls, each with its own authority and receipt.

1. Install the adapter with its Kafka extra

The core distribution is chio-streaming, which depends on chio-sdk. Each broker client is an extra, and PEP 562 lazy imports keep an uninstalled broker's client library from loading.

You should see Kafka adapter ready. The import succeeds without a running broker; the next step connects one.

2. Wrap the consumer and producer

Give the middleware a confluent-kafka consumer, a transactional producer, and a ChioClient pointed at the sidecar. Set CHIO_KAFKA_CAPABILITY_ID to a capability that covers the tool server and the scope mapped for each topic.

import asyncio
import os

from application import process
from chio import ChioClient
from chio_streaming import ChioConsumerConfig, ChioConsumerMiddleware, DLQRouter
from confluent_kafka import Consumer, Producer

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "order-agents",
    "enable.auto.commit": False,
    "isolation.level": "read_committed",
})
producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "transactional.id": "order-agents-tx",
    "enable.idempotence": True,
})
producer.init_transactions()


async def handle(msg, receipt):
    # Reached only on allow; receipt is the signed ChioReceipt.
    process(msg)


async def run() -> None:
    consumer.subscribe(["orders"])
    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",
                topic_map={"orders": "chio-denied-orders"},
                include_original_value=True,
            ),
            config=ChioConsumerConfig(
                capability_id=os.environ["CHIO_KAFKA_CAPABILITY_ID"],
                tool_server="kafka://prod",
                scope_map={"orders": "events:consume:orders"},
                receipt_topic="chio-receipts",
                transactional=True,
                max_in_flight=32,
                consumer_group_id="order-agents",
            ),
        )
        try:
            while True:
                outcome = await middleware.poll_and_process(handle)
                if outcome is not None and not outcome.acked:
                    raise RuntimeError(
                        f"record not committed: {outcome.request_id}"
                    ) from outcome.handler_error
        finally:
            consumer.close()
            producer.flush()


asyncio.run(run())

The middleware resolves each record's tool name from scope_map, keyed on topic, and evaluates it against capability_id and tool_server. One call, await middleware.poll_and_process(handle), polls, evaluates, dispatches, and commits; there is no separate poll and commit pair and no standalone producer wrapper. The middleware API is async, so the loop runs under asyncio.

With transactional=True, the default, ChioConsumerConfig requires receipt_topic and consumer_group_id, and the producer needs a transactional.id and a completed init_transactions(). A configuration missing any of them raises ChioStreamingConfigError at construction.

3. Find a refused record on the dead-letter topic

When the kernel refuses a record, DLQRouter picks the destination (an exact topic_map match, then default_topic) and writes one DLQRecord in the same transaction as the offset commit. Read it back from the topic the router chose for orders.

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic chio-denied-orders --from-beginning --max-messages 1 | jq .

{
  "guard": "…",
  "original_value": { "utf8": "…" },
  "reason": "…",
  "receipt": { … },
  "receipt_id": "rcpt-01d3…7a20",
  "request_id": "chio-evt-…",
  "source": { "offset": 4021, "partition": 3, "topic": "orders" },
  "verdict": "deny",
  "version": "chio-streaming/dlq/v1"
}

You should see a self-describing denial: the reason and guard, the receipt id and the full receipt, the source topic, partition, and offset, and the original value as UTF-8 text (or a hex marker when the bytes do not decode). On the wire the value is canonical JSON with sorted keys. The record also carries four headers: X-Chio-Receipt, X-Chio-Verdict, X-Chio-Deny-Guard, and X-Chio-Deny-Reason.

The dead-letter topic is a feed of records an agent was refused, separate from records whose processing crashed. Load it into a warehouse, group refusals by reason, guard, scope, and agent identity over a rolling window, and alert when one agent starts reaching outside its scope. Repeating patterns point at a misconfigured capability or an attack.

4. Read the receipt committed with its offset

On allow, the middleware produces a receipt envelope to chio-receipts and commits the source offset in one transaction. Read with read_committed so aborted records stay invisible.

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic chio-receipts --isolation-level read_committed \
  --property print.headers=true --property print.key=true --max-messages 1

X-Chio-Receipt:rcpt-01d3…6b14,X-Chio-Verdict:allow  chio-evt-…  {"receipt":{…},"request_id":"chio-evt-…","source_offset":4022,"source_partition":3,"source_topic":"orders","verdict":"allow","version":"chio-streaming/v1"}

You should see the receipt id and verdict in the headers, the request_id as the record key, and the envelope: version chio-streaming/v1, the full receipt, and the source coordinates. If the transaction aborts, its records stay invisible, the offset stays uncommitted, and Kafka redelivers the record. The sidecar call sits outside the transaction, so the authority may hold a receipt for an aborted attempt; retain its log independently when your evidence policy requires it.

The commit mode decides what lands with the offset

ChioConsumerConfig.transactional selects the strategy; the same middleware drives both modes.

Best-effort mode suits brokers without exactly-once support, and it alone accepts on_sidecar_error="deny": a sidecar that is down or times out then yields a local deny routed through the DLQ. That setting covers sidecar failures; a kernel deny always routes to the DLQ. A local refusal, from that setting or from a ChioDeniedError, carries the chio-streaming/synthetic-deny/v1 marker, an [unsigned] reason prefix, and sentinel signature values, so diagnose with it and take authority evidence from signed receipts.

Check each outcome before polling again. A handler failure returns acked=False with handler_error, and the live consumer stays where it is; committing a later offset would skip the failed record. Stop or seek according to your retry policy. Effects your handler performs (HTTP calls, database writes) sit outside the Kafka transaction, so use an outbox when they must commit with the offset, deduplicate them on redelivery, and keep the DLQ on the source cluster, since a remote DLQ cannot share its transaction.

• Replay. Resetting offsets evaluates each record again against current authority. Historical receipts stay as evidence; application idempotency handles repeated effects.
• Replication. Define receipt retention and correlation across clusters explicitly.
• Backpressure. Size the DLQ, the publication timeout, and the consumer pause policy for bursts of refusals, and monitor uncommitted outcomes beside denial rates. max_in_flight caps concurrent evaluations against the sidecar.

Produced events carry the receipt to the next consumer

A choreography has no coordinator, so receipts link the services. Your handler sets X-Chio-Receipt and X-Chio-Verdict on each event it produces, the same headers the middleware writes on its own records; the middleware leaves handler output untouched.

An order service produces order.placed with its receipt id in the header. The payment service's middleware evaluates the record under events:consume:orders and signs a receipt; its handler calls charge_payment as a separate governed call with its own receipt, then produces payment.charged carrying that id. Each downstream consumer reads the inbound header, correlates it against the receipt store, and records receipts for its own calls. The header plus the request_id key reconstruct the trace; inspect any receipt with chio receipt list, explain, audit, or the checkpoint subcommands.

Read more in Receipts.

Schema classification reaches guards as configuration

The middleware sends the kernel the topic, headers, and a digest of the body, never decoded fields. A guard is a WASM component whose only host imports are log, get_config, get_time_unix_secs, and fetch_blob, so it has no network egress and cannot call Schema Registry. Resolve each topic's schema and PII classification in the deployment, then 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 for a topic whose schema has PII-tagged fields.

    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 "{}")
        if topic not in pii_by_topic:
            return Verdict_Deny("schema classification missing for topic")
        # Kafka evaluation carries metadata and a body digest, not event fields.
        pii_fields = pii_by_topic[topic]

        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()

Read more in Custom guards.

Streaming adapter reference

# 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))
val receipts = split.getSideOutput(receiptTag)
val dlq = split.getSideOutput(dlqTag)

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

Next: govern the work the handler starts