Agent & Session - Zero Runtime

The core classes that define an agent and drive a live call: the Agent you subclass to give it instructions and tools, and the AgentSession that runs it, speaking, listening, and managing the call lifecycle. serve and invoke register an agent and start sessions against it.

Agent

What the agent is, what it can do, and how it behaves on a call. Subclass it to give the agent behaviour: methods decorated with @function_tool are registered by the constructor, and the on_* hooks below are called by the session as the call progresses. An instance is inert until zeroruntime.serve registers it and a session is started against it. The pipeline is the other half. This class carries the prompt, the tools and the call-shaped options; Pipeline carries the providers that hear and speak.

Constructor

Agent(instructions: 'str', name: 'Optional[str]' = None, pipeline: 'Any' = None, tools: 'Optional[List[FunctionTool]]' = None, mcp_servers: 'Optional[list]' = None, inherit_context: 'bool' = False, greeting: 'Optional[str]' = None, farewell: 'Optional[str]' = None, wake_up: 'int' = 0, wake_up_message: 'str' = '', call_summary: 'Optional[CallSummary]' = None, agents: "Optional[List['Agent']]" = None, max_session_duration_seconds: 'Optional[int]' = None, tool_timeout_seconds: 'Optional[int]' = None, **deprecated: 'Any') -> 'None'

str

required

The system prompt. Persona, task, and the rules the model is expected to hold to.

Optional[str]

Display name. Defaults to the agent id.

Any

The Pipeline this agent runs on.

Optional[List[FunctionTool]]

@function_tool callables defined elsewhere. Tools defined as methods on the agent are found on their own, so this is for the ones that are not.

Optional[list]

MCPServerStdio and MCPServerHTTP whose tools join the agent’s own. Connected on the first call to initialize_mcp, not here.

bool

default:"False"

Carry the chat history over when another agent hands off to this one. Off means it starts the conversation fresh.

Optional[str]

Spoken as soon as the agent joins, before the caller says anything.

Optional[str]

Spoken on the way out, when the agent ends the call itself.

int

default:"0"

Seconds of caller silence before the agent nudges. 0 disables it; negative is rejected.

str

default:""

What the nudge says.

Optional[CallSummary]

CallSummary — summarise the conversation at teardown and POST it somewhere.

Optional[List['Agent']]

Other agents this one hands off to, kept on alternates. Nothing registers them for you. Every agent takes its id from ZERORUNTIME_AGENT_ID, so agents in one process share an id and Session.add_handoff cannot tell them apart.

Optional[int]

Hard ceiling on the call. The session is ended when it is reached.

Optional[int]

How long to wait for one of this agent’s tools to return before handing the model a timeout. None keeps the default of 30 seconds. Raise it only for an agent whose tools genuinely run long. The case it exists for is a tool that awaits Session.warm_transfer — that runs for supervisor_join_timeout + briefing_timeout, minutes rather than seconds, and the default would kill the round trip while the transfer carried on underneath. The cost is symmetric: a tool that hangs holds the turn for exactly this long.

register_tools

def register_tools(self) -> 'None'

Collect @function_tool methods off self. Called by the constructor. Call it again after adding one at runtime.

update_tools

def update_tools(self, tools: 'List[FunctionTool]') -> 'None'

Replace the tool list outright. This drops the tools that were found on the agent and anything MCP added, so pass the full set you want.

List[FunctionTool]

required

on_enter

async def on_enter(self) -> 'None'

Called once the agent has joined and the session is live. Override it to open the conversation, prime the context, or set up standing configuration such as Session.set_thinking_audio.

on_exit

async def on_exit(self) -> 'None'

Called as the call is ending, before teardown finishes.

on_participant_joined

async def on_participant_joined(self, participant: "'Participant'") -> 'None'

Called when somebody joins the room the agent is in.

'Participant'

required

on_participant_left

async def on_participant_left(self, participant: "'Participant'") -> 'None'

Called when somebody leaves. The agent is not one of them.

'Participant'

required

hangup

async def hangup(self, reason: 'str' = 'agent ended the call', farewell: 'str' = '') -> 'None'

End the call from the agent’s side. A no-op with no live session.

str

default:"agent ended the call"

Recorded against the session.

str

default:""

Spoken before hanging up. Empty uses the agent’s own farewell.

