Kafka

@kafka_agent is @kafka_workflow with an @agent on top. Each record becomes one agent turn: consume, decode and validate, run the agent, encode the result, produce it. Decode failures go to a dead-letter topic before the agent runs, and a redelivered record reuses the stored result instead of calling the LLM again.

pip install "pyrula[agents,kafka]"

A Kafka turn can’t wait for a person: no ctx.interrupt, no requires_approval tool (Approval). A requires_approval=True tool fails the turn with ToolRegistrationError before the model is called; an MCP server with requires_approval fails the bind.

Bind an agent

from pydantic import BaseModel
from pyrula.agents.decorators.agent import agent
from pyrula.agents.kafka import kafka_agent

class Order(BaseModel):
    id: int
    text: str

class Scored(BaseModel):
    id: int
    risk: float

@kafka_agent(input_topic="orders.raw", output_topic="orders.scored", group="scorer",
             input_schema=Order, output_schema=Scored)
@agent(output_type=Scored)
async def score(ctx, order: Order) -> Scored:
    ...

The @agent underneath is optional. Without it, @kafka_agent(name=, timeout=600, capture="lifecycle", llm=, mcp_servers=) builds one; stacked, those arguments are ignored and the @agent owns them.

*_schema takes a pydantic model (JSON), a wire Format (Json/Avro/JsonSchema/Protobuf/Raw/String) or any MessageCodec. No input_schema: the handler gets raw bytes. No output_schema: result is produced as JSON. input_key_schema / output_key_schema do the same for keys. A bytes result fails the run (it’s journaled); return str with String. Registry-backed formats read the URL from schema_registry= or PYRULA_SCHEMA_REGISTRY_URL. Each registry request is bounded at 5 s; raise it with PYRULA_SCHEMA_REGISTRY_TIMEOUT_MS (or schema_registry_timeout_ms on KafkaConnection), up to 300000.

The record’s coordinates are available in the handler:

from pyrula.agents.kafka import kafka_envelope

env = kafka_envelope(ctx)   # env.topic, env.partition, env.offset, env.headers

A batch turn (input_schema=list[Model]) uses kafka_envelopes(ctx) (one per record; kafka_envelope raises there). kafka_envelopes also works on single-record turns.

Run it

from pyrula.agents.kafka import KafkaAgentRunner
from pyrula.agents.stores.valkey import ValkeyStore

runner = KafkaAgentRunner.from_env(          # PYRULA_KAFKA_* for the broker
    score, llm=my_llm, store=ValkeyStore(url=...),
)
await runner.run()

Several agents and workflows in one process share a connection and store:

import asyncio
from pyrula.agents.kafka import KafkaWorker

worker = KafkaWorker([score, classify, enrich], llm=my_llm, store=ValkeyStore(...))
asyncio.run(worker.run())

Redelivery dedup lives in the store: with a durable one, a redelivered offset after a restart reuses the stored result instead of re-running the LLM. InMemoryStore() is for tests and demos (dedup spans one process). A durable binding with no store raises at construction.

Delivery and errors

• Exactly-once processing. A turn’s run id is group:topic:partition:offset:fp, where fp is 16 hex characters of a hash over timestamp, key, value and headers, so a recreated topic’s record at a reused offset is a new turn. A redelivered record re-produces the stored result without re-running the agent. The input offset lives in Kafka, not the store.

• Output. The default delivery="at_least_once" produces then commits, so a crash between the two produces again; dedupe downstream on the pyrula.run_id header. delivery="exactly_once" makes output and offset commit one transaction for read_committed consumers.

  Without KIP-447 group fencing, a zombie consumer can commit duplicates under exactly_once.

• DLQ. A record that fails to decode or validate goes to dlq_topic (default {input_topic}.dlq) before any agent call. on_error is "dlq" (default), "skip" or "fail".

• Retries. A failing turn retries max_retries (default 2) times with backoff, then applies on_error. max_concurrency (8) bounds in-flight records. A store outage is not a retry: the record is held and re-claimed with capped backoff, never dead-lettered, until the store answers or max_store_requeues ends the turn (Kafka workflows).

• Batch mode. input_schema=list[Order] delivers whatever one poll drained per call (max_batch_size, batch_linger_ms shape it); the batch is the retry, on_error and commit unit.

Streaming mechanics are shared with workflows (runtime guarantee matrix). Use @kafka_workflow when the per-record work has no LLM loop.