Delegation

Background task delegation via child rooms. See the Agent Delegation guide for usage examples.

Task Runner

TaskRunner

Bases: ABC

ABC for executing delegated tasks in the background.

Follows the same pluggable-backend pattern as ConversationStore and RoomLockManager.

A task ends exactly once, through on_complete, whatever ends it (RFC §23.1, §23.3): run to its end, it ends completed or failed; cancelled (:meth:cancel, :meth:close, even before it ran), it ends with a cancelled result (:func:~roomkit.tasks.models.cancelled_task_fields). The framework closes the task's span, fires ON_TASK_COMPLETED and hands the result back from that call.

The worker is given task.description, what it reads (the brief, else the task, with the room's conversation when the delegation hands it over), not task.task, the label the task is named by (RFC §23.3 step 4).

submit abstractmethod async

submit(kit, task, *, context=None, on_complete=None)

Start background execution of task.

Parameters:

Name	Type	Description	Default
kit	RoomKit	The RoomKit instance (used to interact with rooms/channels).	required
task	DelegatedTask	The delegated task handle.	required
context	dict[str, Any] | None	Optional context passed to the agent.	None
on_complete	OnCompleteCallback | None	Callback invoked when the task finishes.	None

cancel abstractmethod async

cancel(task_id)

Cancel a task and end it cancelled through its on_complete, run to its end even if this call is cancelled. A task that already ran to its end ends as it stands. Returns True if the task was found and cancelled.

close abstractmethod async

close()

Shutdown the runner: every task it holds ends cancelled, one submitted meanwhile too.

InMemoryTaskRunner

InMemoryTaskRunner()

Bases: TaskRunner

Default task runner — executes tasks as asyncio.Task instances.

Task Models

DelegatedTask

DelegatedTask(*, id, child_room_id, parent_room_id, agent_id, task, brief=None, description=None, hand_back='relay')

Handle for a running delegated task.

Not a Pydantic model — holds mutable state and an asyncio.Event for callers that want to block until the task completes.

task instance-attribute

task = task

What was asked: the label a hand-back, the StatusBus and a turn's notes name the task by (RFC §23.3 step 4).

brief instance-attribute

brief = brief

What the worker is told, when it says more than the task.

description instance-attribute

description = description or brief or task

What the worker reads: the brief, else the task, with the room's conversation when the delegation hands it over.

hand_back instance-attribute

hand_back = hand_back

How the result reaches the room: relay (the notified agent says it in a turn of its own) or post (posted as it is, as its message).

wait async

wait(timeout=None)

Block until the task completes or timeout seconds elapse.

Raises:

Type	Description
TimeoutError	If timeout is exceeded.
RuntimeError	If the task finished without a result.

cancel

cancel()

Mark the task as cancelled and unblock its waiters, and nothing else: the work goes on, and no hook or callback hears of it. To cancel the work, as every task ends (RFC §23.3), use the task runner's cancel.

DelegatedTaskResult

Bases: BaseModel

Result of a completed delegated task.

exception class-attribute instance-attribute

exception = Field(default=None, exclude=True)

The failure itself, its type kept (what the worker's turn raised), for a caller that hands it on; held in memory, never serialized (RFC §23.3 step 6).

Tool Integration

DELEGATE_TOOL module-attribute

DELEGATE_TOOL = AITool(name=DELEGATE_TOOL_NAME, description='Delegate a task to a background agent. The agent works in its own room and you can continue the current conversation. Use when: a task needs a specialist, the user wants something done in the background, or you need to run work in parallel.', parameters={'type': 'object', 'properties': {'agent': {'type': 'string', 'description': 'Target agent ID for the delegated task'}, 'task': {'type': 'string', 'description': 'Clear description of what the agent should do'}, 'context': {'type': 'object', 'description': 'Optional context to pass to the agent (e.g. email, repo URL)'}, 'share_channels': {'type': 'array', 'items': {'type': 'string'}, 'description': 'Channel IDs from the current room to share with the background agent'}}, 'required': ['agent', 'task']})

build_delegate_tool

build_delegate_tool(targets)

Build a delegate tool with constrained agent enum.

Parameters:

Name	Type	Description	Default
targets	list[tuple[str, str | None]]	List of (agent_id, description_or_none) pairs. When empty, returns the generic :data:DELEGATE_TOOL.	required

Returns:

Name	Type	Description
An	AITool	class:AITool whose agent parameter has an enum
	AITool	restricting the AI to only the listed agent IDs.

DelegateHandler

DelegateHandler(kit, *, notify=None, default_share_channels=None, delivery_strategy=None, cache=None, serialize_per_room=False)

Processes delegate_task tool calls by calling kit.delegate().

Supports optional dedup via :class:CompletedTaskCache and per-room concurrency control via asyncio locks.

Parameters:

Name	Type	Description	Default
kit	RoomKit	The RoomKit instance.	required
notify	str | None	Channel ID to notify on completion (defaults to calling agent).	None
default_share_channels	list[str] | None	Channels shared with background agents by default.	None
delivery_strategy	DeliveryStrategy | None	How to deliver results back proactively.	None
cache	CompletedTaskCache | None	Optional completed-task cache for dedup.	None
serialize_per_room	bool	If True, only one delegation per room at a time.	False

handle async

handle(room_id, calling_agent_id, arguments)

Process a delegate_task tool call.

setup_delegation

setup_delegation(channel, handler, *, tool=None)

Serve the delegate tool on an AIChannel.

Declares delegate_task in the channel's turns and serves its calls with handler. Same pattern as setup_handoff(): the task is delegated from the room of the call, read from the tool call context (RFC §23.4), so one agent serving several rooms delegates from each.

Parameters:

Name	Type	Description	Default
channel	AIChannel	The AI channel to wire delegation into.	required
handler	DelegateHandler	The delegate handler that processes tool calls.	required
tool	AITool | None	Optional custom delegate tool (e.g. from :func:build_delegate_tool).	None

Raises:

Type	Description
RuntimeError	Delegation is already set up on the channel.

setup_realtime_delegation

setup_realtime_delegation(channel, handler, *, tool=None)

Serve the delegate tool on a RealtimeVoiceChannel.

Declares delegate_task in the channel's sessions and serves its calls with handler, delegating from the room of the call: the session's room, which the channel installs as the tool call context (RFC §21.4, §23.4).

Parameters:

Name	Type	Description	Default
channel	RealtimeVoiceChannel	The realtime voice channel to wire delegation into.	required
handler	DelegateHandler	The delegate handler that processes tool calls.	required
tool	AITool | None	Optional custom delegate tool (e.g. from :func:build_delegate_tool).	None

Raises:

Type	Description
RuntimeError	Delegation is already set up on the channel.

Background Work

RoomKit.setup_background(agent, worker=None, hand_back="relay") gives an agent the work_in_background tool (BACKGROUND_TOOL, from roomkit.tasks), and RoomKit.start_task(room_id, agent_id, task, hand_back=None) starts the same task from the host. See Long work from an agent.

BackgroundSetup dataclass

BackgroundSetup(worker=None, hand_back='relay')

How an agent's background tasks run: on worker (the agent itself when None), their result relayed or posted (hand_back).

BackgroundMixin

setup_background and start_task: an agent's long work runs as a task, off its turn.

setup_background

setup_background(agent, *, worker=None, hand_back='relay')

Give agent the work_in_background tool (RFC §23.4): it starts long work as a task and ends its turn, the work running apart.

Parameters:

Name	Type	Description	Default
agent	AIChannel	The AI channel given the tool, in every room it serves.	required
worker	str | None	The channel that does its tasks (another agent, an external agent over ACP); the agent itself when None.	None
hand_back	str	How a task's result reaches the room: "relay", the agent says it in a turn of its own, or "post", it is posted as it is, as the agent's message.	'relay'

Raises:

Type	Description
ValueError	An unknown hand_back.
RuntimeError	The tool is already set up on the agent.

start_task async

start_task(room_id, agent_id, task, *, hand_back=None)

Start task for agent_id, an agent of room_id, as its work_in_background call would (RFC §23.4): on its worker, handed the room's last messages, its result relayed or posted back to it. The host's door, and a person's through what the host gives them; an agent with no background set up does the work itself and has its result relayed.

Parameters:

Name	Type	Description	Default
room_id	str	The room the task is for.	required
agent_id	str	The agent the task is for: an intelligence channel attached to the room, an external agent included.	required
task	str	What to do and report.	required
hand_back	str | None	"relay" or "post"; the agent's setup's when None.	None

Raises:

Type	Description
ValueError	agent_id is no agent of the room, task is blank, or hand_back is unknown.

Following Tasks

TaskStatusTool

TaskStatusTool(kit, *, window=500, limit=10)

task_status: the tasks of the room of the call, read from the framework's StatusBus (RFC §23.4).

A tool an agent is given on its own — AIChannel(tools=[TaskStatusTool(kit)]) — independently of the delegation helpers. It answers with the tasks posted for the room of the call (delegations, orchestration workers): those running and those ended, the latest state of each, with its worker, the task, and its result when it completed; task_id narrows the answer to one task. It reads the bus for that room only: one room's tasks never show in another.

Parameters:

Name	Type	Description	Default
kit	RoomKit	The framework whose status_bus the tool reads.	required
window	int	How many of the bus's latest entries are read.	500
limit	int	How many tasks, the latest, an answer lists at most.	10

CancelTaskTool

CancelTaskTool(kit, *, window=500)

cancel_task: cancel one task of the room of the call (RFC §23.4).

A tool an agent is given on its own, as :class:TaskStatusTool — AIChannel(tools=[TaskStatusTool(kit), CancelTaskTool(kit)]) — so that a request the person changed or dropped stops its work. It reaches only the tasks the StatusBus lists for the room of the call: never another room's, nor a strategy's worker run, which that strategy follows itself. A running task ends cancelled; the agent that cancelled it is not handed the cancellation back, the tool's answer having told it.

Parameters:

Name	Type	Description	Default
kit	RoomKit	The framework whose tasks the tool cancels.	required
window	int	How many of the bus's latest entries are read to find the task.	500

room_tasks

room_tasks(entries)

One line per task, from its entries in posting order: the task from its pending entry, its state and result from its latest terminal one, its progress from its latest info one, which never ends it.

An orchestration worker run posts its pending entry before its task has an id: the run's terminal entry, which names the task, joins it.

post_task_progress async

post_task_progress(kit, detail)

From a tool a worker calls in its task's child room: say how far the task got ("12/30 s", "3 sources of 5"), for the room that delegated it (RFC §23.3).

Posts an info entry under the worker, action task, with the task's metadata; it does not end the task. Returns whether it posted: a call outside a task's child room posts nothing.

The tools' names, as a model calls them: TASK_STATUS_TOOL ("task_status") and CANCEL_TASK_TOOL ("cancel_task"), from roomkit.tasks.

Delegation State Tracking

CompletedTaskCache

CompletedTaskCache(ttl_seconds=300.0)

In-memory TTL cache for completed delegation results.

Parameters:

Name	Type	Description	Default
ttl_seconds	float	Seconds before a cached result expires. Defaults to 300 (5 min).	300.0

put

put(room_id, agent_id, task, result)

Store a completed task result.

get

get(room_id, agent_id, task)

Return cached result if present and not expired, else None.

recent_context

recent_context(room_id, limit=5)

Return summaries of recent tasks for a room (for context injection).

Returns task descriptions from newest to oldest, up to limit. Expired entries are skipped.

clear

clear()

Remove all entries.

Delivery Strategies

DeliveryStrategy

Bases: ABC

Controls when content is delivered; report its actual outcome.

deliver abstractmethod async

deliver(ctx)

Deliver the request. A custom strategy may leave its outcome unknown.

Immediate

Bases: DeliveryStrategy

Send now. On a voice session, it plays once the playback under way has been sent whole: a response never cuts the agent off (RFC §12.2).

WaitForIdle

WaitForIdle(buffer=1.0, playback_timeout=15.0)

Bases: DeliveryStrategy

Wait for voice idle, then send; a timeout falls back to immediate.

Queued

Queued(buffer=1.0, playback_timeout=15.0, separator='\n\n')

Bases: DeliveryStrategy

Batch compatible unkeyed requests at idle; preserve each caller's outcome.

Keyed requests publish individually, including duplicate keys. One caller owns the drain; followers wait for their own request to be processed.

DeliveryContext dataclass

DeliveryContext(kit, room_id, content, channel_id=None, metadata=None, addressed_to=None, idempotency_key=None, session_id=None, chain_depth=0, instruction=False, _voice_sessions=None, _voice_channel=None, _wait_for_turn=True)

One request, including its destination pinned before an idle wait.

find_transport_channel_id async

find_transport_channel_id()

Prefer voice, then the first other transport bound to the room.

Voice is a voice channel type or any channel whose realtime model is plugged in at delivery time (a conference with one, :func:hosts_realtime_model). An intelligence channel attached as a transport is not one: an instruction delivered through it would re-enter the channel it is for.

resolve_channel_id async

resolve_channel_id()

Resolve an explicit destination or auto-detect its transport.