cleanup

async def cleanup(self) -> 'None'

Close the MCP connections and drop the tools they contributed. Called for you at teardown. Safe to call twice, and safe with no MCP servers configured.

initialize_mcp

async def initialize_mcp(self) -> 'None'

Connect every configured MCP server and adopt its tools. Runs once; later calls return immediately. zeroruntime.serve calls it while building the agent, so a server that is down fails the build rather than the first tool call.

add_server

async def add_server(self, mcp_server: 'Any') -> 'None'

Connect one MCP server and append what it publishes to tools. Ignores anything that is not an MCPServerStdio or MCPServerHTTP.

Any

required

The server to connect.

on

def on(self, event: 'T', callback: 'Callable[..., Any] | None' = None) -> 'Callable[..., Any]'

Subscribe to an event. Decorator or direct call. Handlers fire in the order they were registered, and registering the same one twice calls it twice.

required

The event name.

Callable[..., Any] | None

The handler. Omitted, this returns a decorator.

Callable[..., Any]

The decorator, or the handler it registered.

off

def off(self, event: 'T', callback: 'Callable[..., Any]') -> 'None'

Unsubscribe one handler. Matched by identity, so it has to be the same object that was registered — a fresh lambda or a re-decorated function will not match. Removing something that was never subscribed does nothing.

required

Callable[..., Any]

required

emit

def emit(self, event: 'T', *args: 'Any') -> 'None'

Call every handler subscribed to event. Returns as soon as the handlers have been started, not when they finish, and never raises: an exception inside a handler is logged against the event name. Ignored once the emitter has been closed.

required

The event name.

AgentContext

What is known about an incoming job, handed to the agent factory. zeroruntime.serve passes one to your factory if it takes an argument, so the agent can be built around the call it is about to take — a caller’s number out of metadata, a different prompt per room.

Fields

str

default:""

The id of this dispatch.

str

default:""

The room the agent is joining.

str

default:""

Which registration the job was routed to.

str

default:""

The auth token for this job. Kept out of repr so it does not end up in a log line.

Dict[str, Any]

default:"…"

Whatever the dispatcher attached — the caller’s number, a tenant id, anything the job was started with.

Room

default:"…"

The Room settings resolved for this job.

CallSummary

Summarise the conversation when it ends, and POST the result.

Fields

bool

default:"True"

Set False to carry the configuration without summarising.

Optional[str]

Where the summary is POSTed. Without one it is generated and logged but sent nowhere.

Dict[str, str]

default:"…"

Sent with the POST — an auth header, usually.

Optional[str]

Replaces the built-in summarising prompt.

Any

A second model for the summary alone. None reuses the session’s. Worth pointing at something cheap: it runs once, over the whole transcript, while the process is already shutting down.

float

default:"0.0"

Seconds the summariser gets. 0 takes the default. Summarising is an LLM round trip over the whole conversation, so it is a different order of work from the rest of teardown. The ceiling is the shutdown grace period, since this runs while the agent is shutting down — asking for more than that cuts the summary off mid-request rather than granting the time.

serve

def serve(agent: 'Any', *, room: 'Any' = None, on_ready: 'Optional[Callable[[], None]]' = None, capacity: 'int' = 10, load_threshold: 'float' = 0.75, initialize_timeout: 'float' = 10.0, log_level: 'str' = 'INFO', host: 'str' = '0.0.0.0', debug: 'bool' = True, debug_port: 'int' = 8081, audio_listener_enabled: 'bool' = False, avatar: 'Any' = None, **moved_to_room: 'Any') -> 'None'

Register agent and serve every call dispatched to it. Every keyword here describes this worker — how many calls it takes, what it logs, where it answers probes. What a session looks like (recording, codec, timeouts and so on) is set on Room and passed as room=. Blocks until interrupted. The agent is registered with ZeroRuntime under its agent id, and every call dispatched to that id builds a fresh agent from the factory — one instance per call, never one shared across them.

Any

required

A callable returning an Agent. Your subclass is usually the callable. An instance is rejected: concurrent calls would share one conversation.

Any

The Room every dispatched call is given. Every Room field can be set here. A dispatched call never passes through zeroruntime.invoke, so this is the only place its session is described; invoke(room=...) inherits from it and overrides only the fields it names.

Optional[Callable[[], None]]

