> ## Documentation Index
> Fetch the complete documentation index at: https://docs.zeroruntime.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Agent & Session

> Python API reference for Agent & Session.

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

```python theme={null}
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'
```

<ParamField path="instructions" type="str" required>
  The system prompt. Persona, task, and the rules the model is expected to hold to.
</ParamField>

<ParamField path="name" type="Optional[str]">
  Display name. Defaults to `agent_id`.
</ParamField>

<ParamField path="pipeline" type="Any">
  The `Pipeline` this agent runs on.
</ParamField>

<ParamField path="tools" type="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.
</ParamField>

<ParamField path="agent_id" type="Optional[str]">
  Required. The name `zeroruntime.serve` registers under and the runtime routes sessions to.
</ParamField>

<ParamField path="mcp_servers" type="Optional[list]">
  `MCPServerStdio` and `MCPServerHTTP` whose tools join the agent's own. Connected on the first call to `initialize_mcp`, not here.
</ParamField>

<ParamField path="inherit_context" type="bool" default="False">
  Carry the chat history over when another agent hands off to this one. Off means it starts the conversation fresh.
</ParamField>

<ParamField path="greeting" type="Optional[str]">
  Spoken as soon as the agent joins, before the caller says anything.
</ParamField>

<ParamField path="farewell" type="Optional[str]">
  Spoken on the way out, when the agent ends the call itself.
</ParamField>

<ParamField path="wake_up" type="int" default="0">
  Seconds of caller silence before the agent nudges. 0 disables it; negative is rejected.
</ParamField>

<ParamField path="wake_up_message" type="str" default="">
  What the nudge says.
</ParamField>

<ParamField path="call_summary" type="Optional[CallSummary]">
  `CallSummary` -- summarise the conversation at teardown and POST it somewhere.
</ParamField>

<ParamField path="agents" type="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.
</ParamField>

<ParamField path="max_session_duration_seconds" type="Optional[int]">
  Hard ceiling on the call. The runtime ends the session when it is reached.
</ParamField>

<ParamField path="tool_timeout_seconds" type="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.
</ParamField>

### register\_tools

```python theme={null}
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

```python theme={null}
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.

<ParamField path="tools" type="List[FunctionTool]" required />

### on\_enter

```python theme={null}
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

```python theme={null}
async def on_exit(self) -> 'None'
```

Called as the call is ending, before teardown finishes.

### on\_participant\_joined

```python theme={null}
async def on_participant_joined(self, participant: "'Participant'") -> 'None'
```

Called when somebody joins the room the agent is in.

<ParamField path="participant" type="'Participant'" required />

### on\_participant\_left

```python theme={null}
async def on_participant_left(self, participant: "'Participant'") -> 'None'
```

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

<ParamField path="participant" type="'Participant'" required />

### hangup

```python theme={null}
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.

<ParamField path="reason" type="str" default="agent ended the call">
  Recorded against the session.
</ParamField>

<ParamField path="farewell" type="str" default="">
  Spoken before hanging up. Empty uses the agent's own `farewell`.
</ParamField>

### cleanup

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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`.

<ParamField path="mcp_server" type="Any" required>
  The server to connect.
</ParamField>

### on

```python theme={null}
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.

<ParamField path="event" type="T" required>
  The event name.
</ParamField>

<ParamField path="callback" type="Callable[..., Any] | None">
  The handler. Omitted, this returns a decorator.
</ParamField>

<ResponseField name="returns" type="Callable[..., Any]">
  The decorator, or the handler it registered.
</ResponseField>

### off

```python theme={null}
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.

<ParamField path="event" type="T" required />

<ParamField path="callback" type="Callable[..., Any]" required />

### emit

```python theme={null}
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.

<ParamField path="event" type="T" required>
  The event name.
</ParamField>

***

## 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

<ParamField path="job_id" type="str" default="">
  The runtime's id for this dispatch.
</ParamField>

<ParamField path="room_id" type="str" default="">
  The room the agent is joining.
</ParamField>

<ParamField path="agent_id" type="str" default="">
  Which registration the job was routed to.
</ParamField>

<ParamField path="token" type="str" default="">
  The auth token for this job. Kept out of `repr` so it does not end up in a log line.
</ParamField>

