Orchestration

Multi-agent orchestration for complex conversational workflows. See the Multi-Agent Orchestration guide for usage examples.

Strategies

Orchestration

Bases: ABC

Abstract base for orchestration strategies.

Orchestration strategies compose existing primitives (ConversationPipeline, ConversationRouter, HandoffHandler) into declarative patterns that can be passed to RoomKit or create_room.

Subclasses must implement:

• :meth:agents — which agents participate in the room.
• :meth:install — wire hooks, tools, and state into a room.

agents abstractmethod

agents()

Return agents to register and attach to the room.

The framework calls this to determine which agents should be registered on the kit and attached to the room at creation time. An external agent over ACP is one where the strategy takes it (a discussion does).

install abstractmethod async

install(kit, room_id)

Wire hooks, tools, and state into the room.

Called after agents are registered and attached. Implementations should install room-scoped hooks, set up handoff tools, and initialise conversation state. Install through :meth:RoomKit.install_strategy (or create_room(orchestration=...)), which claims the room for one strategy and records what the install adds, so :meth:RoomKit.uninstall_strategy can take it back.

uninstall async

uninstall(kit, room_id)

Undo what :meth:install set up that the kit does not record.

:meth:RoomKit.uninstall_strategy calls it first, then removes what it recorded of the install: the room hooks, the tools and turn runners set up on the agents for the room, the agents the install attached, and the room metadata the install wrote (RFC §19.7). A strategy with state of its own elsewhere (a discussion's turns) stops it here. Default: nothing.

Pipeline

Pipeline(agents, routing=None, *, supervisor_id=None, voice_channel_id=None, greet_on_handoff=False, greeting_prompt=None, farewell_prompt=None)

Bases: Orchestration

Linear pipeline orchestration strategy.

Agents are chained in order: the first agent is the entry point, each subsequent agent is reachable via handoff from its predecessor.

Example::

kit = RoomKit(
    orchestration=Pipeline(
        agents=[triage, support, billing],
    ),
)
room = await kit.create_room()

Initialise the pipeline strategy.

Parameters:

Name	Type	Description	Default
agents	list[Agent]	Ordered list of agents. First agent is the entry point. Stages chain linearly (each agent hands off to the next).	required
routing	dict[str, list[str]] | None	Optional keyword hints injected into agent system prompts (not runtime rules). Keys are agent channel IDs, values are keyword lists describing when to route there.	None
supervisor_id	str | None	Optional supervisor agent ID that receives all events for monitoring.	None
voice_channel_id	str | None	Voice channel ID for voice-aware handoffs.	None
greet_on_handoff	bool	Whether agents greet on handoff.	False
greeting_prompt	str | None	Custom greeting prompt template.	None
farewell_prompt	str | None	Custom farewell prompt template.	None

agents

agents()

Return all pipeline agents.

install async

install(kit, room_id)

Wire pipeline routing and handoff into the room.

Swarm

Swarm(agents, entry=None)

Bases: Orchestration

Swarm orchestration strategy.

Every agent can hand off to every other agent with no phase constraints. Routing uses sticky agent affinity — once an agent is active, it keeps handling until it hands off.

Example::

kit = RoomKit(
    orchestration=Swarm(
        agents=[sales, support, billing],
        entry="sales",
    ),
)
room = await kit.create_room()

Initialise the swarm strategy.

Parameters:

Name	Type	Description	Default
agents	list[Agent]	All participating agents.	required
entry	str | None	Channel ID of the entry-point agent. Defaults to the first agent's channel ID.	None

agents

agents()

Return all swarm agents.

install async

install(kit, room_id)

Wire swarm routing and bidirectional handoff into the room.

Supervisor

Supervisor(supervisor, workers, *, strategy=None, auto_delegate=False, async_delivery=False, refine_task=True, refine_instruction=None, delegation_message="I'm dispatching my team to work on this.", wait_for_result=True, share_channels=None, task_timeout=_DEFAULT_TASK_TIMEOUT_SECONDS, max_revisions=_DEFAULT_MAX_REVISIONS)

Bases: _AutoDelegateInstallMixin, _StrategyToolMixin, _PerWorkerToolMixin, Orchestration

Supervisor orchestration strategy.

The supervisor handles all user interaction. Workers are registered on the kit (so delegate() can find them) but are NOT attached to the parent room — they run in child rooms.

Examples::

# Framework-driven: auto-delegate with task refinement
Supervisor(
    supervisor=coordinator,
    workers=[researcher, writer],
    strategy="sequential",
    auto_delegate=True,
)

# Framework-driven: workers get raw user message
Supervisor(
    supervisor=coordinator,
    workers=[technical, business],
    strategy="parallel",
    auto_delegate=True,
    refine_task=False,
)

# Tool-based: AI decides when to delegate
Supervisor(
    supervisor=coordinator,
    workers=[researcher, writer],
    strategy="sequential",
)

# Manual: per-worker tools, AI decides everything
Supervisor(
    supervisor=coordinator,
    workers=[researcher, writer],
)

Initialise the supervisor strategy.

Parameters:

Name	Type	Description	Default
supervisor	Agent	The agent that handles user interaction and delegates tasks.	required
workers	list[Agent]	Agents that run delegated tasks in child rooms.	required
strategy	WorkerStrategy | str | None	Deterministic execution pattern for workers. "sequential": workers run in order, each receiving the previous worker's output. "parallel": all workers run concurrently on the same task. None (default): per-worker delegate_to_<id> tools are injected and the AI decides when to call them.	None
auto_delegate	bool	If True, the framework triggers workers automatically — no tool needed. For sync channels (CLI), blocks until results are ready. For async channels (voice), runs in background. Requires strategy.	False
async_delivery	bool	If True, worker delegation returns immediately and results are handed back, when they complete, to the channel whose call dispatched them (in its session, on a voice channel). This keeps the supervisor's tool-loop clock bounded by its own reasoning time rather than aggregated worker wall-clock time. Applies to: strategy-tool mode (strategy set): the delegate_workers tool fires background workers and returns {"status": "dispatched", ...} immediately; its reply points the supervisor at task_status (:class:~roomkit.tasks.status.TaskStatusTool) only when the supervisor was given it. voice auto_delegate mode: injects a delegate_workers tool on the voice channel; supervisor is not attached (voice handles UI). Lifecycle events (pending / completed / failed) are posted to kit.status_bus for every worker regardless of mode. If False (default), delegation blocks until results are ready (sync mode).	False
refine_task	bool	Controls whether the framework extracts a clean topic from the user's message before sending to workers.	True
refine_instruction	str | None	Custom instruction for topic extraction. Overrides the default.	None
delegation_message	str | None	Message injected into the conversation when workers are dispatched (async mode only). Set to None to disable. Default: "I'm dispatching my team to work on this."	"I'm dispatching my team to work on this."
wait_for_result	bool	When strategy is None, controls whether delegation runs inline (True, default) or in the background (False). Ignored when strategy is set or auto_delegate is True.	True
share_channels	list[str] | None	Channel IDs from the parent room to share with every child room created during delegation. For example, passing ["system", "ws-status"] attaches those channels to each worker's child room so the worker can emit events visible on those channels.	None
task_timeout	float	Per-worker delegation budget in seconds (default 120). Each delegated task is bounded individually so the chain is governed per-task rather than by a single global timeout that can never be coherent across team sizes. A worker that exceeds it is recorded as failed and the run continues.	_DEFAULT_TASK_TIMEOUT_SECONDS
max_revisions	int	In supervised sequential mode, the maximum number of supervisor→worker rework round-trips per step (default 3). The supervisor validates each worker's output; if unacceptable it sends feedback for a rework, up to this many times. On exhaustion the step is delivered flagged as unvalidated rather than looping.	_DEFAULT_MAX_REVISIONS

agents

agents()

Return agents to attach to the room.

async_delivery removes the supervisor from the room ONLY in voice auto_delegate mode, where a RealtimeVoiceChannel handles user interaction directly. In strategy-tool or per-worker-tool modes the supervisor still drives the conversation via its tool loop; background dispatch only affects worker delegation, not user routing.

install async

install(kit, room_id)

Wire supervisor routing and delegation tools.

Loop

Loop(agent, reviewers=None, reviewer=None, max_iterations=3, *, strategy=None, async_delivery=False)

Bases: Orchestration

Loop orchestration strategy.

The producing agent generates output, then reviewers evaluate it. If all reviewers approve, the loop ends. Otherwise, feedback is routed back to the producer for revision.

Examples::

# Single reviewer
Loop(agent=writer, reviewers=[editor], max_iterations=3)

# Multiple reviewers — sequential (chained)
Loop(
    agent=coder,
    reviewers=[security, perf, style],
    strategy="sequential",
)

# Multiple reviewers — parallel (fan-out)
Loop(
    agent=coder,
    reviewers=[security, perf, style],
    strategy="parallel",
)

# Voice — async delivery
Loop(
    agent=writer,
    reviewers=[editor],
    async_delivery=True,
)

Initialise the loop strategy.

Parameters:

Name	Type	Description	Default
agent	Agent	The producing agent.	required
reviewers	list[Agent] | None	List of reviewing agents. For multiple reviewers, use strategy to control execution order.	None
reviewer	Agent | None	Single reviewer (convenience, same as reviewers=[reviewer]).	None
max_iterations	int	Maximum number of produce-review cycles.	3
strategy	WorkerStrategy | str | None	How reviewers execute when there are multiple: "sequential": reviewers chain — each sees the previous reviewer's feedback. "parallel": reviewers fan-out — all review independently, feedback combined. None (default): sequential for multiple reviewers, single reviewer doesn't need a strategy.	None
async_delivery	bool	If True, the loop runs in the background and its outcome is handed back to the voice channel that started it when ready, as a background delegation's result is. The conversation continues uninterrupted.	False

agents

agents()

Return the producer — it presents results to the user.

install async

install(kit, room_id)

Wire the framework-driven loop.

Discussion

Discussion(agents, *, people=None, addressed_only=False, everyone=None, max_turns=None, max_depth=None, done=None, silent_token='(silent)', dispatch=None, dispatch_timeout=5.0)

Bases: Orchestration

Several agents and people hold one conversation; one agent speaks at a time.

Any agent may address any other by @channel_id, a person may address any agent, and who speaks next follows from who was addressed: the strategy queues the agents each message asks for and gives them their turns one at a time, each turn reading the room as it is when it starts. The room is the discussion's: its intelligence channels are the discussion's agents and no other, and no router or other strategy shares it. Text only in this version.

Example::

kit = RoomKit()
room = await kit.create_room(
    orchestration=Discussion(agents=[investigator, dev, sre]),
)
# "@dev can you check the deploy?" asks dev; dev's "@sre please
# roll back" asks sre once dev's turn has ended.

Parameters:

Name	Type	Description	Default
agents	Sequence[AIChannel | ACPChannel]	The agents of the discussion: AI channels, Agents among them, with distinct channel ids.	required
people	Sequence[str] | None	The names agents address people by. None: the names of the room's participants that are neither agents nor bots.	None
addressed_only	bool	A person's message asks only the agents it names; one that names nobody asks no agent.	False
everyone	Sequence[str] | None	The agents a person's message that answers no one asks, in this order (the one that should open first). None: every agent, in agents order.	None
max_turns	int | None	Turns given in the room's lifetime, after which the discussion is over. None: no bound but max_depth.	None
max_depth	int | None	The depth limit of the turns it gives, in place of the kit's max_chain_depth. None: the kit's.	None
done	DoneFn | None	done(room_id), checked before each turn: once it holds, the discussion is over.	None
silent_token	str	The answer of an agent with nothing to add.	'(silent)'
dispatch	DispatchPolicy | None	Decides which agents take a person's message that names none and answers none, in place of asking everyone (rule 18): ClassifierDispatchPolicy asks a classifier who it is for. Not with addressed_only, where such a message asks no agent.	None
dispatch_timeout	float	Seconds a decision may take; past them, or when the policy fails, the message asks everyone it reaches.	5.0

agents

agents()

The agents of the discussion.

install async

install(kit, room_id)

Give the room to the discussion: its agents attached, its stored queue read, its turns started.

Raises:

Type	Description
ValueError	the room binds another intelligence channel or a voice or realtime channel, an agent thinks while it listens, or a router, another strategy or a discussion is installed.

uninstall async

uninstall(kit, room_id)

Give the room back to its agent_response_policy: the turns stopped, the queue dropped.

Discussion's speak queue

SpeakQueue dataclass

SpeakQueue(speaking, queue, listening, asked, waiting, over)

A room's speak queue as the host reads it (RFC §19.7.5).

speaking instance-attribute

speaking

The agent whose turn runs now, or None between turns.

queue instance-attribute

queue

The agents owed a turn, in the order they get it.

listening instance-attribute

listening

The agents that only listen (listen_only).

asked instance-attribute

asked

(agent, person): an agent that asked a person and waits for the answer.

waiting instance-attribute

waiting

Whether the discussion waits for a person's message.

over instance-attribute

over

Whether the discussion is over: it gives no further turn.

SpeakQueueChange

Bases: StrEnum

What changed in a room's speak queue, as ON_SPEAK_QUEUE reports it.

SpeakQueueEvent dataclass

SpeakQueueEvent(room_id, queue, change, channel_ids=(), event_id=None)

Carried to ON_SPEAK_QUEUE hooks: one change of a room's speak queue.

queue instance-attribute

queue

The queue as it is once changed.

channel_ids class-attribute instance-attribute

channel_ids = ()

The agents the change is about (queued, given a turn, listening...).

event_id class-attribute instance-attribute

event_id = None

The event behind the change, when one is (the message that queued, the event a turn answers).

Discussion's dispatch policy

DispatchPolicy

Bases: ABC

Decides which agents of a discussion take a person's unaddressed message.

A policy that raises, or does not decide within the discussion's dispatch_timeout, does not silence the room: the message asks what it asks with no policy (the agents that asked its author, else every candidate), and the decision reported carries the reason fallback. A message that names an agent never reaches a policy: a name is the person's word.

decide abstractmethod async

decide(turn)

The decision on turn: agents among its candidates, in order.

DispatchTurn dataclass

DispatchTurn(room_id, event, recent=(), speakers=dict(), candidates=(), asked=())

What a policy judges: the message, the conversation before it, the candidates.

event instance-attribute

event

The person's message.

recent class-attribute instance-attribute

recent = ()

The conversation before it, oldest first: the room's messages a candidate may read, without tool records.

speakers class-attribute instance-attribute

speakers = field(default_factory=dict)

Who said event and each of recent, by event id: a person by the label the transcript gives them (RFC §6.4), an agent as @channel_id.

candidates class-attribute instance-attribute

candidates = ()

The agents the policy may pick: those that asked the message's author, then everyone in its order; only those the message reaches, less those that only listen.

asked class-attribute instance-attribute

asked = ()

The candidates that asked the message's author a question and wait for the answer (RFC §19.7.5 rule 10): the message may answer them, or not.

DispatchCandidate dataclass

DispatchCandidate(channel_id, name=None, role=None, description=None)

An agent a policy may pick, with the identity the host gave it.

description class-attribute instance-attribute

description = None

What the agent does, as a turn's notes tell the others (rule 14).

DispatchDecision dataclass

DispatchDecision(agents=(), reason='', judgments=dict())

Which agents take a message, in the order they answer, and why.

agents class-attribute instance-attribute

agents = ()

Their channel ids; empty, no agent takes the message.

reason class-attribute instance-attribute

reason = ''

Why, in a few words, for logs and ON_DISPATCH_DECISION.

judgments class-attribute instance-attribute

judgments = field(default_factory=dict)

What the policy weighed, by name: what makes a decision measurable.

DispatchDecisionEvent dataclass

DispatchDecisionEvent(room_id, event, candidates, decision, duration_ms=0)

Carried to ON_DISPATCH_DECISION hooks: one decision on one message.

event instance-attribute

event

The person's message decided on.

decision instance-attribute

decision

As applied: its agents are candidates, each once.

duration_ms class-attribute instance-attribute

duration_ms = 0

How long the policy took: the discussion's bound when it did not decide in time.

ClassifierDispatchPolicy

ClassifierDispatchPolicy(classifier, *, threshold=0.5, max_agents=2, recent=8)

Bases: DispatchPolicy

Picks the candidates a classifier judges the message is for.

Parameters:

Name	Type	Description	Default
classifier	Classifier	Answers the questions, all candidates in one call (JevClassifier, LLMClassifier...). Its failure is the discussion's fallback: every candidate takes the message.	required
threshold	float	The probability of yes a candidate needs. None reaching it: no agent takes the message (thanks, small talk).	0.5
max_agents	int	How many candidates take it at most, likeliest first.	2
recent	int	How many earlier messages the classifier reads.	8

state

state(turn)

What the classifier reads: the team, the conversation, the message.

MockDispatchPolicy

MockDispatchPolicy(decisions=(), *, delay=0.0, error=None)

Bases: DispatchPolicy

Decides from a script: each call takes the next decision, given as a :class:DispatchDecision or as the agents' channel ids, and the last one repeats. With no script, the message asks what it asks with no policy: the candidates that asked its author, else every candidate. delay holds each decision back, error is raised instead. Records every turn.

Agents

Agent

Agent(channel_id, *, provider=None, name=None, role=None, description=None, purpose=None, scope=None, voice=None, greeting=None, language=None, auto_greet=True, identity_in_prompt=True, **kwargs)

Bases: AIChannel

AI agent with structured identity metadata.

Extends :class:AIChannel with role, description, scope, voice, and greeting fields. The first three are auto-injected into the system prompt as an identity block, unless identity_in_prompt=False (a host that renders identity in its own prompt keeps the fields readable without the block); voice is read by :meth:ConversationPipeline.install to auto-wire the voice map; greeting is spoken directly via TTS when a new voice session becomes ready (controlled by auto_greet).

When provider is omitted the agent is config-only — it holds identity and prompt data for speech-to-speech orchestration via :class:RealtimeVoiceChannel but cannot generate responses itself.

Example::

# Full agent (STT → LLM → TTS)
triage = Agent(
    "agent-triage",
    provider=GeminiAIProvider(config),
    role="Triage receptionist",
    description="Routes callers to the right specialist",
    voice="21m00Tcm4TlvDq8ikWAM",
    system_prompt="Greet callers warmly.",
)

# Config-only agent (speech-to-speech)
triage = Agent(
    "agent-triage",
    role="Triage receptionist",
    description="Routes callers to the right specialist",
    voice="Aoede",
    system_prompt="Greet callers warmly.",
)

is_config_only property

is_config_only

Whether this agent has no AI provider (config-only mode).

system_prompt property

system_prompt

The base system prompt for this agent.

build_identity_block

build_identity_block(language=None)

Public access to :meth:_build_identity_block for orchestration callers.

ConversationRouter

ConversationRouter(rules=None, default_agent_id=None, supervisor_id=None)

Routes events to the appropriate agent.

Installed as a BEFORE_BROADCAST sync hook (priority -100) that stamps routing metadata on events. The EventRouter reads this metadata to skip non-targeted intelligence channels.

Usage::

router = ConversationRouter(
    rules=[...],
    default_agent_id="agent-triage",
    supervisor_id="agent-supervisor",
)

kit.hook(
    HookTrigger.BEFORE_BROADCAST,
    execution=HookExecution.SYNC,
    priority=-100,
)(router.as_hook())

Initialize the router.

Parameters:

Name	Type	Description	Default
rules	list[RoutingRule] | None	Routing rules evaluated in ascending priority order (lower number = evaluated first, i.e. higher precedence).	None
default_agent_id	str | None	Fallback agent when no rule matches.	None
supervisor_id	str | None	Agent that receives escalations.	None

select_agent

select_agent(event, context, state)

Determine which agent should handle this event.

Answers for unaddressed events only. An addressed event never reaches here — :meth:as_hook returns before calling this, because an address outranks every rule below (RFC §19.4, step 0).

Priority order: 1. If active_agent_id is set and the agent is still in the room -> sticky affinity (agent keeps handling) 2. Evaluate rules in priority order -> first match wins 3. Fall back to default_agent_id

as_hook

as_hook()

Return a BEFORE_BROADCAST sync hook function.

The hook stamps event.metadata with routing information that EventRouter uses to filter intelligence channels — except on an addressed event, which the router leaves alone (RFC §19.4, step 0).

install

install(kit, agents, *, agent_aliases=None, phase_map=None, hook_priority=-100)

Wire routing and handoff in one call.

Registers this router as a BEFORE_BROADCAST sync hook, builds a HandoffHandler, and calls setup_handoff on every agent.

Returns the HandoffHandler for further customisation.

RoutingRule

Bases: BaseModel

A routing rule mapping conditions to an agent.

Rules are evaluated in ascending priority order (lower values first). The first matching rule wins. Use negative priorities to ensure a rule is evaluated before the default 0, or positive values to defer it.

priority class-attribute instance-attribute

priority = 0

Evaluation order — lower values are checked first (default 0).

RoutingConditions

Bases: BaseModel

Conditions for a routing rule to match.

All specified conditions are ANDed — every non-None field must match.

Conversation State

ConversationState

Bases: BaseModel

Tracks conversation progress within a room.

All fields are optional with sensible defaults so rooms without orchestration have zero overhead.

transition

transition(to_phase, to_agent=None, reason='', metadata=None)

Create a new state with a phase transition recorded.

Returns a new instance (immutable pattern) — the original is never modified.

ConversationPhase

Bases: StrEnum

Built-in conversation phases.

Users can use custom string values — routing and state do not restrict phases to this enum.

PhaseTransition

Bases: BaseModel

Audit record for a phase change.

get_conversation_state

get_conversation_state(room)

Extract typed ConversationState from room metadata.

Returns a fresh default state if no orchestration state exists.

set_conversation_state

set_conversation_state(room, state)

Return a room copy with updated conversation state.

Does NOT persist: :func:save_conversation_state does, without a full room write.

save_conversation_state async

save_conversation_state(store, room_id, state)

Persist state as room_id's conversation state, writing its metadata key alone: a full room write from a room read earlier would undo what was written since, the room's register of authors among it (RFC §10.1 step 12).

Raises:

Type	Description
RoomNotFoundError	room_id does not exist.

Pipeline

ConversationPipeline

ConversationPipeline(stages, default_phase=None, supervisor_id=None)

Generates routing rules for sequential agent workflows.

Example::

pipeline = ConversationPipeline(
    stages=[
        PipelineStage(phase="analysis", agent_id="agent-discuss", next="coding"),
        PipelineStage(phase="coding", agent_id="agent-coder", next="review"),
        PipelineStage(phase="review", agent_id="agent-reviewer",
                      next="report", can_return_to={"coding"}),
        PipelineStage(phase="report", agent_id="agent-writer", next=None),
    ],
    supervisor_id="agent-supervisor",
)
router = pipeline.to_router()

stages property

stages

The pipeline stages.

to_router

to_router()

Generate a ConversationRouter from this pipeline.

get_phase_map

get_phase_map()

Return agent_id -> default phase mapping for HandoffHandler.

get_allowed_transitions

get_allowed_transitions()

Return phase -> allowed next phases for validation.

install

install(kit, agents, *, agent_aliases=None, hook_priority=-100, greet_on_handoff=False, voice_channel_id=None, greeting_prompt=None, farewell_prompt=None)

Wire routing and handoff in one call.

Creates a router from this pipeline, registers it as a BEFORE_BROADCAST sync hook, builds a HandoffHandler with the pipeline's phase map and transition constraints, and calls setup_handoff on every agent.

When greet_on_handoff is True, two extra hooks are registered:

• ON_HANDOFF (async): blocks the old agent's farewell via a BEFORE_TTS flag, then sends an INSTRUCTION addressed to the new agent on voice_channel_id to prompt it to greet — never stored as the caller's words (RFC §10.1.1).
• BEFORE_TTS (sync): blocks TTS while a handoff is pending.

Returns (router, handler) for further customisation.

PipelineStage

Bases: BaseModel

A stage in a conversation pipeline.

Handoff

HANDOFF_TOOL module-attribute

HANDOFF_TOOL = AITool(name=HANDOFF_TOOL_NAME, description="Transfer this conversation to another agent or specialist. Use when: the user asks to speak with someone else, the user wants to be transferred or go back to a previous agent, the conversation needs expertise you don't have, or your task is complete and the next step requires a different agent. Always provide a clear summary of the conversation so far.", parameters={'type': 'object', 'properties': {'target': {'type': 'string', 'description': "Target agent ID or alias (e.g., 'agent-advisor', 'agent-coder', 'human')"}, 'reason': {'type': 'string', 'description': 'Why the handoff is needed'}, 'summary': {'type': 'string', 'description': 'Summary of the conversation so far and what the next agent needs to know to continue effectively'}, 'next_phase': {'type': 'string', 'description': "Optional: conversation phase to transition to (e.g., 'handling', 'review', 'escalation')"}, 'channel_escalation': {'type': 'string', 'enum': ['same', 'voice', 'email', 'sms'], 'description': "Whether to escalate to a different channel. 'same' keeps the current channel."}}, 'required': ['target', 'reason', 'summary']})

build_handoff_tool

build_handoff_tool(targets)

Build a handoff tool with constrained target 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:HANDOFF_TOOL.	required

Returns:

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

HandoffHandler

HandoffHandler(kit, router, agent_aliases=None, phase_map=None, allowed_transitions=None, known_agents=None, on_handoff_complete=None, event_channel_id=None)

Processes handoff tool calls.

Intercepts the handoff_conversation tool, updates conversation state, emits a system event, and returns the result to the agent.

greeting_map property writable

greeting_map

Agent ID to greeting text mapping.

agents property writable

agents

Agent ID to Agent instance mapping.

on_handoff_complete property writable

on_handoff_complete

Callback invoked after a successful handoff.

get_room_language

get_room_language(room, agent_id)

Get effective language: room override > agent default.

handle async

handle(room_id, calling_agent_id, arguments)

Process a handoff request.

Steps: 1. Resolve target agent (alias -> channel_id) 2. Validate target exists in room 3. Update ConversationState 4. Persist state to room metadata 5. Emit system event in room 6. Return result to calling agent

send_greeting async

send_greeting(room_id, *, channel_id=None)

Send the initial agent greeting for a room.

Looks up the current active agent from conversation state and has it say its greeting through :meth:RoomKit.send_greeting, which stores it as the agent's line and delivers it as one: injected with the assistant intent into each :class:RealtimeVoiceChannel session (RFC §12.4), otherwise broadcast to the room so a voice channel speaks it. A greeting is never sent as something the user said, which the agent would answer. The room's language, when set, reaches a realtime session first as a silent instruction.

Call this after setting the initial conversation state::

await save_conversation_state(kit.store, room_id, ConversationState(...))
await handler.send_greeting(room_id, channel_id="voice")

Does nothing when the active agent has no greeting configured.

set_language async

set_language(room_id, language, *, channel_id=None)

Change the conversation language for a room.

Stores the language in conversation state and, for realtime sessions, reconfigures the active agent's session with an updated system prompt that includes the language instruction.

Parameters:

Name	Type	Description	Default
room_id	str	The room to update.	required
language	str	Language name or code (e.g. "French", "fr").	required
channel_id	str | None	Voice channel ID. Falls back to the event channel configured by install().	None

HandoffRequest

Bases: BaseModel

Parsed from agent's handoff tool call arguments.

HandoffResult

Bases: BaseModel

Result returned to the calling agent.

HandoffMemoryProvider

HandoffMemoryProvider(inner)

Bases: _MemoryWrapper

Injects handoff context (summary, reason) into agent prompts.

Wraps an inner MemoryProvider and prepends handoff information when the conversation state indicates a recent handoff. Its window, ingestion and lifecycle are the inner provider's.

setup_handoff

setup_handoff(channel, handler, *, tool=None, room_id=None)

Serve the handoff tool on an AIChannel.

• Declares the handoff tool in the channel's turns: in every room it serves, or in room_id's only (a strategy installed in that room)
• Serves handoff_conversation calls with handler
• Hands off in the room of the call, read from the tool call context (RFC §19.6): one agent serves every room it is attached to

Parameters:

Name	Type	Description	Default
channel	AIChannel	The AI channel to wire handoff into.	required
handler	HandoffHandler	The handoff handler that processes tool calls.	required
tool	AITool | None	Optional custom handoff tool (e.g. from :func:build_handoff_tool). Defaults to the generic :data:HANDOFF_TOOL.	None
room_id	str | None	The one room the handoff serves, when a strategy installs it for that room (RFC §19.7). None serves every room.	None

Raises:

Type	Description
RuntimeError	The handoff is already set up where it would be declared.

Structured results

Delegated workers can hand their result back through the submit_result tool instead of a free-text message scraped from the room — a forced, parseable handoff. See Structured results in the Agent Delegation guide. Enabled per call via kit.delegate(require_structured_result=True), and used internally by the supervised sequential flow.

kit.delegate(result_tool=...) forces another tool than submit_result, by passing a ResultTool: the supervised flow's verdict comes back through submit_verdict this way. The capture is scoped to the child room, so two delegations to the same agent at once never read each other's result.

ResultTool dataclass

ResultTool(tool, normalize, on_missing, reminder)

The tool a delegated agent must call to hand its work back, and how it is read.

A delegation run with require_structured_result injects :attr:tool, captures its call, reminds the agent with :attr:reminder when a turn ends without it, and gives up after the retries with :attr:on_missing. The call is how the structure is forced, so it works with any provider that calls tools, whatever its support for a response schema. :data:SUBMIT_RESULT (a worker's result) is the default; the supervised flow hands its verdict back through its own tool.

Attributes:

Name	Type	Description
tool	AITool	The tool injected into the agent for the delegation.
normalize	Callable[[dict[str, Any]], dict[str, Any]]	Turns the call's arguments into the payload returned.
on_missing	MissingResult	Builds the payload returned when the agent never called the tool, from role, last_output and attempts (keywords).
reminder	str	The message sent when a turn ends without the call.

name property

name

The tool's name.

matches

matches(tool_name)

Whether tool_name is a call of this tool.

A function-calling provider calls it by its bare name; an agent whose tools come from an MCP server sees it prefixed (mcp__<server>__<name>). Both count.

MissingResult

Bases: Protocol

Builds the payload returned when a delegated agent never called its tool.

SUBMIT_RESULT module-attribute

SUBMIT_RESULT = ResultTool(tool=SUBMIT_RESULT_TOOL, normalize=normalize_result, on_missing=orchestration_fail, reminder='You did not submit a result. You MUST now call the `submit_result` tool with your final structured result. Do NOT reply with plain text or a question — call submit_result.')

SUBMIT_RESULT_TOOL module-attribute

SUBMIT_RESULT_TOOL = AITool(name=SUBMIT_RESULT_TOOL_NAME, description='Submit your FINAL result for this task. You MUST call this exactly once, when your work is done — it is the ONLY way to hand your work to the next step. Do NOT end your turn with a plain message or a question back to the user; call submit_result with your structured result.', parameters={'type': 'object', 'properties': {'status': {'type': 'string', 'enum': ['completed', 'failed'], 'description': 'completed if you did the task; failed only if you genuinely could not.'}, 'summary': {'type': 'string', 'description': 'One or two sentences summarizing what you produced.'}, 'data': {'type': 'object', 'description': 'Your structured result, for the next step to build on.'}, 'deliverables': {'type': 'array', 'items': {'type': 'object', 'properties': {'title': {'type': 'string'}, 'url': {'type': 'string'}}}, 'description': 'Concrete artifacts you produced (e.g. a published report URL).'}, 'reason': {'type': 'string', 'description': 'If status is failed, explain why.'}}, 'required': ['status', 'summary']})

is_submit_result

is_submit_result(tool_name)

Whether tool_name is a submit_result call.

A function-calling provider calls the injected tool by its bare name; a claude_code worker calls the gateway-exposed tool, which the sandbox surfaces with an MCP prefix (mcp__<server>__submit_result). Match both so the capture is delivery-agnostic.

normalize_result

normalize_result(arguments)

Coerce a submit_result tool-call payload into the canonical shape.

orchestration_fail

orchestration_fail(*, role, last_output, attempts)

Fail payload the orchestration submits on a worker's behalf when it never called submit_result after attempts tries.

Distinct from a worker self-reporting a task failure (status="failed" via the tool): by="orchestration" marks a MECHANISM-level failure, carrying the worker's last raw output as explanatory context so the next step and the supervisor understand precisely what went wrong.