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 and pipelines. 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

Run it

One agent:

from pyrula.agents.kafka import KafkaAgentRunner

runner = KafkaAgentRunner.from_env(score, llm=my_llm)   # PYRULA_KAFKA_* for the broker
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())

With the default in-memory store, redelivery dedup spans the process lifetime. Pass a durable store= (Valkey or Postgres) so a restart does not re-run the LLM for offsets it already completed.

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. @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.