<ParamField path="metadata" type="Dict[str, Any]" default="…">
  Whatever the dispatcher attached -- the caller's number, a tenant id, anything the job was started with.
</ParamField>

<ParamField path="room_options" type="Room" default="…">
  The `Room` the runtime resolved for this job.
</ParamField>

***

## CallSummary

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

### Fields

<ParamField path="enabled" type="bool" default="True">
  Set False to carry the configuration without summarising.
</ParamField>

<ParamField path="endpoint" type="Optional[str]">
  Where the summary is POSTed. Without one it is generated and logged but sent nowhere.
</ParamField>

<ParamField path="headers" type="Dict[str, str]" default="…">
  Sent with the POST -- an auth header, usually.
</ParamField>

<ParamField path="instruction" type="Optional[str]">
  Replaces the built-in summarising prompt.
</ParamField>

<ParamField path="llm" type="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.
</ParamField>

<ParamField path="timeout_s" type="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.
</ParamField>

***

## serve

```python theme={null}
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.

<ParamField path="agent" type="Any" required>
  A callable returning an `Agent`. Your subclass is usually the callable. An instance is rejected: concurrent calls would share one conversation.
</ParamField>

<ParamField path="room" type="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.
</ParamField>

<ParamField path="on_ready" type="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.
</ParamField>

<ParamField path="capacity" type="int" default="10">
  Concurrent sessions this worker accepts. The `ZERORUNTIME_MAX_CONCURRENT_SESSIONS` environment variable wins when it is set to a number.
</ParamField>

<ParamField path="load_threshold" type="float" default="0.75">
  Fraction of capacity above which the worker reports itself loaded and the registry prefers another.
</ParamField>

<ParamField path="initialize_timeout" type="float" default="10.0">
  Inert. Warned about and ignored.
</ParamField>

<ParamField path="log_level" type="str" default="INFO">
  Applied to the `zeroruntime` logger, and installs colored logging if nothing else has configured the root logger.
</ParamField>

<ParamField path="host" type="str" default="0.0.0.0">
  Interface the status server binds.
</ParamField>

<ParamField path="debug" type="bool" default="True">
  Set False to skip the status server entirely.
</ParamField>

<ParamField path="debug_port" type="int" default="8081">
  Port for the status server.
</ParamField>

<ParamField path="audio_listener_enabled" type="bool" default="False">
  Inert. Warned about and ignored.
</ParamField>

<ParamField path="avatar" type="Any">
  Inert -- the avatar is a pipeline slot. Put it on `Pipeline(avatar=...)`.
</ParamField>

***

## invoke

```python theme={null}
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.

<ParamField path="agent_id" type="str" required>
  The id passed to `zeroruntime.serve`.
</ParamField>

<ParamField path="room" type="Optional[Any]">
  A `Room` layered over the one `serve()` was given -- name only what differs. Without `room_id` a room is created.
</ParamField>

<ParamField path="sip" type="Optional[Any]">
  A `Sip` leg to dial out on, for an outbound call.
</ParamField>

<ParamField path="labels" type="Optional[Dict[str, str]]">
  Not applied. Warned about and ignored.
</ParamField>

<ParamField path="metadata" type="Optional[Dict[str, Any]]">
  Attached to the job, and readable from `AgentContext`.
</ParamField>

<ParamField path="recording_config" type="Optional[Dict[str, Any]]">
  Only `enabled` travels; anything else is warned about and dropped.
</ParamField>

<ParamField path="session_id" type="Optional[str]">
  Not applied. Warned about and ignored.
</ParamField>

<ParamField path="runtime_address" type="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.
</ParamField>

<ParamField path="timeout" type="float" default="30.0">
  Seconds to wait for the runtime to accept.
</ParamField>

<ResponseField name="returns" type="Dict[str, str]">
  `session_id`, `room_id` and `worker_id`, plus `playground_url` when the room asked for one.
</ResponseField>

***

## 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

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

<ParamField path="stream" type="Any" required />

<ParamField path="accepted" type="'pb.SessionAccepted'" required />

<ParamField path="outbound" type="'asyncio.Queue'" required />

<ParamField path="stub" type="Any" />

<ParamField path="auth_token" type="Optional[str]" />

<ParamField path="tools" type="Any" />

### log\_playground

```python theme={null}
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

```python theme={null}
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.

