Kafka Workflows

@kafka_workflow runs a durable @workflow per Kafka record: consume, decode, run the workflow’s checkpointed steps, encode, produce, with exactly-once delivery, DLQ-before-processing, retries, and redelivery dedup. It is the same runtime that powers @kafka_agent, minus the LLM loop.

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(...),                   # optional; durable redelivery dedup
)
asyncio.run(worker.run())

With the default in-memory store, redelivery dedup spans the process lifetime. Pass a durable store= so a restart doesn’t re-run already-completed offsets.

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 gives a durable=False binding no store automatically; passing store= with durable=False is an error. 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, since a parked run would stall its partition indefinitely. 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.