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, 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)

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

setup_realtime_delegation

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

Wire delegation into a RealtimeVoiceChannel's tool chain.

Injects the delegate tool dict into channel._tools and wraps channel._tool_handler to intercept delegate_task calls. Room ID is resolved from the current voice session via get_current_voice_session() + channel.session_rooms.

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

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 and how content is delivered to a channel.

deliver abstractmethod async

deliver(ctx)

Deliver the content according to this strategy.

Immediate

Bases: DeliveryStrategy

Send now. May interrupt ongoing TTS playback.

WaitForIdle

WaitForIdle(buffer=1.0, playback_timeout=15.0)

Bases: DeliveryStrategy

Wait for TTS/speech to finish, then send.

Parameters:

Name	Type	Description	Default
buffer	float	Extra seconds to wait after playback ends (default 1.0).	1.0
playback_timeout	float	Max seconds to wait for playback (default 15.0).	15.0

Queued

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

Bases: DeliveryStrategy

Add to queue, deliver at next idle window.

Multiple deliveries are batched into a single message.

Parameters:

Name	Type	Description	Default
buffer	float	Extra seconds to wait after playback ends (default 1.0).	1.0
playback_timeout	float	Max seconds to wait for playback (default 15.0).	15.0
separator	str	Text between batched items (default newline).	'\n\n'

DeliveryContext dataclass

DeliveryContext(kit, room_id, content, channel_id=None, metadata=None)

Context passed to delivery strategies.

find_transport_channel_id async

find_transport_channel_id()

Find the best transport channel in the room.

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

resolve_channel_id async

resolve_channel_id()

Resolve the target channel — explicit or auto-detected.