<ResponseField name="returns" type="AsyncIterator[Event]">
  One `Event` per server message.
</ResponseField>

### wait

```python theme={null}
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.

<ParamField path="log" type="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.
</ParamField>

### aclose

```python theme={null}
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

```python theme={null}
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.

<ParamField path="text" type="str" required>
  What to say.
</ParamField>

<ParamField path="interrupt" type="bool" default="False">
  Cut off whatever is playing instead of queueing behind it.
</ParamField>

<ParamField path="interruptible" type="Optional[bool]">
  Whether the caller can barge in over this line. `None` leaves the pipeline's own setting alone.
</ParamField>

<ParamField path="add_to_chat_context" type="Optional[bool]">
  Whether the line joins the conversation history. `None` defers to the runtime.
</ParamField>

<ParamField path="audio_data" type="Optional[bytes]">
  Pre-rendered audio to play instead of sending `text` through TTS. `text` is still what the transcript records.
</ParamField>

<ResponseField name="returns" type="'UtteranceHandle'">
  A handle for this utterance -- await it to know when it finished, or read it to find out it was interrupted.
</ResponseField>

### reply

```python theme={null}
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.

<ParamField path="instructions" type="str" required>
  What to tell the model for this turn alone.
</ParamField>

<ParamField path="frames" type="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.
</ParamField>

<ParamField path="interruptible" type="Optional[bool]">
  Whether the caller can barge in over the answer.
</ParamField>

<ParamField path="wait_for_playback" type="Optional[bool]">
  Resolve the handle only once the audio has finished playing, rather than when generation completes.
</ParamField>

<ResponseField name="returns" type="'UtteranceHandle'">
  A handle for the utterance this produces.
</ResponseField>

### get\_context\_history

```python theme={null}
async def get_context_history(self, *, limit: 'int' = 0, timeout: 'float' = 10.0) -> 'list'
```

Fetch the conversation as the agent process holds it.

<ParamField path="limit" type="int" default="0">
  Newest N messages. 0 is all of them.
</ParamField>

<ParamField path="timeout" type="float" default="10.0">
  Seconds to wait for the answer.
</ParamField>

<ResponseField name="returns" type="list">
  The messages, oldest first.
</ResponseField>

### get\_metrics

```python theme={null}
async def get_metrics(self, *, timeout: 'float' = 10.0) -> 'list'
```

Fetch the per-turn latency metrics collected so far.

<ParamField path="timeout" type="float" default="10.0" />

<ResponseField name="returns" type="list" />

### change\_pipeline

```python theme={null}
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`.

<ParamField path="pipeline" type="Any" required>
  The `Pipeline` to switch to.
</ParamField>

<ParamField path="instructions" type="str" default="">
  New system prompt to apply with the swap. Empty keeps the current one.
</ParamField>

<ParamField path="timeout" type="float" default="30.0">
  Seconds to wait for the runtime to confirm.
</ParamField>

<ResponseField name="returns" type="str">
  The mode the session is running in after the swap.
</ResponseField>

### change\_component

```python theme={null}
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.

<ParamField path="stt" type="Any" default="NO_CHANGE" />

<ParamField path="llm" type="Any" default="NO_CHANGE" />

<ParamField path="tts" type="Any" default="NO_CHANGE" />

<ParamField path="vad" type="Any" default="NO_CHANGE" />

<ParamField path="turn_detector" type="Any" default="NO_CHANGE" />

<ParamField path="denoise" type="Any" default="NO_CHANGE" />

<ParamField path="instructions" type="str" default="" />

<ParamField path="timeout" type="float" default="30.0" />

<ResponseField name="returns" type="str" />

### play\_background\_audio

```python theme={null}
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)`.

<ParamField path="file" type="Any">
  A path, or a `BackgroundAudio` carrying one along with its volume and looping. A disabled `BackgroundAudio` plays nothing.
</ParamField>

<ParamField path="volume" type="float" default="1.0">
  Gain, when `file` is a path.
</ParamField>

<ParamField path="looping" type="bool" default="False">
  Restart at the end, when `file` is a path.
</ParamField>

<ParamField path="override_thinking" type="Optional[bool]">
  Let this bed play over the thinking sound rather than yielding to it.
</ParamField>

### set\_thinking\_audio

```python theme={null}
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.

<ParamField path="file" type="Optional[str]" />