Called on its own thread once the agent is registered. Where zeroruntime.invoke belongs, for a process that starts its own calls rather than waiting to be dispatched to.

int

default:"10"

Concurrent sessions this worker accepts. The ZERORUNTIME_MAX_CONCURRENT_SESSIONS environment variable wins when it is set to a number.

float

default:"0.75"

Fraction of capacity above which the worker reports itself loaded and new calls prefer another worker.

float

default:"10.0"

Inert. Warned about and ignored.

str

default:"INFO"

Applied to the zeroruntime logger, and installs colored logging if nothing else has configured the root logger.

str

default:"0.0.0.0"

Interface the status server binds.

bool

default:"True"

Set False to skip the status server entirely.

int

default:"8081"

Port for the status server.

bool

default:"False"

Inert. Warned about and ignored.

Any

Inert — the avatar is a pipeline slot. Put it on Pipeline(avatar=...).

invoke

def invoke(agent_id: 'Optional[str]' = None, *, room: 'Optional[Any]' = None, sip: 'Optional[Any]' = None, labels: 'Optional[Dict[str, str]]' = None, metadata: 'Optional[Dict[str, Any]]' = None, recording_config: 'Optional[Dict[str, Any]]' = None, session_id: 'Optional[str]' = None, runtime_address: 'Optional[str]' = None, timeout: 'float' = 30.0) -> 'Dict[str, str]'

Start a session against an agent this process is already serving. Blocking, and deliberately so — it returns once the session has been accepted, which is the point at which the room is joinable. That also means it cannot be called from inside a running event loop: call it from serve(on_ready=...), which runs on its own thread, or await ZeroRuntimeChannel.start from async code. The agent has to be served by this process; invoke cannot start a call on an agent served by another one, because the pipeline, credentials and tools all live where zeroruntime.serve was called. The agent is the one zeroruntime.serve registered, under ZERORUNTIME_AGENT_ID or "zeroruntime_agent".

Optional[str]

Deprecated and ignored; the id comes from the environment.

Optional[Any]

A Room layered over the one serve() was given — name only what differs. Without room_id a room is created.

Optional[Any]

A Sip leg to dial out on, for an outbound call.

Optional[Dict[str, str]]

Not applied. Warned about and ignored.

Optional[Dict[str, Any]]

Attached to the job, and readable from AgentContext.

Optional[Dict[str, Any]]

Only enabled is applied; anything else is warned about and dropped. Prefer Room(recording=...).

Optional[str]

Not applied. Warned about and ignored.

Optional[str]

Start against a different ZeroRuntime address than the one serve() is connected to. The channel is then owned by this call and closed when the session ends.

float

default:"30.0"

Seconds to wait for the session to be accepted.

Dict[str, str]

session_id, room_id and worker_id, plus playground_url when the room asked for one.

Session

One live call, and everything you can do to it while it runs. You are handed one rather than constructing one: zeroruntime.invoke returns it, and inside a @function_tool or an on_* hook it is the session the agent is bound to. Every method here is sent to the running agent, so they are all awaited and none of them block the call. The object stays usable after the call ends — ended goes True and the send methods become no-ops — so a hook can read ended_reason without guarding first.

Constructor

Session(stream: 'Any', accepted: "'pb.SessionAccepted'", outbound: "'asyncio.Queue'", *, stub: 'Any' = None, auth_token: 'Optional[str]' = None, tools: 'Any' = None) -> 'None'

Any

required

'pb.SessionAccepted'

required

'asyncio.Queue'

required

Any

Optional[str]

Any

log_playground

def log_playground(self) -> 'None'

Announce the join URL on stdout, at most once per session. Called for you when a session starts in playground mode, and safe to call again: a session prints its URL only once. It prints rather than logs so the URL is easy to copy into a browser.

events

async def events(self) -> 'AsyncIterator[Event]'

Iterate everything that happens on the call, until it ends. Keeps yielding briefly past the ended event so the last transcripts and metrics are not cut off mid-teardown.

AsyncIterator[Event]

One Event per thing that happened.

wait

async def wait(self, *, log: 'bool' = True) -> 'None'

Consume events until the call ends, logging as it goes. The usual last line of a script that started a session and has nothing else to do.

bool

default:"True"

