Kafka Workflows

@kafka_workflow runs a durable @workflow per Kafka record: consume, decode, run the workflow’s checkpointed steps, encode, produce. It handles exactly-once processing, retries, redelivery dedup, and routes undecodable records to a dead-letter queue (DLQ) before the workflow runs. It is the same runtime that powers @kafka_agent, minus the LLM loop.

Two guarantees, worth keeping distinct. Processing is exactly-once by default: a redelivered record whose run already completed is not re-run - the result is read back from the run store. Output delivery is at-least-once by default: a crash between produce and offset commit re-produces on redelivery, never loses. Dedupe downstream on key, or set delivery="exactly_once" for transactionally atomic output+offset.

Every guarantee in this guide has a proving test; the runtime guarantee matrix lists them by binding and mode.

It lives in the [kafka] extra:

pip install 'pyrula-workflows[kafka]'

Bind a workflow to topics

from pydantic import BaseModel
from pyrula.workflows.kafka import kafka_workflow, KafkaWorker

class Event(BaseModel):
    id: int
    ip: str

class Enriched(BaseModel):
    id: int
    geo: str

@kafka_workflow(input_topic="events", group="enricher",
                output_topic="events.enriched",
                input_schema=Event, output_schema=Enriched)
async def enrich(ctx, event: Event) -> Enriched:
    geo = await ctx.step("geo", lambda: lookup(event.ip))
    return Enriched(id=event.id, geo=geo)

The *_schema params take a pydantic model (sugar for JSON), a wire Format (Json/Avro/JsonSchema/Protobuf/Raw/String), or any MessageCodec. Decode failures are poison and route to the DLQ before the workflow runs.

Run it

import asyncio
from pyrula.workflows.kafka import KafkaConnection, KafkaWorker
from pyrula.workflows.stores.valkey import ValkeyStore

worker = KafkaWorker(
    [enrich],
    connection=KafkaConnection.from_env(),   # PYRULA_KAFKA_* ; from_env is the default
    store=ValkeyStore(...),                   # run state + redelivery dedup live here
)
asyncio.run(worker.run())

Run state and redelivery dedup live in the store, so the store decides what a restart costs. With ValkeyStore, a redelivered offset after a crash reuses the stored result. With MemoryStore(), dedup only covers the current process, and a restart re-runs completed records. A durable binding needs one of the two passed explicitly: with no store it raises at construction rather than falling back to MemoryStore.

from pyrula.workflows.stores.memory import MemoryStore

worker = KafkaWorker([enrich], store=MemoryStore())   # tests, single-process demos

Skip durability for throughput

For an idempotent handler that doesn’t need checkpoint resume or exactly-once processing, set durable=False to run it straight through with no run-event store, no redelivery dedup, and no replay:

@kafka_workflow(
    input_topic="events", group="enricher",
    input_schema=Event, output_topic="events.enriched", output_schema=Enriched,
    durable=False,
)
async def enrich(ctx, event: Event) -> Enriched:
    return Enriched(id=event.id, geo=lookup(event.ip))

This is at-least-once processing: a retried or redelivered record re-runs and re-produces, so dedupe downstream on the record key. ctx.step still works but executes inline without persisting, so a retry re-runs every step. KafkaWorker hands its store= only to durable bindings (and to the cloud bridge’s ack ledger); a durable=False binding never sees it. A KafkaRunner built directly with store= and durable=False raises. durable=False cannot be combined with delivery="exactly_once", which needs the run store.

Straight-through only

Kafka workflows must run to completion in one pass. A run that pauses (ctx.sleep, ctx.interrupt, ctx.wait_for_signal) is treated as an error and follows the binding’s on_error policy (dlq by default, fail or skip), since a parked run would stall its partition. A retried record resumes the existing run from its checkpointed steps, so completed steps are not re-run.

Relation to agents

@kafka_agent (in pyrula.agents.kafka) is this exact runtime plus an LLM agent loop: @kafka_agent = @kafka_workflow + @agent. The streaming mechanics (delivery, DLQ, dedup, commit) are identical; agents add the turn-scoped store and the LLM execution on top.