# Kafka

> Drive an agent from a Kafka topic with exactly-once processing.

Source: https://docs.pyrula.io/agents/serving/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](https://docs.pyrula.io/agents/authoring/tools.md#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

```python
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:

```python
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

```python
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:

```python
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](https://docs.pyrula.io/workflows/guides/kafka.md#a-store-outage-holds-the-record)).
- **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](https://docs.pyrula.io/workflows/guides/kafka-guarantees.md)). Use `@kafka_workflow` when the per-record work has no LLM loop.