Set False to drain silently. State frames are never logged either way — they arrive every few seconds and say nothing a reader wants.

aclose

async def aclose(self) -> 'None'

Release everything this call holds. Idempotent, and safe after any ending. Every way a call can finish ends up here, so you rarely need to call it yourself; async with and stop do. Call it when you drop a session some other way, so its connection is closed.

say

async def say(self, text: 'str', *, interrupt: 'bool' = False, interruptible: 'Optional[bool]' = None, add_to_chat_context: 'Optional[bool]' = None, audio_data: 'Optional[bytes]' = None) -> "'UtteranceHandle'"

Speak an exact line. No model involved.

str

required

What to say.

bool

default:"False"

Cut off whatever is playing instead of queueing behind it.

Optional[bool]

Whether the caller can barge in over this line. None leaves the pipeline’s own setting alone.

Optional[bool]

Whether the line joins the conversation history. None keeps the default.

Optional[bytes]

Pre-rendered audio to play instead of sending text through TTS. text is still what the transcript records. Called from inside a @function_tool, the line is spoken while the tool runs. If the tool is cancelled by an interruption, lines sent after that are not spoken, and their handles report them interrupted. Lines sent from a task the tool starts (asyncio.create_task) are not tied to the tool call.

'UtteranceHandle'

A handle for this utterance — await it to know when it finished, or read it to find out it was interrupted.

reply

async def reply(self, instructions: 'str', *, frames: 'int' = 0, interruptible: 'Optional[bool]' = None, wait_for_playback: 'Optional[bool]' = None) -> "'UtteranceHandle'"

Ask the model for a line, generated against instructions. Unlike say, this is a generation: the model sees the conversation so far plus these instructions, and what it produces is spoken. Called from inside a @function_tool, the reply is generated and spoken while the tool runs, in order with the tool’s say calls. As with say, a reply sent after the tool was cancelled by an interruption is not spoken. wait_for_playback does not apply there.

str

required

What to tell the model for this turn alone.

int

default:"0"

How many of the newest camera frames to show the model. The count travels, not the pixels — they are captured on the call’s side. Needs Room(vision=True); without it there is no video track to capture from.

Optional[bool]

Whether the caller can barge in over the answer.

Optional[bool]

Resolve the handle only once the audio has finished playing, rather than when generation completes.

'UtteranceHandle'

A handle for the utterance this produces.

get_context_history

async def get_context_history(self, *, limit: 'int' = 0, timeout: 'float' = 10.0) -> 'list'

Fetch the conversation history as the agent currently holds it.

int

default:"0"

Newest N messages. 0 is all of them.

float

default:"10.0"

Seconds to wait for the answer.

list

The messages, oldest first.

get_metrics

async def get_metrics(self, *, timeout: 'float' = 10.0) -> 'list'

Fetch the per-turn latency metrics collected so far.

float

default:"10.0"

list

change_pipeline

async def change_pipeline(self, pipeline: 'Any', *, instructions: 'str' = '', timeout: 'float' = 30.0) -> 'str'

Replace the whole pipeline mid-call. Everything not named is rebuilt from the pipeline you pass, so state a complete one. To change a provider or two and leave the rest alone, use change_component.

Any

required

The Pipeline to switch to.

str

default:""

New system prompt to apply with the swap. Empty keeps the current one.

float

default:"30.0"

Seconds to wait for the swap to be confirmed.

str

The mode the session is running in after the swap.

change_component

async def change_component(self, *, stt: 'Any' = NO_CHANGE, llm: 'Any' = NO_CHANGE, tts: 'Any' = NO_CHANGE, vad: 'Any' = NO_CHANGE, turn_detector: 'Any' = NO_CHANGE, denoise: 'Any' = NO_CHANGE, instructions: 'str' = '', timeout: 'float' = 30.0) -> 'str'

Swap the named components and leave the rest of the call alone. Only the components you pass change; anything left out keeps running as it is. Passing None removes a component (denoise=None turns noise cancellation off). Must be called while an agent is bound to the session, e.g. from a @function_tool or an on_* hook.

Any

default:"NO_CHANGE"

New speech-to-text provider, or a FallbackSTT chain.

Any

default:"NO_CHANGE"

New language model, or a FallbackLLM chain.

Any

default:"NO_CHANGE"

New text-to-speech provider, or a FallbackTTS chain.

