Queues and consumers¶

Available · pip install agent-adapter (no extra)

Every queue gives you a message body and a delivery id. handle_envelope takes both, runs one call (or one resume) through the executor, and returns a JSON reply. It imports no broker library, so the same function serves SQS, Pub/Sub, Kafka, RQ, Dramatiq, Arq and Temporal.

File	Does
`agent_adapter/handlers/envelope.py`	`parse_envelope`: strict keys, ids from the delivery id
`agent_adapter/handlers/consume.py`	`handle_envelope(executor, payload, *, tools=, message_id=, source=)`
`examples/backend_queue.py`	an async consumer, a sync job, a Temporal activity, a Pub/Sub push body
`tests/docs/test_backend_handler_examples.py`	runs them over a fake queue, offline

Data enters as the message body (JSON object, text or bytes), goes to the executor, and leaves as the reply you ack with. The envelope format is on Serverless.

An async consumer¶

async def consume_once(queue: FakeQueue, *, source: str = "kafka") -> list[dict[str, Any]]:
    """One poll of an async consumer (aiokafka, aiobotocore SQS, Pub/Sub, Arq, Taskiq)."""
    replies = []
    for message in await queue.receive():
        try:
            reply = await handle_envelope(
                backend().executor, message.body, message_id=message.id, source=source
            )
        except AdapterError as exc:
            if not exc.retryable:  # a bad message or a denied call: retrying cannot help
                await queue.dead_letter(message)
            continue  # retryable: no ack, so the broker redelivers it
        await queue.ack(message)  # ok and pending_approval are both done for this message
        replies.append(reply)
    return replies

flowchart TD
  M["message (body, delivery id)"] --> H["handle_envelope"]
  H -- "ok / pending_approval" --> A["ack"]
  H -- "retryable error" --> N["no ack: redelivered"]
  H -- "other AdapterError" --> D["dead-letter, ack"]

Reply or error	Do
`ok`	ack
`pending_approval`	ack: the call waits in the store; a resume message runs it
`already_done`	ack: a redelivered resume; its decision already ran, nothing runs again
`exc.retryable` (`ToolTimeout`, model rate limits)	do not ack; the broker redelivers
any other `AdapterError` (bad envelope, denied, unknown tool)	dead-letter; retrying cannot help

Sync jobs and Temporal¶

RUNNER = LoopRunner()  # one warm loop per worker process, shared by every job


def tool_job(payload: Mapping[str, Any]) -> dict[str, Any]:
    """A sync job body (RQ, Dramatiq, Huey): errors raise, so the runner records them."""
    return RUNNER.run(lambda: handle_envelope(backend().executor, payload), timeout=TIMEOUT_S)


async def tool_activity(payload: Mapping[str, Any]) -> dict[str, Any]:
    """A Temporal activity: never call the adapter from workflow code (it is replayed)."""
    return await handle_envelope(backend().executor, payload)

Pub/Sub push¶

async def pubsub_push(body: Mapping[str, Any]) -> dict[str, Any]:
    """A Pub/Sub push request body: base64 `data`, a stable `messageId`."""
    message = body["message"]
    data = base64.b64decode(message["data"])
    return await handle_envelope(
        backend().executor, data, message_id=message["messageId"], source="pubsub"
    )

Per broker¶

Broker	Loop	Body and delivery id	Notes
SQS on Lambda	sync	use `tool_handler` (Serverless)	partial batch failures built in
SQS (aiobotocore)	async	`Body`, `MessageId`	delete after the reply; visibility timeout > the tool's timeout
Pub/Sub pull	async or callback threads	`message.data`, `message.message_id`	callback threads: `run_sync`, then `message.ack()`
Pub/Sub push	HTTP	`pubsub_push` above	keep the push endpoint authenticated (OIDC token)
Kafka (aiokafka, confluent AsyncIO)	async	`msg.value`, `f"{topic}:{partition}:{offset}"`	commit the offset after the reply
RQ	sync, fork per job	`tool_job(payload)`	the executor is built in the child (`functools.cache`)
Dramatiq, Huey	sync threads	`tool_job(payload)`	a long-lived loop (`LoopRunner`) saves a loop per job
Arq, Taskiq	async	`await handle_envelope(...)`	the job id as `message_id`
Celery	sync	see Celery
Temporal	async activities	`tool_activity`	never in workflow code; a pending approval becomes a signal wait, then a resume activity

Idempotency¶

Rule	Why
pass the delivery id as `message_id`, or put `run_id` and `call_id` in the envelope	a redelivered message is the same call: a `confirm` tool finds its own record instead of asking twice
use one `source` per broker	ids from two brokers never collide
make side-effecting tools idempotent (pass `call_id` to the external API)	a crash after the tool ran, before the ack, runs it again
a resume is safe to redeliver	its decision is used once; a duplicate replies `already_done` and runs nothing

Security¶

Anyone who can publish to the queue can run the tools you serve. Restrict publish rights, pass tools= to handle_envelope, and keep destructive tools behind confirm: a message never approves anything (framework approvals), only a reviewer's decision in the shared store does. Records the HTTP tool server parked are refused (ToolUsageError), whether a message names their approval_id or their run_id and call_id; they resume over the server's own /approvals routes. Logs name tools, ids and error kinds, never arguments or results.

See also: Serverless, Pipelines, Approval stores.