Delivery Service

kit.deliver() brings content into a room from outside it — a background task's result, an external event, a scheduled nudge, something another room produced — at a moment the room's channels allow: voice playback, user speech, idle detection. The room's agents read it and act on it.

Quick start

from roomkit import RoomKit, WaitForIdle

kit = RoomKit(delivery_strategy=WaitForIdle(buffer=3.0))

# A background search finished: the room's agent reads it and tells the user
await kit.deliver("room-id", content="Flight search finished: 3 options found.")

Not a way to text the room's correspondent

On a text transport (SMS, email, WhatsApp, WebSocket...), the content enters the room as a message received on that channel, from the framework's own sender (system). It is stored, the room's agents read it and answer it, and the correspondent receives their answer, never the content itself; with no agent in the room, nothing is sent. A deliver("room", "Your payment was confirmed.") on an SMS room reports sent and texts nothing. To send the correspondent a message, see Sending a message to the correspondent.

Use cases

• Delegation results — workers finish in the background, results delivered to user
• External events — webhook arrives, voice agent mentions it
• Scheduled notifications — timer fires, agent speaks
• Cross-room results — something in room B relevant to room A

Strategies

Strategies control when content is delivered:

from roomkit import Immediate, WaitForIdle, Queued

# Send now; on voice, it plays once the playback under way is sent whole
kit = RoomKit(delivery_strategy=Immediate())

# Wait for AI + user silence, then deliver after buffer
kit = RoomKit(delivery_strategy=WaitForIdle(buffer=3.0))

# Batch multiple deliveries into one message at next idle window
kit = RoomKit(delivery_strategy=Queued(buffer=2.0, separator="\n\n"))

Strategy	When it delivers	Best for
Immediate()	Now; on voice, once the playback under way is sent whole (call interrupt() first to cut it)	Urgent alerts, text channels
WaitForIdle(buffer)	After AI stops speaking + user stops talking + buffer	Voice conversations
Queued(buffer, separator)	Batches multiple items, delivers at next idle	High-frequency results

String shorthand:

await kit.deliver("room", content="hello", strategy="immediate")
await kit.deliver("room", content="hello", strategy="wait_for_idle")
await kit.deliver("room", content="hello", strategy="queued")

WaitForIdle details

WaitForIdle is voice-aware:

• VoiceChannel: waits for wait_playback_done() (TTS finished) + buffer
• A channel hosting a realtime model (RealtimeVoiceChannel, RealtimeAudioVideoChannel, a ConferenceChannel with a realtime model plugged in): waits for wait_idle() (the model's answer ended and reached the transport, nobody heard speaking) + buffer
• Text channels: delivers immediately (no playback to wait for)

WaitForIdle(
    buffer=3.0,            # seconds to wait after idle detected
    playback_timeout=15.0, # max seconds to wait for playback
)

Channel-aware delivery

kit.deliver() auto-detects the best transport channel in the room. It picks, among the transports bound to the room:

1. A voice channel first (most latency-sensitive): a voice channel type, or any channel whose realtime model is plugged in at delivery time (a conference with one included), whatever order it was attached in;
2. otherwise the first other transport bound (WebSocket, SMS, etc.).

Each kind then receives the content its own way:

• A channel hosting a realtime model — injected via inject_text() into the model's session (system intent with instruction=True)
• VoiceChannel — synthetic inbound message → TTS
• Other transports — synthetic inbound message

An intelligence channel attached as a transport is never picked: an instruction delivered through it would re-enter the agent it is for.

Override with channel_id:

await kit.deliver("room", content="hello", channel_id="voice-main")

Naming a ConferenceChannel with a realtime model plugged in injects into the model's room session, and is unavailable (voice_session_unavailable) until that session connects. A strategy that waits pins the session first: if the model is unplugged during the wait, the delivery is refused (voice_session_replaced), never published to the room instead.

A channel of your own takes part by inheriting RealtimeModelHost (get_room_sessions(), inject_text(), wait_idle()); hosts_realtime_model(channel) is the predicate the delivery paths read.

Sending a message to the correspondent

A message the host itself sends to the room's customer (a confirmation, a verification link) is published from a channel of the host's own with send_event. The room's transports deliver it like any other message:

from roomkit import TextContent
from roomkit.channels import WebSocketChannel

kit.register_channel(WebSocketChannel("bank"))   # the host's own channel
await kit.attach_channel(room_id, "bank")

await kit.send_event(
    room_id, "bank", TextContent(body="Your payment was confirmed."),
    addressed_to=[],   # no agent is asked to answer it
)

The message stays in the room's history, where the agents read it on their next turn; restrict its visibility to the transports that should carry it to keep it from them.

Framework default

Set the default strategy on RoomKit:

kit = RoomKit(delivery_strategy=WaitForIdle(buffer=3.0))

# All deliver() calls use WaitForIdle unless overridden
await kit.deliver("room", content="result")

# Override per call
await kit.deliver("room", content="urgent!", strategy=Immediate())

Observable outcomes

Every call returns a DeliveryOutcome:

Status	Meaning
queued	The delivery backend accepted the item; no transmission is established yet.
sent	The text event was published, or the realtime provider accepted the injection.
blocked	A hook, room state or permission refused the request.
unavailable	A channel, addressed agent or voice session could not be reached.
failed	Enqueue, strategy, provider or delivery execution failed.
unknown	Voice submission is uncertain, or a custom strategy returned no outcome.

Read reason for a machine-readable explanation, error for structured failure information and unavailable_targets for unresolved addresses. event_id can be present on an unavailable/failed result: publication may succeed while agent solicitation or delivery fails. A failure after publication is not retried by the worker as a new text event.

Text destination availability comes from the committed delivery plan and its execution, so an agent leaving after processing does not turn a successful delivery into a failure. inbound.unavailable_targets is updated when its deferred handle finishes; replays do not re-evaluate the original destinations.

sent does not establish that the agent completed its turn. Direct Immediate and WaitForIdle calls wait for the existing text delivery handle when doing so cannot deadlock the room. Queued returns at publication, so its drain can accept requests from tools in an ongoing turn. Its inbound.delivery handle can be awaited separately. turn_complete reports known text completion, including failed completion; it is false for a replay or realtime injection.

result = await kit.deliver(
    "room", "Background result ready", addressed_to=["agent-a"],
    idempotency_key="external-event:42",
)
if result.inbound is not None and result.inbound.delivery is not None:
    turn = await result.inbound.delivery.wait()
    # Inspect turn.error, turn.response_events and turn.response_metadata.

The live inbound object is excluded from JSON serialization. Persisted queue items and hook metadata contain the serializable outcome snapshot. A queued call's delivery_item_id links it to the later AFTER_DELIVER observation; there is no promise that the initial queued result updates itself.

Addressing an external event

result = await kit.deliver(
    "conversation", "The background task finished.",
    channel_id="websocket",       # Transport/source attribution
    addressed_to=["agent-a"],     # Intelligence channel asked to act
    idempotency_key="task-result:42",
    metadata={"event_type": "task.finished", "task_id": "42"},
)

addressed_to=None retains the room's routing policy. [] stores and broadcasts the event without soliciting an intelligence channel. Unknown or unreadable addresses are reported as unavailable; another agent is never selected in their place. Valid addresses in a mixed list can still act. The stored event retains the whole address, and addressing does not change visibility, permissions or muting. A configured supervisor retains the RFC's supervisory exception.

Naming an intelligence channel in channel_id selects its room's transport for compatibility. Use addressed_to to choose which intelligence channel acts.

Delivering an instruction

Content the application produces for an agent, such as a background task's result or a scheduled nudge, is not something a participant said. Pass instruction=True to deliver it as the application's direction:

result = await kit.deliver(
    "conversation", "Tell the caller their order shipped.",
    addressed_to=["agent-a"],     # Required through the text pipeline
    instruction=True,
)

Through the text pipeline it is an INSTRUCTION event (see Directing an agent): the addressed agent takes it as its input for one turn and answers, and nothing is stored. It needs addressed_to and no idempotency_key; otherwise the outcome is blocked (instruction_unaddressed, instruction_not_idempotent). Sent to a realtime channel (channel_id= without addressed_to), it is injected with the system intent instead of user. The strategy, the delivery hooks and a delivery backend apply unchanged. The hooks see an event of type INSTRUCTION.

Text idempotency and retries

A key identifies one publication within a room, even if a later call supplies a different body or address. Concurrent calls with that key return the original event_id; replays set duplicate=True and do not run the agent again. The existing locked inbound pipeline owns this decision. Different keys, rooms, or calls without a key remain distinct. Empty keys are refused.

Keys last as long as the configured ConversationStore retains them. The memory store loses them on restart; SQLite and Postgres preserve committed keys across restart until the corresponding event/key is removed. Distributed deployments need a shared store and an appropriate shared lock manager. Custom stores must implement the same idempotency contract; an older store that cannot resolve a seen key may return a duplicate refusal without an event id.

A stored event does not prove its turn completed. If a process crashes after commit, replaying the key avoids a second trigger but does not resume the lost turn. Durable delivery-lane outboxes and crash recovery are separate capabilities.

Selecting a realtime session

result = await kit.deliver(
    "call-room", "Your result is ready.", channel_id="voice-main",
    session_id=session.id, strategy=WaitForIdle(buffer=1.0),
)

session_id requires an explicit realtime channel_id and cannot be combined with addressed_to. Without a session id, an explicitly selected realtime channel needs exactly one active session; otherwise the result is unavailable.

Idle strategies wait only for the selected realtime sessions. A busy session outside that selection does not delay delivery. Each pinned session is checked again immediately before injection; any sessions already reached remain listed if a later injection fails or becomes unavailable. Idle strategies pin that session before waiting and verify it again at injection. An ended/replaced session never selects a replacement. For deferred delivery, provide the original session id: resolution otherwise happens when the worker executes, not when the producer enqueues.

Unaddressed room-only calls retain realtime fanout. Supplying a non-null addressed_to uses the text pipeline even on a realtime transport; it never turns that intelligence address into direct session injection. Transport delivery and visibility still follow the room's ordinary broadcast rules.

Muted/read-only bindings receive silent context injection; unreadable bindings are refused. Anam cannot inject context silently and explicitly refuses these injections (voice_silent_injection_unsupported) instead of triggering speech.

Voice idempotency and uncertainty

Supply an idempotency_key to deduplicate proactive realtime injection:

result = await kit.deliver(
    "call-room", "Your result is ready.", channel_id="voice-main",
    session_id=session.id, idempotency_key="task-result:42",
)
session_result = result.session_outcomes.get(session.id)

The identity is (room_id, channel_id, session_id, idempotency_key). Concurrent calls and later replays reuse the recorded result with duplicate=True. The store atomically reserves the identity before calling the provider; a unique attempt token prevents a former owner from changing another attempt's result. Reusing the same identity for different effective text (after BEFORE_DELIVER) returns blocked with voice_idempotency_conflict. Use a stable key and body for each external event.

sent means the adapter's send operation completed; it does not prove audio was heard or a turn finished. When an exception, cancellation, abandoned reservation or lost persistence confirmation makes submission uncertain, the outcome is unknown. Replaying that key does not inject again. This can leave an announcement undelivered when a crash occurred before submission; the remote service and the store do not share an atomic transaction. Inspect an uncertain result instead of automatically issuing a new key, which would authorize a new delivery.

Only an explicit provider VoiceInjectionResult(status="not_sent", retryable=True) permits a retry after calling the provider: it guarantees no submission or pending submission exists. The worker dead-letters uncertain injections without retrying them. Gemini input queued locally during a tool call is reported as unknown (voice_provider_queued); its later flush does not update the receipt.

session_outcomes retains each selected session's outcome, including successes, duplicates, failures and uncertainty. session_ids lists successful submissions. A retry can submit to a session that safely refused while skipping successful or uncertain sessions. The aggregate status prioritizes uncertainty, failure, unavailability and refusal over success; inspect individual outcomes for partial delivery. Room-only fanout resolves active sessions on each execution. Supply both channel_id and the original session_id to keep retries tied to one call; its known receipt remains replayable after that session ends.

Receipts last until their room is deleted. The memory store loses them when its process ends. SQLite and Postgres retain them across restart; a Redis delivery queue alone does not make an in-memory ConversationStore durable. Multi-worker deployments need a shared durable store. Shared application locks allow callers to wait for an active attempt; the store's atomic claim prevents duplicate submission even without those locks. Reservations have no automatic expiry or reclamation that could reauthorize an uncertain submission.

Custom stores must implement get_voice_delivery, claim_voice_delivery and complete_voice_delivery with the same atomic semantics. Unsupported stores refuse keyed voice delivery before sending (voice_idempotency_unsupported). Custom realtime providers may still return None, but this reports unknown (voice_acceptance_unreported); return VoiceInjectionResult to report a known submission boundary. ON_REALTIME_TEXT_INJECTED fires only for a provider's explicit sent result. AFTER_DELIVER exposes all ordinary attempt outcomes.

Calls without a key and direct channel.inject_text() calls do not use these reservations. The feature does not resume interrupted agent turns.

Run the voice idempotency example with uv run python examples/voice_delivery_idempotency.py. It demonstrates concurrent delivery, a lost confirmation, a safe retry and a SQLite restart using mock audio providers, with no credentials or network service.

Queued grouping

Reuse a Queued instance to batch compatible requests. Rooms, transports, intelligence addresses, session targets, intents (message or instruction) and metadata must match. Requests with idempotency keys publish individually, retaining each key and outcome attribution. A request arriving during transmission is drained too. Cancelling the drain wakes waiting callers; it does not roll back committed events.

Hooks

from roomkit import HookTrigger, HookExecution, HookResult

@kit.hook(HookTrigger.BEFORE_DELIVER)
async def before_deliver(event, ctx):
    # This synchronous gate may return HookResult.block("policy") or modify text.
    return HookResult.allow()

@kit.hook(HookTrigger.AFTER_DELIVER, execution=HookExecution.ASYNC)
async def after_deliver(event, ctx):
    outcome = event.metadata["delivery_outcome"]
    logger.info("Delivery %s: %s", outcome["status"], outcome["reason"])

BEFORE_DELIVER runs synchronously; an explicit refusal stops delivery, while hook infrastructure failures follow the RFC's fail-open rule. AFTER_DELIVER observes refusals as well as successful and failed attempts, with effective rewritten content. Hook observation is best-effort. Neither hook changes the inbound pipeline ordering or its own broadcast hooks.

Hook	When	Payload
BEFORE_DELIVER	Before strategy executes	Address and key on the event, whose type is INSTRUCTION for an instruction delivery; channel_id, strategy, delivery_item_id, session_id in metadata
AFTER_DELIVER	After an execution attempt	Same fields, plus delivery_outcome and error (null on success/replay)

Integration with orchestration

The Supervisor strategy uses kit.deliver() internally:

• Sync mode (async_delivery=False): results returned inline, no delivery needed
• Async mode (async_delivery=True): workers run in background, and their results are handed back to the supervisor with kit.deliver(..., instruction=True), each worker's output bounded, when the strategy allows

from roomkit import RoomKit, Supervisor, WaitForIdle

kit = RoomKit(
    delivery_strategy=WaitForIdle(buffer=3.0),
    orchestration=Supervisor(
        supervisor=coordinator,
        workers=[analyst_1, analyst_2],
        strategy="parallel",
        auto_delegate=True,
        async_delivery=True,
    ),
)

See the Orchestration guide for full Supervisor documentation.

Persistent delivery backends

By default, kit.deliver() executes in-process — if the process crashes, pending deliveries are lost. For production deployments, configure a delivery backend to decouple enqueue from execution:

from roomkit import RoomKit, InMemoryDeliveryBackend, WaitForIdle

# In-memory backend (single process, no persistence)
kit = RoomKit(
    delivery_strategy=WaitForIdle(buffer=3.0),
    delivery_backend=InMemoryDeliveryBackend(),
)

async with kit:
    await kit.deliver("room", content="Background result ready.")
    # Item is enqueued → worker loop executes delivery asynchronously

How it works

When a delivery_backend is configured:

1. kit.deliver() serializes the request into a DeliveryItem and calls backend.enqueue()
2. A background worker loop calls backend.dequeue() to claim items
3. The worker deserializes the strategy and executes strategy.deliver()
4. Consume sent/refused items with ack(); retry transient unavailable/failed outcomes with nack(); dead-letter non-retryable failures and uncertain voice submissions.

kit.deliver()
  → serialize strategy + content → DeliveryItem
  → backend.enqueue(item)
  → return (non-blocking)

Worker loop (background):
  → backend.dequeue() → claim items
  → BEFORE_DELIVER hook
  → strategy.deliver(ctx)
  → AFTER_DELIVER hook
  → backend.ack() or backend.nack()

Redis backend

For multi-worker deployments, use RedisDeliveryBackend with Redis Streams:

from roomkit import RoomKit, WaitForIdle
from roomkit.delivery import RedisDeliveryBackend

kit = RoomKit(
    delivery_strategy=WaitForIdle(buffer=3.0),
    delivery_backend=RedisDeliveryBackend("redis://localhost:6379"),
)

Requires pip install roomkit[redis].

Features:

• Consumer groups distribute items across workers automatically
• At-least-once delivery via Redis Streams PEL (Pending Entries List)
• Bounded dead-letter stream for items that exhaust retries
• Delayed retries held in a sorted set on the server's clock
• Injected client support for connection pooling

import redis.asyncio as redis

pool = redis.ConnectionPool.from_url("redis://localhost:6379")
client = redis.Redis(connection_pool=pool)

backend = RedisDeliveryBackend(
    client=client,
    stream_prefix="myapp:delivery",
    group_name="myapp-workers",
    max_dead_letter_size=10_000,
)

Available backends

Backend	Persistence	Multi-worker	Install
InMemoryDeliveryBackend	No	No	Built-in
RedisDeliveryBackend	Yes	Yes	roomkit[redis]

Retry and dead-letter

A retryable failure is retried up to max_retries times after the first attempt (default 3, so four attempts at most), as a binding's RetryPolicy counts them. Each retry waits first, so a destination that is briefly unavailable (a voice session reconnecting) has time to come back: min(retry_delay * 2 ** (n - 1), max_retry_delay) seconds before retry n, 1 s and 60 s by default. Both are set on the backend; retry_delay=0 retries at once:

backend = InMemoryDeliveryBackend(retry_delay=2.0, max_retry_delay=30.0)

RedisDeliveryBackend keeps a waiting retry in a sorted set ({stream_prefix}:delayed), timed by the Redis server's clock, and the next worker to dequeue puts it back on the stream once due; no two workers take the same retry. get_queue_depth() counts the retries waiting out their delay.

Exhausted or non-retryable failures move to the dead-letter queue with item.outcome, address, key and session target preserved. Explicit hook refusals are consumed without retry and retain a blocked outcome:

# Inspect dead-lettered items, the most recent first in every backend
dead = await backend.get_dead_letter_items(limit=50)
for item in dead:
    print(f"{item.id}: {item.error}")

# Check queue depth
depth = await backend.get_queue_depth()

Backward compatibility

Calls without the new options remain valid. Without a backend, execution stays in-process; callers that ignore the returned outcome can continue doing so. Custom strategies returning None execute but report unknown.

Run uv run python examples/external_event_delivery.py for a network-free example with two agents and a replayed external event.