Any

default:"NO_CHANGE"

New voice activity detector.

Any

default:"NO_CHANGE"

New end-of-turn detector.

Any

default:"NO_CHANGE"

New noise cancellation.

str

default:""

New system prompt to apply with the swap. Empty keeps the current one.

float

default:"30.0"

Seconds to wait for the swap to be confirmed.

str

The mode the session is running in after the swap.

play_background_audio

async def play_background_audio(self, file: 'Any' = None, *, volume: 'float' = 1.0, looping: 'bool' = False, override_thinking: 'Optional[bool]' = None) -> 'None'

Start a sound bed under the call. Needs the room’s mixing track, Room(background_audio=True).

Any

A path, or a BackgroundAudio carrying one along with its volume and looping. A disabled BackgroundAudio plays nothing.

float

default:"1.0"

Gain, when file is a path.

bool

default:"False"

Restart at the end, when file is a path.

Optional[bool]

Let this bed play over the thinking sound rather than yielding to it.

set_thinking_audio

async def set_thinking_audio(self, file: 'Optional[str]' = None, *, volume: 'float' = 0.3) -> 'None'

Play a sound while the agent is generating its reply. Standing configuration rather than a one-shot: once set, the session starts it at the top of every generation and stops it when the answer begins, for as long as the session lasts. Set it from on_enter. file=None uses the built-in thinking sound. Any file libav can decode works — wav, mp3, ogg, flac, m4a. Needs the room’s mixing track, Room(background_audio=True), like all audio that is not speech. Without it nothing plays, and a warning is logged in your process.

Optional[str]

float

default:"0.3"

stop_background_audio

async def stop_background_audio(self) -> 'None'

Stop the sound bed. Leaves the thinking sound alone.

subscribe_to_pubsub

async def subscribe_to_pubsub(self, pubsub_config: 'PubSubSubscribeConfig') -> 'None'

Listen on one pubsub topic for the rest of the call.

PubSubSubscribeConfig

required

The topic and its handler.

publish_to_pubsub

async def publish_to_pubsub(self, pubsub_config: 'PubSubPublishConfig') -> 'None'

Publish one pubsub frame into the room.

PubSubPublishConfig

required

The topic, body and options to publish with.

add_message

async def add_message(self, role: 'str', content: 'str', *, agent_id: 'str' = '', replace: 'bool' = False) -> 'None'

Write a message into the conversation without speaking it. Useful for handing the model something it should know — a CRM lookup, a verification result — without a turn being spent saying it out loud.

str

required

system, developer, user or assistant. A ChatRole works too.

str

required

The message body.

str

default:""

Whose context to write into, when the call has more than one agent. Empty means the one running.

bool

default:"False"

Replace the previous message of this role instead of appending — how a rolling system prompt is kept from growing.

append_instructions

async def append_instructions(self, text: 'str', *, delegation_id: 'Optional[str]' = None) -> 'None'

Give a duplex model an instruction it follows from now on. The way to redirect a duplex model mid-call or ask it for exact wording. Only duplex models (OpenAILive) take it, as with the other appends; for any other it is logged here as a warning and not sent.

str

required

The instruction, at most 500 tokens.

Optional[str]

Tie it to a delegation the model made; None makes it session-wide.

append_thinking

async def append_thinking(self, text: 'str', *, delegation_id: 'Optional[str]' = None) -> 'None'

Give a duplex model something to know without saying it. Quiet context: facts the model should use if they come up. Not a place for secrets — it can still reach speech later.

str

required

The context, at most 500 tokens.

Optional[str]

Tie it to a delegation the model made; None makes it session-wide.

append_commentary

async def append_commentary(self, text: 'str', *, delegation_id: 'Optional[str]' = None) -> 'None'

Give a duplex model something to say aloud, in its own words.

str

required

What to say, at most 500 tokens. The model paraphrases it.

Optional[str]

Tie it to a delegation the model made; None makes it session-wide.

add_handoff

async def add_handoff(self, to_agent: 'str', *, from_agent: 'str' = '', reason: 'str' = '') -> 'None'

Hand the conversation to another agent, by id. The target has to be registered — zeroruntime.serve registers one agent_id per call — and whether it inherits the history is that agent’s inherit_context.

str

required

The agent_id taking over.

str

default:""

