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, agent_id: 'Optional[str]' = 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) -> 'None'

str

required

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

Optional[str]

Display name. Defaults to 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[str]

Required. The name zeroruntime.serve registers under and the runtime routes sessions to.

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: zeroruntime.serve registers one agent_id per call, and Session.add_handoff names the target by that id.

Optional[int]

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

Optional[int]

How long the runtime waits for one of this agent’s tools to return before handing the model a timeout. None keeps the runtime’s own 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 the runtime knows about a 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 runtime’s id for 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 the runtime 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 runtime’s 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 runtime’s SIGTERM grace, since this runs while the agent is shutting down — asking for more than that gets the process killed 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 left here describes this worker — how many calls it takes, what it logs, where it answers probes. Nothing here describes the session: that is Room’s job, and the twelve keywords which used to say it a second time are gone, because two spellings of recording that could disagree is a question with no right answer. Blocks until interrupted. The agent is registered with the runtime’s registry, 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. All 21 fields are reachable, including observability, join_meeting, wait_for_participant and idle_timeout_seconds, which no keyword ever covered. 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 the registry prefers another.

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: 'str', *, 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 runtime has accepted the session, 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 registered in this process. There is no dispatch RPC to route a start to another one: the pipeline, credentials and tool schemas all live where zeroruntime.serve was called.

str

required

The id passed to zeroruntime.serve.

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 travels; anything else is warned about and dropped.

Optional[str]

Not applied. Warned about and ignored.

Optional[str]

Start against a different runtime 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 runtime to accept.

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 a message to the agent process, 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. Both the direct-start path and the registry worker path call this, and invoke() adds a third; the guard is what keeps one session to one URL. It prints rather than logs because the URL is the one line the operator is waiting to copy, and a log record buries it behind level and module columns that are noise for a value you paste into a browser.

events

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

Iterate everything the runtime reports, until the call 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 server message.

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 routes here. Closing the transport is the part that cannot be skipped: the outbound queue feeds a generator gRPC holds inside a task, and a live task is a collection root.

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 defers to the runtime.

Optional[bytes]

Pre-rendered audio to play instead of sending text through TTS. text is still what the transcript records.

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

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 in the agent process. 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 as the agent process 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 runtime to confirm.

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.

Any

default:"NO_CHANGE"

Any

default:"NO_CHANGE"

Any

default:"NO_CHANGE"

Any

default:"NO_CHANGE"

Any

default:"NO_CHANGE"

Any

default:"NO_CHANGE"

str

default:""

float

default:"30.0"

str

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 takes the runtime’s own 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 the SDK declines and says so in the runtime’s log only, so this reports it back as a diagnostic instead.

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.

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 runtime to report the outcome.

dict

What the runtime reported about the transfer.

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'

Ask the runtime to 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 runtime-side; this process 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 at the runtime, 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, as the transport labels it.

dict

default:"…"

Whatever metadata the peer joined with.

PubSubSubscribeConfig

One topic to listen on, and what to call for each frame. Passed to Session.subscribe_to_pubsub.

Fields

str

required

The topic to subscribe to.

Any

Called once per frame with the raw frame as a dict — the whole thing as the transport delivered it, since pubsub promises no schema beyond message. Sync or async; a coroutine is awaited. A handler that declares a second parameter is additionally told whether the frame is backlog:: def cb(frame, backlog): if backlog: return # already in the topic before we joined Subscribing replays whatever the topic already held, so a one-argument handler sees that history as ordinary traffic — which is the transport’s own shape, and worth knowing about before replying to it.

PubSubPublishConfig

One frame to publish into the room. Passed to Session.publish_to_pubsub.

Fields

str

required

The topic to publish on.

str

default:""

The text body.

Optional[dict]

Structured data alongside it, JSON-encoded on the way out.

dict

default:"…"

Transport options. Only sendOnly — a list of participant ids to deliver to — is carried; anything else is dropped, because the wire has nowhere to put it. Omitted, everyone subscribed to the topic gets the frame.

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]

VideoSDK token for the room. Falls back to the environment. It travels in StartSession.credentials, never in params_json.

bool

default:"False"

Ask the runtime for a playground URL for this session.

bool

default:"False"

Deliver camera frames to the agent process.

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 and the same runtime serves recorded and unrecorded sessions side by side. 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 leaves the runtime’s default (TRUE) in place.

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 SDK refuses, and the failure is a silent one from the caller’s side.

Optional[bool]

Whether the agent joins the meeting. None leaves the runtime’s default (TRUE) in place — see to_proto.

Optional[bool]

Whether the agent waits for someone to join before speaking. None leaves the runtime’s default (TRUE) in place.

Optional[str]

Override the signaling base URL.

Optional[Observability]

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

Optional[int]

Hard cap on session length, in seconds. Carried in SessionLimits.max_session_duration_seconds rather than in RoomSpec, so it is read by the transport and not by to_proto. None inherits the runtime’s own ceiling.

Optional[int]

Seconds with nobody attached before the session is reclaimed. Carried in SessionLimits.inactivity_timeout_seconds, same as above.

Optional[int]

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 one runtime, 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 runtime other than the configured one, or to ask a runtime 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 RPC 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 runtime 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 runtime, 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 at the runtime, by id.

str

required

The session to destroy.

str

default:"client requested destroy"

Recorded against it.

Any

The runtime’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 runtime reclaims the session. 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

​add_handoff

​process_text

​get_participants

​transfer_call

​interrupt

​end

​stop

​detach

​destroy

​on

​off

​emit

​current_session

​Participant

​Fields

​PubSubSubscribeConfig

​Fields

​PubSubPublishConfig

​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

add_handoff

process_text

get_participants

transfer_call

interrupt

end

stop

detach

destroy

on

off

emit

current_session

Participant

Fields

PubSubSubscribeConfig

Fields

PubSubPublishConfig

Fields

UtteranceHandle

Constructor

done

Room

Fields

Sip

Fields

to_metadata

UNSET

ZeroRuntimeChannel

Constructor

use_tls

connect

aclose

status

session_info

destroy

attach

start