Kafka

@kafka_agent binds an @agent to Kafka topics. Each record becomes one agent turn: consume, decode and validate the payload, run the agent, encode the result, produce it. Decode failures route to a dead-letter topic before the agent ever runs, and a redelivered record reuses the already-computed result instead of calling the LLM again.

It is the @kafka_workflow runtime plus the agent loop: @kafka_agent is @kafka_workflow with an @agent on top.

pip install 'pyrula-agents[kafka]'

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 *_schema params accept a pydantic model (sugar for JSON), a wire Format (Json/Avro/JsonSchema/Protobuf/Raw/String), or any MessageCodec. The same model defines the schema everywhere it is used, the same way it does for @kafka_workflow. Registry-backed formats read the URL from schema_registry= or PYRULA_SCHEMA_REGISTRY_URL.

The Kafka envelope (topic, partition, offset, headers) is reachable inside 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]) gets one envelope per record from kafka_envelopes(ctx); kafka_envelope(ctx) raises there. kafka_envelopes also works on a single-record turn (a one-element list), so a handler can always use it.

Run it

One agent:

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

The store is where redelivery dedup lives, and for an agent that is the line item that costs money: with a durable store, a redelivered offset after a restart reuses the stored result instead of re-running the LLM. InMemoryStore() is accepted for tests and single-process demos - dedup then spans this process only. A durable binding with no store raises at construction rather than silently defaulting to memory.

Delivery and errors

• Exactly-once processing, at-least-once output. The result is committed to the durable store with the offset, then produced. A redelivered record after a crash re-produces the stored result without re-running the agent. Dedupe downstream on the record key, or set an exactly-once sink.
• DLQ before the LLM. A record that fails to decode or validate is poison and goes to {input_topic}.dlq before any agent call. Set on_error to "dlq" (default), "skip", or "fail".
• Retries. Each policy retries max_retries times with backoff before its on_error disposition.
• Batch mode. input_schema=list[Order] delivers a list[Order] of whatever one poll drained per handler call. The batch is the retry, on_error, and commit unit.

Relation to workflows

The streaming mechanics (delivery, DLQ, dedup, commit, EOS) live in pyrula.workflows.kafka and are identical for agents and workflows - the runtime guarantee matrix proves it, row by row. @kafka_agent adds the turn-scoped store and the LLM execution. Use @kafka_workflow when the per-record work is a durable workflow with no LLM loop.