Who is handing off. Empty means the agent running.

str

default:""

Recorded with the handoff, and shown in traces.

process_text

async def process_text(self, text: 'str') -> 'None'

Feed text to the agent as though the caller had said it. The full turn runs: the model answers and the answer is spoken.

str

required

get_participants

async def get_participants(self, *, timeout: 'float' = 10.0) -> 'list'

Who is in the room right now, as Participant records.

float

default:"10.0"

list

transfer_call

async def transfer_call(self, transfer_to: 'str', *, timeout: 'float' = 30.0) -> 'dict'

Cold-transfer the call and leave. The caller is handed to transfer_to with no introduction and the agent drops out.

str

required

Where to send them — a SIP URI or a phone number.

float

default:"30.0"

Seconds to wait for the transfer’s outcome.

dict

Details of the transfer’s outcome, as a dict.

interrupt

async def interrupt(self, *, force: 'bool' = False) -> 'None'

Stop the agent talking.

bool

default:"False"

Cut off even an utterance that was marked non-interruptible.

end

async def end(self, reason: 'str' = 'client ended the session', farewell: 'str' = '') -> 'None'

End the call. A no-op once it has ended.

str

default:"client ended the session"

Recorded against the session.

str

default:""

Spoken before hanging up.

stop

async def stop(self, reason: 'str' = 'client requested stop') -> 'None'

End the call and release it. What async with calls.

str

default:"client requested stop"

detach

async def detach(self) -> 'None'

Stop watching this session without ending the call. The agent keeps running on the call; your code just stops listening. For a call that should actually stop, use end or destroy.

destroy

async def destroy(self, reason: 'str' = 'client requested destroy') -> 'None'

Tear the session down immediately, without a goodbye. The blunt one: no farewell, no teardown hooks worth waiting for. end is the polite version.

str

default:"client requested destroy"

on

def on(self, event: 'T', callback: 'Callable[..., Any] | None' = None) -> 'Callable[..., Any]'

Subscribe to an event. Decorator or direct call. Handlers fire in the order they were registered, and registering the same one twice calls it twice.

required

The event name.

Callable[..., Any] | None

The handler. Omitted, this returns a decorator.

Callable[..., Any]

The decorator, or the handler it registered.

off

def off(self, event: 'T', callback: 'Callable[..., Any]') -> 'None'

required

Callable[..., Any]

required

emit

def emit(self, event: 'T', *args: 'Any') -> 'None'

required

The event name.

current_session

def current_session() -> "'Session'"

The session whose callback is currently running. Valid anywhere the SDK invokes your code on a session’s behalf: a @function_tool, a lifecycle hook, a pipeline hook. Useful for code that runs on the call without holding a reference to it — a module-level @pipeline.on handler, say. Anything defined on the agent should reach the call through self.session instead, and a pubsub handler subscribed with Session.subscribe_to_pubsub already has the session in hand.

'Session'

Participant

Somebody in the room with the agent.

Fields

str

required

The room’s id for this peer.

str

required

Display name, which may be empty.

str

default:""

The peer’s send/receive mode.

dict

default:"…"

Whatever metadata the peer joined with.

UtteranceHandle

One thing the agent is saying, and how it went. Returned by Session.say and Session.reply. Await it to block until the line finishes — awaiting resolves whether it played out or was cut off, so check interrupted rather than assuming it was heard.

handle = await session.say("One moment while I look that up.")
await handle
if handle.interrupted:
    ...

Awaiting is optional. Ignore the handle and the line still plays.

Constructor

UtteranceHandle(utterance_id: 'str') -> 'None'

str

required

done

def done(self) -> 'bool'

Whether the utterance has settled, either way. Never blocks.

bool

Room

Which room the agent joins. room_id=None asks for a new one.

Fields

Optional[str]

Existing room to join. A new room is created when None.

str

default:"Agent"

Display name the agent publishes under.

Optional[str]

The 0.1.2 spelling of name. Folded into it when set.

Optional[str]

ZeroRuntime auth token for the room. Falls back to the ZERORUNTIME_AUTH_TOKEN environment variable. Kept out of repr.

bool

default:"False"

Ask for a playground URL for this session.

bool

default:"False"

Deliver camera frames to the agent.

Optional[str]

