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.

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 running task. Returns True if found and cancelled.

close abstractmethod async

close()

Shutdown the runner, cancelling all in-flight tasks.

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)

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.

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

DelegatedTaskResult

Bases: BaseModel

Result of a completed delegated task.

Tool Integration

DELEGATE_TOOL module-attribute

DELEGATE_TOOL = AITool(name='delegate_task', 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)

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

handle async

handle(room_id, calling_agent_id, arguments)

Process a delegate_task tool call.

setup_delegation

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

Wire delegation into an AIChannel's tool chain.

Injects the delegate tool and wraps the tool handler to intercept delegate_task calls. Same pattern as setup_handoff().

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

Task Delivery Strategies

BackgroundTaskDeliveryStrategy

Bases: ABC

Controls proactive delivery of background task results.

After a delegated task completes, the framework:

1. Injects the result into the notified agent's system prompt (always).
2. Fires the ON_TASK_COMPLETED hook (always).
3. Calls strategy.deliver(ctx) (this method).

The strategy decides whether/when to send a synthetic inbound message that triggers the agent to respond about the result.

deliver abstractmethod async

deliver(ctx)

Deliver the task result proactively.

Parameters:

Name	Type	Description	Default
ctx	TaskDeliveryContext	Delivery context with access to the framework, result, and helper methods.	required

ImmediateDelivery

ImmediateDelivery(prompt=None)

Bases: BackgroundTaskDeliveryStrategy

Deliver immediately via a synthetic inbound message.

Sends process_inbound() right away, which triggers the agent to generate a response about the completed task. For voice this may interrupt ongoing TTS playback.

WaitForIdleDelivery

WaitForIdleDelivery(prompt=None, playback_timeout=15.0, interrupt_playback=False)

Bases: BackgroundTaskDeliveryStrategy

Wait for TTS playback to finish, then deliver.

For voice channels, waits until VoiceChannel.wait_playback_done() returns before sending the synthetic inbound message. This avoids interrupting the agent mid-sentence.

For non-voice channels (text, WebSocket, etc.), delivers immediately.

ContextOnlyDelivery

Bases: BackgroundTaskDeliveryStrategy

No proactive delivery — wait for the user's next message.

The result is available in the agent's system prompt and will be referenced when the user next speaks. This is the default behavior and maintains backward compatibility.

TaskDeliveryContext dataclass

TaskDeliveryContext(kit, result, notify_channel_id)

Context passed to delivery strategies after task completion.

room_id property

room_id

Parent room that should receive the delivery.

find_transport_channel_id async

find_transport_channel_id()

Find the best transport channel in the parent room.

Prefers voice channels (most latency-sensitive), then falls back to any other transport channel.

Returns:

Type	Description
str | None	Channel ID of a transport channel, or None if none found.