<ParamField path="volume" type="float" default="0.3" />

### stop\_background\_audio

```python theme={null}
async def stop_background_audio(self) -> 'None'
```

Stop the sound bed. Leaves the thinking sound alone.

### subscribe\_to\_pubsub

```python theme={null}
async def subscribe_to_pubsub(self, pubsub_config: 'PubSubSubscribeConfig') -> 'None'
```

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

<ParamField path="pubsub_config" type="PubSubSubscribeConfig" required>
  The topic and its handler.
</ParamField>

### publish\_to\_pubsub

```python theme={null}
async def publish_to_pubsub(self, pubsub_config: 'PubSubPublishConfig') -> 'None'
```

Publish one pubsub frame into the room.

<ParamField path="pubsub_config" type="PubSubPublishConfig" required>
  The topic, body and options to publish with.
</ParamField>

### add\_message

```python theme={null}
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.

<ParamField path="role" type="str" required>
  `system`, `developer`, `user` or `assistant`. A `ChatRole` works too.
</ParamField>

<ParamField path="content" type="str" required>
  The message body.
</ParamField>

<ParamField path="agent_id" type="str" default="">
  Whose context to write into, when the call has more than one agent. Empty means the one running.
</ParamField>

<ParamField path="replace" type="bool" default="False">
  Replace the previous message of this role instead of appending -- how a rolling system prompt is kept from growing.
</ParamField>

### add\_handoff

```python theme={null}
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`.

<ParamField path="to_agent" type="str" required>
  The `agent_id` taking over.
</ParamField>

<ParamField path="from_agent" type="str" default="">
  Who is handing off. Empty means the agent running.
</ParamField>

<ParamField path="reason" type="str" default="">
  Recorded with the handoff, and shown in traces.
</ParamField>

### process\_text

```python theme={null}
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.

<ParamField path="text" type="str" required />

### get\_participants

```python theme={null}
async def get_participants(self, *, timeout: 'float' = 10.0) -> 'list'
```

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

<ParamField path="timeout" type="float" default="10.0" />

<ResponseField name="returns" type="list" />

### transfer\_call

```python theme={null}
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.

<ParamField path="transfer_to" type="str" required>
  Where to send them -- a SIP URI or a phone number.
</ParamField>

<ParamField path="timeout" type="float" default="30.0">
  Seconds to wait for the runtime to report the outcome.
</ParamField>

<ResponseField name="returns" type="dict">
  What the runtime reported about the transfer.
</ResponseField>

### interrupt

```python theme={null}
async def interrupt(self, *, force: 'bool' = False) -> 'None'
```

Stop the agent talking.

<ParamField path="force" type="bool" default="False">
  Cut off even an utterance that was marked non-interruptible.
</ParamField>

### end

```python theme={null}
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.

<ParamField path="reason" type="str" default="client ended the session">
  Recorded against the session.
</ParamField>

<ParamField path="farewell" type="str" default="">
  Spoken before hanging up.
</ParamField>

### stop

```python theme={null}
async def stop(self, reason: 'str' = 'client requested stop') -> 'None'
```

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

<ParamField path="reason" type="str" default="client requested stop" />

### detach

```python theme={null}
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

```python theme={null}
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.

<ParamField path="reason" type="str" default="client requested destroy" />

### on

```python theme={null}
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.

<ParamField path="event" type="T" required>
  The event name.
</ParamField>

<ParamField path="callback" type="Callable[..., Any] | None">
  The handler. Omitted, this returns a decorator.
</ParamField>

<ResponseField name="returns" type="Callable[..., Any]">
  The decorator, or the handler it registered.
</ResponseField>

### off

```python theme={null}
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.

<ParamField path="event" type="T" required />

<ParamField path="callback" type="Callable[..., Any]" required />

### emit

```python theme={null}
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.

<ParamField path="event" type="T" required>
  The event name.
</ParamField>

***

## current\_session

```python theme={null}
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.

<ResponseField name="returns" type="'Session'" />

***

## Participant

Somebody in the room with the agent.

### Fields

<ParamField path="id" type="str" required>
  The room's id for this peer.
</ParamField>

<ParamField path="name" type="str" required>
  Display name, which may be empty.
</ParamField>

<ParamField path="mode" type="str" default="">
  The peer's send/receive mode, as the transport labels it.