Codec the room negotiates — "opus" (the default when None), "pcmu", "pcma", "g722". This is what tells a meeting from a phone call: opus is 48 kHz, pcmu/pcma are 8 kHz and g722 is 16 kHz, and that rate reaches the VAD and STT. Leave it None for meetings; set "pcmu" or "pcma" for SIP so the pipeline sizes itself to the narrowband audio instead of resampling it up to 48 kHz and back down again.

bool

default:"False"

Record the session. Off unless asked for — recording a call is a decision about the person on the other end, so the default has to be the safe one. Audio is always captured when this is on; the two below add tracks to it.

bool

default:"False"

Also record camera video. Needs recording.

Also record the screen share. Needs recording and vision — there is no screen-share track on a session that is not receiving video.

Optional[str]

Publish the agent under a fixed participant id. None lets the room mint one, which is what a single-agent session wants.

Optional[bool]

End the session when the last participant leaves. None keeps the default (True).

bool

default:"False"

Open the room’s mixing audio track. Nothing plays by itself — this is the track, not the sound. Start audio on it with session.play_background_audio(file=..., volume=..., looping=...) and stop it with session.stop_background_audio(). Required before either call: without the track the call is refused, and the caller hears nothing.

Optional[bool]

Whether the agent joins the meeting. None keeps the default (True).

Optional[bool]

Whether the agent waits for someone to join before speaking. None keeps the default (True).

Optional[str]

Override the signaling base URL.

Optional[Observability]

Where this session’s telemetry goes. None takes the defaults.

Optional[int]

Hard cap on session length, in seconds. The session is ended when it is reached. None inherits the default ceiling. Must be >= 0.

Optional[int]

Seconds with no participant in the room before the session is ended. None keeps the default. Must be >= 0.

Optional[int]

Seconds without conversation — no turn started or finished by either the caller or the agent — before the session is ended, even though people are still connected. Guards against a call that has stalled and would otherwise run until session_timeout_seconds. None or 0 leaves it off. Choose it against how long a caller may legitimately go quiet (a hold, a long think), because it ends the call on them. Checked every few seconds, so the session may run slightly past it; your deployment may also enforce a tighter limit. Must be >= 0.

Sip

The telephony leg of a session — who is being called, and from what.

Fields

Optional[str]

Destination number/address for an outbound call.

Optional[str]

Caller id presented on the call.

Optional[str]

Call direction/type.

Optional[str]

Identifier correlating the call.

Optional[str]

URL to receive call-event callbacks.

Dict[str, str]

default:"…"

Additional key/values merged into the metadata map.

to_metadata

def to_metadata(self) -> 'Dict[str, str]'

This leg as the metadata dict the room is created with.

Dict[str, str]

UNSET

UNSET = UNSET

“You did not pass this.” Distinct from every real value, including False. serve(recording=False) and leaving recording out have to mean different things once a Room template exists — the first overrides the template, the second defers to it — and no ordinary default can express that, because the value a caller most often means to override with is the default itself.

ZeroRuntimeChannel

A connection to a ZeroRuntime server, and the sessions started over it. zeroruntime.serve and zeroruntime.invoke own one for you. Hold one directly to start sessions from async code, to reach a ZeroRuntime address other than the configured one, or to ask what capacity it has. Connect explicitly with await channel.connect(), or use it as an async context manager — every method connects on first use anyway. TLS is decided by the target unless secure says otherwise: local hosts are plaintext, port 443 is TLS. ZERORUNTIME_INSECURE=1 forces plaintext.

Constructor

ZeroRuntimeChannel(target: 'Optional[str]' = None, *, auth_token: 'Optional[str]' = None, secure: 'Optional[bool]' = None, options: 'Optional[list]' = None) -> 'None'

Optional[str]

Optional[bool]

Optional[list]

use_tls

def use_tls(self) -> 'bool'

Whether this channel will speak TLS. secure wins if it was set, then ZERORUNTIME_INSECURE, then the target itself: a local host is plaintext and port 443 is TLS.

bool

connect

async def connect(self, *, timeout: 'float' = 5.0) -> "'ZeroRuntimeChannel'"

Open the connection. Returns immediately if it is already open.

float

default:"5.0"

Seconds to wait for the channel to become ready. 0 returns without waiting, and the first request finds out instead.

'ZeroRuntimeChannel'

This channel, so it can be chained.

aclose

async def aclose(self) -> 'None'

Close the connection. Sessions started over it stop being watched.

status

async def status(self) -> 'ZeroRuntimeStatus'

Ask the server what capacity it has.

ZeroRuntimeStatus

Its ZeroRuntimeStatus.

session_info

async def session_info(self, session_id: 'str') -> 'Any'

Look up one session’s state, pid, room and uptime. Works for any session on this server, including ones this process did not start.

str

required

Any

destroy

async def destroy(self, session_id: 'str', reason: 'str' = 'client requested destroy') -> 'Any'

Tear a session down immediately, by id.

str

required

The session to destroy.

str

default:"client requested destroy"

Recorded against it.

Any

The server’s response.

attach

async def attach(self, session_id: 'str', *, tools: 'Any' = None, heartbeat_interval_ms: 'int' = 0) -> 'Session'

Attach to a session that is already running.

str

required

The session to attach to.

Any

Tools to answer this session’s tool calls with.

int

default:"0"

How often to ping.

Session

The attached Session.

start

async def start(self, *, agent: 'Any' = None, pipeline: 'Any' = None, room: 'Any' = None, sip: 'Any' = None, tools: 'Any' = None, metadata: 'Optional[dict]' = None, agent_id: 'Optional[str]' = None, extra_credentials: 'Optional[dict]' = None, inactivity_timeout_s: 'Optional[float]' = None, max_duration_s: 'Optional[float]' = None) -> 'Session'

Start a session and return it, ready to run. The async counterpart to zeroruntime.invoke, and what to reach for inside async code, where invoke cannot be called.

Any

The Agent to run. Its pipeline and tools are used unless overridden here.

Any

The Pipeline to run, when there is no agent to take one from.

Any

The Room to join.

Any

A Sip leg, for an outbound call. Folded into the metadata.

Any

Tools to expose. Defaults to the agent’s.

Optional[dict]

Attached to the session.

Optional[str]

Recorded in the metadata, so traces name the agent.

Optional[dict]

Vendor keys beyond the ones the pipeline’s providers name for themselves.

Optional[float]

Seconds without a participant before the session is ended. Defaults to the room’s setting.

Optional[float]

Hard ceiling on the call. Defaults to the room’s, then the agent’s.

Session

The live Session.

​Agent

​Constructor

​register_tools

​update_tools

​on_enter

​on_exit

​on_participant_joined

​on_participant_left

​hangup

​cleanup

​initialize_mcp

​add_server

​on

​off

​emit

​AgentContext

​Fields

​CallSummary

​Fields

​serve

​invoke

​Session

​Constructor

​log_playground

​events

​wait

​aclose

​say

​reply

​get_context_history

​get_metrics

​change_pipeline

​change_component

​play_background_audio

​set_thinking_audio

​stop_background_audio

​subscribe_to_pubsub

​publish_to_pubsub

​add_message

​append_instructions

​append_thinking

​append_commentary

​add_handoff

​process_text

​get_participants

​transfer_call

​interrupt

​end

​stop

​detach

​destroy

​on

​off

​emit

​current_session

​Participant

​Fields

​UtteranceHandle

​Constructor

​done

​Room

​Fields

​Sip

​Fields

​to_metadata

​UNSET

​ZeroRuntimeChannel

​Constructor

​use_tls

​connect

​aclose

​status

​session_info

​destroy

​attach

​start

Agent

Constructor

register_tools

update_tools

on_enter

on_exit

on_participant_joined

on_participant_left

hangup

cleanup

initialize_mcp

add_server

on

off

emit

AgentContext

Fields

CallSummary

Fields

serve

invoke

Session

Constructor

log_playground

events

wait

aclose

say

reply

get_context_history

get_metrics

change_pipeline

change_component

play_background_audio

set_thinking_audio

stop_background_audio

subscribe_to_pubsub

publish_to_pubsub

add_message

append_instructions

append_thinking

append_commentary

add_handoff

process_text

get_participants

transfer_call

interrupt

end

stop

detach

destroy

on

off

emit

current_session

Participant

Fields

UtteranceHandle

Constructor

done

Room

Fields

Sip

Fields

to_metadata

UNSET

ZeroRuntimeChannel

Constructor

use_tls

connect

aclose

status

session_info

destroy

attach

start