</ParamField>

<ParamField path="meta" type="dict" default="…">
  Whatever metadata the peer joined with.
</ParamField>

***

## PubSubSubscribeConfig

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

Passed to `Session.subscribe_to_pubsub`.

### Fields

<ParamField path="topic" type="str" required>
  The topic to subscribe to.
</ParamField>

<ParamField path="cb" type="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.
</ParamField>

***

## PubSubPublishConfig

One frame to publish into the room.

Passed to `Session.publish_to_pubsub`.

### Fields

<ParamField path="topic" type="str" required>
  The topic to publish on.
</ParamField>

<ParamField path="message" type="str" default="">
  The text body.
</ParamField>

<ParamField path="payload" type="Optional[dict]">
  Structured data alongside it, JSON-encoded on the way out.
</ParamField>

<ParamField path="options" type="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.
</ParamField>

***

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

```python theme={null}
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

```python theme={null}
UtteranceHandle(utterance_id: 'str') -> 'None'
```

<ParamField path="utterance_id" type="str" required />

### done

```python theme={null}
def done(self) -> 'bool'
```

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

<ResponseField name="returns" type="bool" />

***

## Room

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

### Fields

<ParamField path="room_id" type="Optional[str]">
  Existing room to join. A new room is created when `None`.
</ParamField>

<ParamField path="name" type="str" default="Agent">
  Display name the agent publishes under.
</ParamField>

<ParamField path="agent_name" type="Optional[str]">
  The 0.1.2 spelling of `name`. Folded into it when set.
</ParamField>

<ParamField path="auth_token" type="Optional[str]">
  VideoSDK token for the room. Falls back to the environment. It travels in `StartSession.credentials`, never in `params_json`.
</ParamField>

<ParamField path="playground" type="bool" default="False">
  Ask the runtime for a playground URL for this session.
</ParamField>

<ParamField path="vision" type="bool" default="False">
  Deliver camera frames to the agent process.
</ParamField>

<ParamField path="audio_codec" type="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.
</ParamField>

<ParamField path="recording" type="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.
</ParamField>

<ParamField path="recording_video" type="bool" default="False">
  Also record camera video. Needs `recording`.
</ParamField>

<ParamField path="recording_screen_share" type="bool" default="False">
  Also record the screen share. Needs `recording` and `vision` -- there is no screen-share track on a session that is not receiving video.
</ParamField>

<ParamField path="agent_participant_id" type="Optional[str]">
  Publish the agent under a fixed participant id. `None` lets the room mint one, which is what a single-agent session wants.
</ParamField>

<ParamField path="auto_end_session" type="Optional[bool]">
  End the session when the last participant leaves. `None` leaves the runtime's default (TRUE) in place.
</ParamField>

<ParamField path="background_audio" type="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.
</ParamField>

<ParamField path="join_meeting" type="Optional[bool]">
  Whether the agent joins the meeting. `None` leaves the runtime's default (TRUE) in place -- see `to_proto`.
</ParamField>

<ParamField path="wait_for_participant" type="Optional[bool]">
  Whether the agent waits for someone to join before speaking. `None` leaves the runtime's default (TRUE) in place.
</ParamField>

<ParamField path="signaling_base_url" type="Optional[str]">
  Override the signaling base URL.
</ParamField>

<ParamField path="observability" type="Optional[Observability]">
  Where this session's telemetry goes. `None` takes the runtime's defaults.
</ParamField>

<ParamField path="session_timeout_seconds" type="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.
</ParamField>

<ParamField path="no_participant_timeout_seconds" type="Optional[int]">
  Seconds with nobody attached before the session is reclaimed. Carried in `SessionLimits.inactivity_timeout_seconds`, same as above.
</ParamField>

<ParamField path="idle_timeout_seconds" type="Optional[int]" />

***

## Sip

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

### Fields

<ParamField path="call_to" type="Optional[str]">
  Destination number/address for an outbound call.
</ParamField>

<ParamField path="call_from" type="Optional[str]">
  Caller id presented on the call.
</ParamField>

<ParamField path="call_type" type="Optional[str]">
  Call direction/type.
</ParamField>

<ParamField path="call_id" type="Optional[str]">
  Identifier correlating the call.
</ParamField>

<ParamField path="webhook_url" type="Optional[str]">
  URL to receive call-event callbacks.
</ParamField>

<ParamField path="extra" type="Dict[str, str]" default="…">
  Additional key/values merged into the metadata map.
</ParamField>

### to\_metadata

```python theme={null}
def to_metadata(self) -> 'Dict[str, str]'
```

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

<ResponseField name="returns" type="Dict[str, str]" />

***

## UNSET

```python theme={null}
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

```python theme={null}
ZeroRuntimeChannel(target: 'Optional[str]' = None, *, auth_token: 'Optional[str]' = None, secure: 'Optional[bool]' = None, options: 'Optional[list]' = None) -> 'None'
```

<ParamField path="target" type="Optional[str]" />

<ParamField path="auth_token" type="Optional[str]" />

<ParamField path="secure" type="Optional[bool]" />

<ParamField path="options" type="Optional[list]" />

### use\_tls

```python theme={null}
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.

<ResponseField name="returns" type="bool" />

### connect

```python theme={null}
async def connect(self, *, timeout: 'float' = 5.0) -> "'ZeroRuntimeChannel'"
```

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

<ParamField path="timeout" type="float" default="5.0">
  Seconds to wait for the channel to become ready. 0 returns without waiting, and the first RPC finds out instead.
</ParamField>

<ResponseField name="returns" type="'ZeroRuntimeChannel'">
  This channel, so it can be chained.
</ResponseField>

### aclose

```python theme={null}
async def aclose(self) -> 'None'
```

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

### status

```python theme={null}
async def status(self) -> 'ZeroRuntimeStatus'
```

Ask the runtime what capacity it has.

<ResponseField name="returns" type="ZeroRuntimeStatus">
  Its `ZeroRuntimeStatus`.
</ResponseField>

### session\_info

```python theme={null}
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.

<ParamField path="session_id" type="str" required />

<ResponseField name="returns" type="Any" />

### destroy

```python theme={null}
async def destroy(self, session_id: 'str', reason: 'str' = 'client requested destroy') -> 'Any'
```

Tear a session down at the runtime, by id.

<ParamField path="session_id" type="str" required>
  The session to destroy.
</ParamField>

<ParamField path="reason" type="str" default="client requested destroy">
  Recorded against it.
</ParamField>

<ResponseField name="returns" type="Any">
  The runtime's response.
</ResponseField>

### attach

```python theme={null}
async def attach(self, session_id: 'str', *, tools: 'Any' = None, heartbeat_interval_ms: 'int' = 0) -> 'Session'
```

Attach to a session that is already running.

<ParamField path="session_id" type="str" required>
  The session to attach to.
</ParamField>

<ParamField path="tools" type="Any">
  Tools to answer this session's tool calls with.
</ParamField>

<ParamField path="heartbeat_interval_ms" type="int" default="0">
  How often to ping.
</ParamField>

<ResponseField name="returns" type="Session">
  The attached `Session`.
</ResponseField>

### start

```python theme={null}
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.

<ParamField path="agent" type="Any">
  The `Agent` to run. Its pipeline and tools are used unless overridden here.
</ParamField>

<ParamField path="pipeline" type="Any">
  The `Pipeline` to run, when there is no agent to take one from.
</ParamField>

<ParamField path="room" type="Any">
  The `Room` to join.
</ParamField>

<ParamField path="sip" type="Any">
  A `Sip` leg, for an outbound call. Folded into the metadata.
</ParamField>

<ParamField path="tools" type="Any">
  Tools to expose. Defaults to the agent's.
</ParamField>

<ParamField path="metadata" type="Optional[dict]">
  Attached to the session.
</ParamField>

<ParamField path="agent_id" type="Optional[str]">
  Recorded in the metadata, so traces name the agent.
</ParamField>

<ParamField path="extra_credentials" type="Optional[dict]">
  Vendor keys beyond the ones the pipeline's providers name for themselves.
</ParamField>

<ParamField path="inactivity_timeout_s" type="Optional[float]">
  Seconds without a participant before the runtime reclaims the session. Defaults to the room's setting.
</ParamField>

<ParamField path="max_duration_s" type="Optional[float]">
  Hard ceiling on the call. Defaults to the room's, then the agent's.
</ParamField>

<ResponseField name="returns" type="Session">
  The live `Session`.
</ResponseField>
