Realtime Voice (Speech-to-Speech)

RoomKit's RealtimeVoiceChannel enables speech-to-speech AI conversations using providers like OpenAI Realtime API, Google Gemini Live, xAI Grok Realtime, and ElevenLabs Conversational AI. Audio flows directly between the client and the AI provider — no separate STT/TTS stages.

RealtimeVoiceChannel is the 1:1 surface — one human, one session. The same providers also join multi-party meetings as the conference's intelligence, mixed N→1 onto one session per room: see Speech-to-speech in a conference (RFC §12.10.12).

How It Differs from VoiceChannel

Aspect	VoiceChannel	RealtimeVoiceChannel
Audio flow	Client → STT → AI → TTS → Client	Client ↔ Provider (direct)
VAD	Client-side (AudioPipeline)	Server-side (provider)
Transcription	Generated by STT provider	Provided by realtime provider
Tool calling	Via AIChannel + hooks	Direct provider callbacks
Latency	Higher (STT + AI + TTS)	Lower (single round-trip)

---

Quick Start

from __future__ import annotations

from roomkit import RealtimeVoiceChannel, RoomKit
from roomkit.voice.backends.mock import MockVoiceBackend
from roomkit.providers.openai.realtime import OpenAIRealtimeProvider

provider = OpenAIRealtimeProvider(
    api_key="sk-...",
)  # model defaults to gpt-realtime-2.1

transport = MockVoiceBackend()

channel = RealtimeVoiceChannel(
    "voice-realtime",
    provider=provider,
    transport=transport,
    system_prompt="You are a helpful voice assistant.",
    voice="alloy",
    input_sample_rate=16000,
    output_sample_rate=24000,
    emit_transcription_events=True,
)

kit = RoomKit()
kit.register_channel(channel)

---

RealtimeVoiceChannel Configuration

channel = RealtimeVoiceChannel(
    channel_id="voice-realtime",
    provider=provider,                    # RealtimeVoiceProvider
    transport=transport,                  # VoiceBackend
    system_prompt="...",                  # AI instructions
    voice="alloy",                        # Provider voice preset
    tools=[my_tool],                      # Tool objects — definitions + handlers extracted automatically
    temperature=0.7,                      # Generation temperature
    input_sample_rate=16000,              # Audio from client
    output_sample_rate=24000,             # Audio to provider
    transport_sample_rate=None,           # Transport rate (auto-resamples if different)
    emit_transcription_events=True,       # Emit transcriptions as RoomEvents
    tool_handler=my_tool_handler,         # Optional — for MCP, auditing, or custom dispatch
    mute_on_tool_call=False,              # Mute mic during tool execution
    tool_result_max_length=16384,         # Truncate large tool results
)

Parameter	Default	Description
provider	required	Realtime AI provider (OpenAI, Gemini, xAI Grok, ElevenLabs, Deepgram)
transport	required	Audio transport backend
system_prompt	None	AI system instructions
voice	None	Voice preset name
tools	None	Tool objects or JSON definitions — when Tool objects are passed, definitions and handlers are extracted automatically
temperature	None	Sampling temperature
input_sample_rate	16000	Client → provider sample rate
output_sample_rate	24000	Provider → client sample rate
transport_sample_rate	None	Transport rate; auto-resamples if mismatched
emit_transcription_events	True	Create RoomEvents from transcriptions
tool_handler	None	async (name: str, args: dict) -> str — optional when Tool objects are passed via tools
mute_on_tool_call	False	Mute mic while a tool runs: it stays muted until the last call in flight ends
tool_result_max_length	16384	Max chars of a tool result the model reads: a handler's, a refusal, a Tool Search result, a skill reference. An activated skill's instructions and a complete schema read with list_tools(name=...) are never cut

Every tool call, whatever door it came through (the provider's function call, a call recovered from speech, a reasoning backend's call), is served the same way: the pre-execution gate, the handler inside the tool call context, ON_TOOL_CALL, the bound above, then its one result and its one report. An id names its call until the call's result goes out: a second call under it before then is refused and reported, and sends nothing, since the id's result is the first call's; a call under it after then is a new call and gets its own answer, even while the first is still finishing its report. Every provider frees the id as it sends the result, or as it reports the call abandoned, and the channel frees it at the same step: nothing is sent for an abandoned call, its observers told or not, and a call issued under its id is a new one. So does a call a reconnect its own handler caused, which runs on: the new connection never issued its id. A call that came without an id, or under an id in flight, takes the path of any call: on an ended session it is reported cancelled, a call_tool transport is unwrapped so its report names the tool, and it waits behind the transcription that precedes it. A call that named no tool reaches the channel too, which refuses it before the gate and answers under its id. Ending the session reports each call it interrupted as cancelled.

What the provider owes in return is the same everywhere. A refused, failed or unserved call goes back through submit_tool_error: ElevenLabs marks it as a tool error and Gemini Live sends it under the function response's error key (a served result goes under result, whatever its text), the others send it like any result, its {"error": ...} body saying what happened. Every call the provider abandons is reported through on_tool_call_cancelled, whatever the cause: Gemini's cancellation, a reconnect (a handoff included), an ElevenLabs tool_timeout_s, a connection lost or closed. The channel cancels its handler and reports it once, cancelled, and frees its id at once, as the provider does when it reports the abandonment: a call the vendor issues under that id while the handler cleans up is a new call, answered, and nothing is sent for the abandoned one. A provider whose model calls no tool (Anam, PersonaPlex) has supports_tools set to False: the channel declares it no tool, advertises no skill or Tool Search in its prompt, and warns once at construction.

A provider reads a call's arguments with the shared rule, readable_arguments: a mapping when they read as an object, the text the model wrote when they do not (invalid JSON, an array, a cut fragment). The channel, and a conference, refuse a call that arrives as text before the gate: no handler runs on a mapping that only passes for arguments, the model reads Tool call arguments unreadable, and the observers receive the refusal, as in the AI channel's tool loop. A wire that says whether the response cut a call (OpenAI Realtime's and GPT-Live's item status: "incomplete", after the output cap or a cancel) is read with realtime_call_arguments(raw, cut=...), exported from roomkit.providers.ai beside readable_arguments: a cut call runs only when its argument text arrived and reads, and otherwise reaches the channel marked cut (CutArguments), which the model reads as Tool call cut off. OpenAI Realtime and xAI hand a call on once its item is done (response.output_item.done), the first event that carries that status. A provider also checks its endpoint's tool-name rule when it declares the session's tools, the way an AI provider checks a turn's (see Tool Calling): an endpoint that accepts the declaration can still fail the call later, opaquely, once the model reaches for it. A tool given to the channel or a conference under a name no vendor accepts is refused at construction, as an AITool is.

A provider keeps the calls it issued on the book RealtimeVoiceProvider holds per session: _book_tool_call(session, call_id, payload) when the call arrives (False for a call without an id or under an id in flight, which is handed on all the same for the channel to refuse), _answerable_tool_call first in submit_tool_result and submit_tool_error (it returns the payload, a name or a pending future, and drops with a log a result for an id abandoned, never issued, or issued by a connection that is gone), and _abandon_open_tool_calls(session) when a connection ends, which reports each abandoned call once. A session connected again under the same id abandons its previous connection's calls first.

from roomkit.providers.ai import readable_arguments, realtime_call_arguments


class MyProvider(RealtimeVoiceProvider):
    @property
    def supports_tools(self) -> bool:
        return True  # the default; False declares no tool to the model

    async def _on_function_call(self, session, event) -> None:
        # A mapping, or the model's text when it does not read as one.
        arguments = readable_arguments(event["arguments"])
        # A wire that tells a cut reads the call through the cut rule instead:
        # realtime_call_arguments(event["arguments"], cut=event["status"] == "incomplete")
        # Booked with what answering it takes (here its name); a call that
        # cannot be booked is handed on all the same: the channel refuses it.
        self._book_tool_call(session, event["id"], event["name"])
        await self._fire(
            self._tool_call_callbacks, session, event["id"], event["name"], arguments,
            label="tool_call",
        )

    async def submit_tool_result(self, session, call_id, result) -> None:
        held, name = self._answerable_tool_call(session, call_id)
        if not held:
            return  # abandoned or never issued: logged, nothing sent
        await self._send(session, {"call_id": call_id, "name": name, "output": result})

    async def submit_tool_error(self, session, call_id, result) -> None:
        # Only when the protocol has an error marker; the default sends
        # the result through submit_tool_result.
        held, name = self._answerable_tool_call(session, call_id)
        if held:
            await self._send(
                session, {"call_id": call_id, "name": name, "output": result, "is_error": True}
            )

    async def _start(self, session) -> None:
        # A session's own tasks run in a fresh context, not the caller's
        # (which may be a tool handler reconnecting the session).
        self._receiver = self._session_task(self._receive(session), name="receive")

    async def _on_connection_lost(self, session) -> None:
        # The calls still open die with the connection: report them, once.
        await self._abandon_open_tool_calls(session)

The transport is accepted before the AI provider finishes its handshake, so a caller on a phone line can already be speaking. That audio is buffered in order and flushed the moment the session goes live, up to a bound of roughly thirty seconds of 16 kHz mono PCM; past it, input is dropped with a single warning rather than growing without limit behind a dead handshake. A failure anywhere in that startup — transport, provider, a malformed negotiated sample rate, or the client-ready notification — tears the whole session back down instead of leaving a connection nobody owns.

---

Tool Search with fixed declarations

Tool Search activates automatically when a session's catalogue exceeds tool_search_threshold (20 by default), decided per session on the tools it declares: a pipeline's active agent, the tools a session was opened with or a room's orchestration tools may make one session's catalogue larger than the channel's. Set tool_search=True to enable it for a smaller catalogue or False to expose the catalogue directly. Unless tool_search=False, the channel serves find_tools and list_tools itself, and call_tool on a provider whose declarations are fixed, so a host tool under one of those names is refused at construction. A session whose catalogue Tool Search does not hide declares none of them, and a call to one is refused as undeclared, as on a text turn. tool_search_pinned keeps selected tools directly callable alongside discovery. The tools orchestration adds to the channel stay declared the same way: a voice supervisor's delegate_workers, a voice loop's delegate_loop, setup_realtime_delegation's delegate_task, and a voice pipeline's handoff_conversation. The agent is told to call them, not to find them.

Providers that support mid-session reconfiguration receive native declarations for the tools matched by find_tools, once the call is served and its result went out. ON_TOOL_CALL's SYNC hooks judge find_tools and list_tools before the model reads them, as any other call: a hook that blocks the search reveals nothing and reconfigures nothing, and a hook's replacement is what the model reads. Providers with fixed declarations, such as Gemini 3.1 Live, use three stable functions instead:

Function	Result
find_tools(query)	Matching tool names and short descriptions; queries use English keywords
list_tools(name="calendar")	One tool's complete description and parameter schema
call_tool(name="calendar", arguments_json='{"action":"list"}')	The result of calling the named tool with the decoded arguments

Without name, list_tools(category=...) returns the compact inventory of every tool the session can call, declared or hidden (the orchestration and skill tools included, the policy's refusals and skill-gated tools left out), as on a text turn; on a fixed-declaration provider, list_tools(name=...) and call_tool reach every tool it lists. A schema lookup returns one complete JSON document and is not cut by the business tool-result length limit. The tool name and an operation such as list are separate: the operation belongs in the tool's arguments when its schema declares one.

This transport does not reconnect the session after a search and does not add permissions. Supply only the session's authorized catalogue. Unknown and excluded names both return an unavailable-in-this-session refusal. Arguments, skill gates, and BEFORE_TOOL_USE are checked before the same handler used by native calls. ON_TOOL_CALL observes the real tool name and arguments with the provider's original call ID. Invalid arguments, explicit gate refusals, and execution failures remain distinguishable.

call_tool is reserved when fixed-declaration Tool Search is active; a caller catalogue containing that name is rejected. Infrastructure functions are called directly rather than recursively through call_tool.

Ending a session cancels its in-flight tool tasks and rejects late calls. It does not cancel another session's tasks or undo external effects that have already occurred.

The realtime Tool Search example runs Gemini Live against 112 fictional tools, captures tool-call traces and speech, and verifies calendar and project operations in one connection. Its input is injected text; it checks the Live tool protocol rather than speech recognition. It requires GEMINI_API_KEY and performs no business mutations.

Passing provider_config

Everything provider-specific — VAD tuning, transcription model, Gemini's sensitivities, an ElevenLabs first message — travels in one dict called provider_config. It is not a channel constructor argument: it reaches the provider through the session metadata, so two sessions on the same channel can run different settings.

provider_config = {
    "turn_detection_type": "semantic_vad",   # keys are provider-specific
}

session = await channel.start_session(
    room_id="room-1",
    participant_id="caller-1",
    connection=None,
    metadata={"provider_config": provider_config},
)

Every provider_config snippet below shows only the dict — pass it as metadata={"provider_config": ...} exactly as above. Unknown keys are ignored silently, so a typo produces default behaviour rather than an error: check the key against the provider reference.

server_vad is not a provider_config key

Whether the provider does its own endpointing is derived from the channel's pipeline, not from provider_config: a pipeline carrying a VAD stage puts OpenAI, Gemini and Grok in manual mode (the channel sends the activity signals from the local VAD), and without one their server-side VAD runs. ElevenLabs and Deepgram always do their own turn-taking — Deepgram warns that it ignored the request, ElevenLabs takes it silently.

---

What inject_text's role means

The role is an intent, not a wire field: each provider maps it onto what its API offers.

• role="system" is an instruction — how to behave, or what to do now.
• role="user" is content: a user turn the model answers on a turn-based provider, words the model says aloud on a full-duplex one.
• role="assistant" is a line the agent says now (a configured greeting). A provider with a speech primitive uses it (Deepgram's InjectAgentMessage, Anam's talk); the others ask the model to say it. It never reaches the model as the user's words, which the model would answer.
• silent=True adds any of them as context without asking for a response.

Intent	OpenAI Realtime	Gemini Live	OpenAI GPT-Live
system	system message, then a response	a user turn (no system role in turns)	instructions append
user	user message, then a response	user turn	commentary: said aloud, paraphrased
assistant	system message asking to say the line, then a response	user turn asking to say the line	instructions append asking to say the line
silent	the message, no response	turn left incomplete; a marked context text once audio flows	thinking append

So anything that directs the model is role="system", an instruction to greet included; a greeting the application wrote itself is role="assistant". Sent as role="user", a full-duplex model says it — or improvises around it — instead of following it:

# Right on every provider
await channel.inject_text(session, "Greet the user by first name, then listen.", role="system")

# Wrong on GPT-Live: the model voices the instruction as its own words
await channel.inject_text(session, "Greet the user by first name, then listen.")

---

Which model is speaking

provider.model_name names the model behind a session, for a log line, a span attribute or a diagnostic. Read it as the best identifier this provider can give, not as a guaranteed model id: a service that exposes no end-to-end model falls back to the provider name.

Provider	model_name
OpenAIRealtimeProvider	the realtime model it connects to (e.g. gpt-realtime-2.1)
GeminiLiveProvider	the Live model it connects to
XAIRealtimeProvider	the Grok realtime model it connects to
DeepgramAgentProvider	its think model — the stage that decides what the agent says, not its listen or speak stage
AnamRealtimeProvider	the persona's llm_id when the persona is inline; the provider name when the persona lives in Anam Lab
ElevenLabsRealtimeProvider	the provider name — the agent is configured in the dashboard
PersonaPlexRealtimeProvider	the provider name — one self-hosted model, no id to give

A provider written outside RoomKit inherits the default and needs no change; override the property when your service names a model. MockRealtimeProvider() reports the default, and MockRealtimeProvider(model="…") stands in for a provider that names one — so a test can exercise either shape.

---

What a session reports as usage

What a spoken turn consumed is known to the service that billed it, so the provider reports it and RoomKit relays it unaltered (RFC §12.4.2). Two public surfaces carry that report.

on_usage(callback) — fired on every report. This is the one to bill a call from:

def record(session, usage):
    cached = usage.get("cached_content_token_count", 0)
    ledger.add(session.id, fresh=usage["input_tokens"] - cached, cached=cached)

provider.on_usage(record)

The callback takes (session, usage), may be sync or async, and runs on a task of its own — a slow listener never holds up the provider's event loop, and one that raises is logged without reaching the others or the session.

session.last_usage — the snapshot the last report left behind. Empty until the first report, and a copy: writing to it changes nothing.

async def turn_done(session):
    logger.info("turn cost %s", session.last_usage)

provider.on_response_end(turn_done)

Read it at the end of a turn, and only there. The next report replaces it, and the realtime channel clears it once it has closed the turn's telemetry span, so a ledger built by polling comes out short — that is what on_usage is for.

What the keys mean

input_tokens and output_tokens are the only two keys the framework fixes. Everything beside them is the provider's own, under the names its API uses, and an absent key means unreported, not zero.

Provider	Keys beside the two totals
OpenAIRealtimeProvider	input_token_details, output_token_details — the API's own dicts (cached, text and audio tokens)
GeminiLiveProvider	cached_content_token_count, thoughts_token_count, tool_use_prompt_token_count, total_token_count, plus the modality maps prompt_tokens_details, cache_tokens_details, response_tokens_details and tool_use_prompt_tokens_details, each {"AUDIO": n, "TEXT": n}
OpenAILiveProvider (GPT-Live)	live_seconds — this model bills session duration, never restated as tokens — and, in hosted delegation, backend with the backend model's own tokens
Other providers	none: the two totals, when the service reports them at all

A spoken turn is mostly audio, the modalities are priced apart and the cached share is priced apart again, so the two totals alone cannot say what a session spent its context on. A host that prices a call reads the breakdown; one that only logs a number does not have to.

examples/realtime_usage_accounting.py runs the whole thing on the mock provider — no key, no microphone — including what a poller misses.

---

OpenAI Realtime API

WebSocket-based speech-to-speech with server-side VAD.

from __future__ import annotations

from roomkit.providers.openai.realtime import OpenAIRealtimeProvider

provider = OpenAIRealtimeProvider(
    api_key="sk-...",
    model="gpt-realtime-2.1",              # default; -2.1-mini is cheaper
    base_url=None,                         # Custom endpoint (optional)
)

Reasoning-capable models (gpt-realtime-2 and later) accept a reasoning effort through provider_config:

provider_config = {
    "reasoning_effort": "low",  # minimal|low|medium|high|xhigh
}

Omit the key and the field never reaches the session, which is what non-reasoning models need. reasoning_effort and image_detail are validated against their allowed values, so a typo raises ValueError at connect or reconfigure() time instead of travelling to the API verbatim.

Images

gpt-realtime-2.1 and later accept image input, so a picture can be put in front of the model inside the live conversation:

await channel.inject_image(
    session,
    image_bytes,
    "image/png",                       # PNG and JPEG only
    prompt="What do you see here?",    # optional, travels in the same item
)

An "image_detail": "low" in provider_config trades fidelity for tokens. Left unset, the API's own default applies, which resolves to high detail — worth setting explicitly on a session that injects frames repeatedly. It is local input policy rather than a session.update field, so a reconfigure() that changes only image_detail is applied to the live session without sending anything on the wire.

Pass silent=True to add the image as context without asking for a spoken answer. Providers without image support (xAI Grok) raise NotImplementedError, which RealtimeVoiceChannel catches and logs.

VAD Configuration

Turn detection is configured with flat provider_config keys — the provider assembles the API's nested turn_detection object itself:

# Semantic VAD — the default
provider_config = {
    "turn_detection_type": "semantic_vad",
    "eagerness": "high",               # low, medium, high, auto
    "interrupt_response": True,
    "create_response": True,
}

# Server VAD (energy-based)
provider_config = {
    "turn_detection_type": "server_vad",
    "threshold": 0.5,
    "silence_duration_ms": 800,
    "prefix_padding_ms": 300,
    "idle_timeout_ms": 8000,           # optional; turn times out on silence
    "interrupt_response": True,
    "create_response": True,
}

# Manual turn management (no automatic VAD)
provider_config = {"turn_detection_type": None}

Left unset, turn_detection_type defaults to semantic_vad on a channel whose pipeline has no VAD stage (and to no turn detection at all when it has one — see the note above). RoomKit sends an explicit turn_detection: null when local VAD is selected; merely omitting the field would leave OpenAI's default server VAD active. semantic_vad improves semantic end-of-turn timing, but its speech_started event is not an acoustic echo filter. Speaker leakage must still be removed by AEC or gated before the audio reaches OpenAI.

threshold, silence_duration_ms, prefix_padding_ms and idle_timeout_ms apply to server_vad only; eagerness applies to semantic_vad only. interrupt_response and create_response work with both.

Local speaker and microphone example

The local OpenAI example uses WebRTC AEC and continuous WebRTC noise suppression by default:

OPENAI_API_KEY=... \
    uv run python examples/realtime_voice_local_openai.py

Variable	Default	Description
AEC	webrtc	webrtc, speex, or 0 to disable echo cancellation
DENOISE	webrtc	webrtc, rnnoise, sherpa, or 0 to disable continuous noise suppression
AEC_DELAY_MS	0	Fixed speaker-to-microphone delay in milliseconds; 0 lets the local backend give AEC3 the delay its devices' clocks report, on every captured frame
BARGE_IN_GUARD_MS	2000	Silence sent to provider-side VAD after physical playback starts while local AEC converges; 0 disables the guard
MUTE_MIC	1	1 mutes capture during assistant playback (stable half-duplex); 0 opts into full-duplex/barge-in
OPENAI_LANGUAGE	automatic	Optional input transcription language such as fr; useful when ambiguous noise is decoded in another script

Keep DENOISE=webrtc when using open speakers. OpenAI's server VAD cancels an active response when it detects new speech; without continuous local noise suppression, residual echo at the beginning of a playback turn can therefore produce a false interruption. This stage complements OpenAI's far_field input noise reduction and runs on every captured frame, whether the AEC is active or not.

OpenAI may finish generating (response.done) several seconds before the local speaker drains its buffered audio. Its WebSocket VAD can otherwise hear that still-playing response, commit it as a new user turn, and start a reply loop. The example therefore defaults to MUTE_MIC=1: stable half-duplex operation, with no barge-in while the assistant speaks. Use MUTE_MIC=0 only with headphones or a calibrated full-duplex AEC setup. AEC_DELAY_MS should include the speaker and microphone device latency when a fixed value is used. With AEC_DELAY_MS=0, LocalAudioBackend seeds AEC3 with the sum of PortAudio's two stream latencies before playback, then, from the first captured frame, tells it the delay the devices' clocks give (outputBufferDacTime, inputBufferAdcTime), as browsers do: on a MacBook the latencies added up to 296 ms where the clocks put the delay at 42 ms. A host API that reports no clock keeps the seed.

In full-duplex mode, BARGE_IN_GUARD_MS=2000 protects only playback onset. During that interval, RoomKit still processes and records the real microphone signal, so AEC and noise suppression continue converging, but it forwards equal-duration PCM silence to OpenAI. After two seconds, normal barge-in resumes. Set the value to 0 for an echo-free path if interruptions must work from the first sample.

Fragments such as はい, Cyrillic words, or other unexpected scripts in a user transcription are not byte-decoding corruption: they are the transcription model interpreting ambiguous residual audio while language detection is automatic. Removing the false VAD turn fixes the source. For a monolingual session, OPENAI_LANGUAGE=fr additionally constrains input transcription and improves recognition accuracy.

WebSocket interruption context

OpenAI generates audio faster than real time, while RealtimeVoiceChannel controls physical playback. On a true barge-in, RoomKit now measures elapsed playback from the transport callback and sends conversation.item.truncate for the latest assistant audio item. The duration is capped at the amount of audio actually received, preventing an API error after a playback underrun. This keeps OpenAI's conversation context aligned with what the user heard even when response.done arrived before the speaker finished.

Tool results and the caller's turn

The model is asked to go on (response.create) once per response, when every call it made has its result. That request waits while the caller is speaking: between the pipeline VAD's speech start and end in manual mode, between the server VAD's speech_started and speech_stopped otherwise. Results that land during a barge-in are then answered together with what the caller said, by the request that closes the caller's turn (the server's own, under server VAD, unless OpenAI's create_response is false). No request goes out while another is in progress, active or sent and not yet begun, whichever path sends it: inject_text, the end of the caller's turn, or a continuation. A caller's turn whose request met a response in progress is answered once that response ends, unless a request sent since already covers it, and a request the server rejects with an error no longer counts as in progress. xAI Grok speaks the same wire and behaves the same way; it has no create_response setting, its server VAD always answers the turn.

Available Voices

alloy, echo, shimmer, breeze, cinnamon, juniper, sage (varies by model)

Input Transcription

# Configure which model transcribes user input
provider_config = {
    "stt_model": "gpt-4o-transcribe",
}

---

OpenAI GPT-Live (full-duplex)

GPT-Live (gpt-live-1) is a different API from OpenAI Realtime. The model listens and speaks at the same time, handles being interrupted by itself, and holds no tools: it delegates reasoning and tool use to a backend model while the conversation continues (RFC §12.4.1).

from __future__ import annotations

from roomkit.providers.openai.live import HostedReasoning, OpenAILiveProvider

provider = OpenAILiveProvider(
    api_key="sk-...",
    model="gpt-live-1",                    # default
    delegation=HostedReasoning(            # OpenAI runs the backend model
        model="gpt-5.6-terra",
        instructions="You are the backend of a voice assistant. ...",
        reasoning_effort="low",
    ),
    turn_gap_ms=800,                       # quiet that closes a synthesized turn
)

Install with pip install roomkit[realtime-openai] (the websockets extra).

What changes on a full-duplex session

• No barge-in path. RealtimeVoiceChannel reads provider.full_duplex and leaves interruption to the model: no playback flush, no interrupt() or truncate_audio(), no gating of the model's audio while the user speaks. Keep an AEC (or headphones) so the model does not hear itself; the microphone must stay open during playback for the model to be talked over.
• Boundaries are synthesized. The wire carries no response or speech events. The provider groups transcript deltas per speaker and closes a turn after turn_gap_ms of quiet; that is when the final transcription and on_response_end fire. Partial transcriptions carry deltas for both roles.
• A pipeline VAD is observation only. Its events still reach ON_SPEECH_START / ON_SPEECH_END, but they never interrupt the model, and InterruptionConfig has no effect.
• The session is fixed at start. Model, instructions, voice, audio format and delegation mode are set once by session.start. reconfigure() appends a changed system prompt to the instructions and, in hosted mode, updates the backend's tools without replacing the session; a voice or codec change reconnects, and the hosted calls still open on the old connection are reported cancelled. supports_mid_session_reconfigure is False, so skills default to inline_full.
• Text injection is paraphrased. inject_text(role="system") appends instructions; role="user" adds spoken context the model relays in its own words, or silent context with silent=True. There is no user turn once the session started, so an instruction — the opening greeting included — is role="system", which is how OpenAI documents having the model speak first (see What inject_text's role means). Long texts are split at the API's 500-token per-append bound, and a text RoomKit set apart in a frame keeps it in each append: a <worker_output> block is closed and opened again around every piece, a quote such as Marie: “…” too, after its author (RFC §12.4.1). inject_image() is not available.
• One audio format for both directions, chosen from the channel's output_sample_rate: PCM at 16 or 24 kHz, or G.711 at 8 kHz with provider_config={"codec": "pcmu"} or "pcma". A different input_sample_rate is resampled by the provider.
• Usage is seconds, not tokens. session.last_usage["live_seconds"] carries the cumulative live duration; a hosted backend's tokens land under session.last_usage["backend"] with the backend model's name. Both are reported through on_usage as they land — see What a session reports as usage.

Two delegation modes

Mode	Who runs the backend	Tools	Results
HostedReasoning(model=...)	OpenAI (Responses API)	the channel's tools, served by tool_handler / ON_TOOL_CALL as usual	function results via submit_tool_result; the backend resumes once every call of a response is answered
IntegratorReasoning() (default)	your ReasoningBackend on the channel	the channel's tools, executed by the backend through the channel gate	text, relayed as spoken or silent context

See the Reasoning Delegation guide for the integrator mode. ON_REALTIME_DELEGATION fires for both modes with the delegation id and its target ("hosted" or "integrator").

provider_config keys

codec (pcm, pcmu, pcma) and history (a list of {"role", "text"} text messages seeding the session, at most 128).

Voices, pace and tone

The voice is fixed for the session (voice= on the channel or in the session metadata); changing it means a new session, which is what reconfigure(voice=...) does. Fourteen voices: marin (default) and cedar, plus regional ones — quartz, ripple (Australian), vesper (British), willow, stone (Irish), gleam, meridian (North American), delta, cinder (Southern US), beacon (Filipino), bossa, tempo (Brazilian Portuguese). available_voices() lists them with language and gender. Custom voices from your own recording exist on the API but need OpenAI's approval.

There is no speed parameter. Delivery is steered through the instructions, and GPT-Live follows them closely — OpenAI's own prompting guide suggests:

Speak warmly and naturally, at an unhurried pace. Be clear and direct, not
overly cheerful. For routine questions, give one or two short sentences.
Use moderate backchannels. Stop speaking when the user interrupts.

Pace and tone can also be adjusted mid-session, since instructions are appended rather than replaced:

await channel.inject_text(session, "Slow down and speak more calmly.", role="system")

Pronunciation follows the same route ("Say Rosalia as roh-sah-LEE-ah"), and instructions written in the target language steer the spoken language.

Local speaker and microphone example

examples/realtime_voice_local_openai_live.py (hosted backend, weather tools) and examples/realtime_voice_local_openai_live_backend.py (Claude backend, three-step rebooking).

---

Google Gemini Live

Persistent streaming connection with session resumption and advanced features.

from __future__ import annotations

from roomkit.providers.gemini.realtime import GeminiLiveProvider

provider = GeminiLiveProvider(
    api_key="your-gemini-key",
    model="gemini-3.8-live",
)

gemini-3.8-live is the default. gemini-3.8-live-extended-thinking is the same model with configurable reasoning; both replaced gemini-3.1-flash-live-preview, which stays in the catalog flagged deprecated so an existing deployment keeps resolving.

What changed with the 3.8 generation

Google changed four contracts. RoomKit absorbs all four, so a configuration written for 3.1 keeps working on 3.8 — but knowing which is which explains why some settings silently stop applying.

Contract	Through 3.1	From 3.8
End of response	turn_complete	the model speaks several times per request; only interaction_status IDLE ends it
Affective dialog	enable_affective_dialog	removed from the API
Proactive audio	proactive_audio toggled it	permanently on, stating it is an error
Tool calls	blocking by default	run in the background while the model keeps talking

A field the target model no longer accepts is dropped with a WARNING naming the field and the model, never forwarded — a refused setup takes the whole session down, and an inherited config is not a reason to lose the session.

Advanced Configuration

Gemini supports several unique features via provider_config (the voice itself stays a channel parameter — voice="Aoede"):

provider_config = {
    # VAD sensitivity — LOW or HIGH; the API has no MEDIUM
    "start_of_speech_sensitivity": "HIGH",
    "end_of_speech_sensitivity": "LOW",
    "silence_duration_ms": 500,

    # Reasoning — gemini-3.8-live-extended-thinking only, where it is
    # REQUIRED: that model answers a missing level with 1007 and closes the
    # socket. RoomKit falls back to `low` when you name none.
    # low | medium | high. `minimal` is refused before the round trip.
    "thinking_level": "medium",

    # Which input the server folds into a turn. 3.8 defaults to
    # TURN_INCLUDES_AUDIO_ACTIVITY_AND_ALL_VIDEO, which bills every frame
    # on a video session.
    "turn_coverage": "TURN_INCLUDES_ONLY_ACTIVITY",

    # When a background tool result is delivered. Opt-in, and nothing is
    # sent by default: gemini-3.8-live-extended-thinking closes the session
    # with 1007 over this field, and the models that take it deliver a
    # background result sensibly without it.
    # WHEN_IDLE | INTERRUPT | SILENT
    "tool_response_scheduling": "WHEN_IDLE",

    # Inbound transcription
    "transcription": {
        "language_auto": True,               # switch language mid-conversation
        "language_hints": ["fr-FR", "en-US"],
        "custom_vocabulary": ["RoomKit", "Aubervilliers"],
        "diarization": True,
    },

    # Generation parameters
    "top_p": 0.8,
    "top_k": 40,
    "max_output_tokens": 2048,

    # Non-interruptible mode
    "no_interruption": True,

    # Language
    "language": "en-US",

    # Pre-3.8 only — dropped with a warning on the 3.8 family
    "proactive_audio": True,
    "enable_affective_dialog": True,
    "thinking_budget": 1024,
}

Feature	Description
Background tool calls	3.8 runs tools while the model keeps speaking
Thinking level	Reasoning depth on extended-thinking, where it is required (RoomKit defaults it to low)
Automatic language switch	language_auto follows the caller mid-call
Custom vocabulary	Bias transcription towards your own terms
Turn coverage	Which input the server bills into a turn
Session resumption	Preserves context across reconfiguration
Non-interruptible	Prevent user barge-in during responses
Proactive audio	Pre-3.8 toggle; permanently on from 3.8
Affective dialog	Pre-3.8 only; removed from the API
Thinking budget	Pre-3.8 only; replaced by thinking_level

Background tool calls

From 3.8 a tool call no longer freezes the conversation. RoomKit declares the tools NON_BLOCKING, so the model can say it is checking and carry on while the work runs. The result's delivery is left to the model: scheduling is sent only if you ask for it, because gemini-3.8-live-extended-thinking refuses the field outright and closes the session over it.

A single tool can opt back into the old behaviour where the model allows it, by carrying behavior in its declaration:

tools = [{"name": "charge_card", "description": "...", "parameters": {...},
          "behavior": "BLOCKING"}]

gemini-3.8-live-extended-thinking answers a hard error to BLOCKING, so RoomKit downgrades the request there and says so in a warning rather than losing the session.

examples/realtime_background_tools_mic.py is the one that makes it obvious: ask for a stock level out loud and you hear the agent keep talking through a six-second lookup, with every turn that overlapped the call marked in the terminal. examples/realtime_background_tools.py proves the same thing without a microphone, for CI and for a machine with no audio device.

A background call can be overtaken by the caller. Talk over the model while the lookup runs and Gemini sends tool_call_cancellation: it will not read the result. RoomKit cancels the handler's task (your handler sees asyncio.CancelledError at its next await), drops any result that still arrives, and reports the call to ON_TOOL_CALL's async observers with cancelled=True beside is_error=True. A call orphaned by a reconnect takes the same path, blocking or not: call ids are connection-scoped, and the new socket never issued them. Driving the provider directly, register provider.on_tool_call_cancelled(callback), called as (session, call_ids). examples/realtime_tool_call_cancelled.py walks the whole path on the mock provider, with no key needed. The other providers have no such event, but fire the callback for the calls they drop themselves: ElevenLabs for a call its tool_timeout_s cuts, GPT-Live for the calls a restart drops, and every provider for the calls still open when its connection is lost or closed.

A call the model could not write

Gemini Live can end a turn on a function call it failed to produce, with turn_complete_reason set to MALFORMED_FUNCTION_CALL: no call arrives, and the turn would end in silence. RoomKit tells the model, as a system injection, that its call did not run and that it should call the tool again with valid arguments or answer without it. It does so once, until the user speaks again, so a model that keeps failing does not loop on the reminder.

Available Voices

Aoede, Fenrir, Kore, Pax, Breeze, Charon, Ember, Orion, Stella, and more.

Session Resumption

Gemini preserves conversation context when reconfigured: reconfigure() reconnects with the session's resumption handle. That is what an agent handoff relies on:

# Start with general assistant
session = await channel.start_session(room_id, participant_id, connection)

# Hand off to specialist — context is preserved
await channel.reconfigure_session(
    session,
    system_prompt="You are a billing specialist.",
    voice="Kore",
    tools=billing_tools,
)

gemini-3.8-live keeps the original instruction

A session resumed on gemini-3.8-live runs under the system instruction it started with: the one a reconfiguration sends is ignored. Measured: an agent started as "Tina" and reconfigured as "Bill" still answers "Tina" when asked its name (gemini-3.1-flash-live-preview and 2.5 take the new instruction). RoomKit covers two cases:

• a session with no conversation yet (nothing sent, nothing heard) reconnects without its handle, so it starts under the new instruction with nothing lost;
• a session with a conversation resumes with its context, and the new instruction rides its next non-silent inject_text, which the model follows at once: the greeting of a ConversationPipeline handoff with greet_on_handoff=True, for one.

After a direct reconfigure_session(system_prompt=...) on a 3.8 session that has a conversation, follow it with a non-silent inject_text(..., role="system") (a greeting, an announcement): until then the model still runs under the old instruction.

---

xAI Grok Realtime

WebSocket-based speech-to-speech using xAI's Grok models with server-side VAD, built-in transcription, and native web/X search tools.

from __future__ import annotations

from roomkit.providers.xai.config import XAIRealtimeConfig
from roomkit.providers.xai.realtime import XAIRealtimeProvider

provider = XAIRealtimeProvider(
    XAIRealtimeConfig(
        api_key="xai-...",
        model="grok-3-fast",                   # Default model
        voice="eve",                           # Default voice
        transcription_model="grok-2-audio",    # Input transcription model
    )
)

# Or with keyword arguments:
provider = XAIRealtimeProvider(
    api_key="xai-...",
    model="grok-3-fast",
)

VAD Configuration

provider_config = {
    "turn_detection_type": "server_vad",    # server_vad (default)
    "threshold": 0.5,
    "silence_duration_ms": 800,
    "prefix_padding_ms": 300,
}

Flat keys as on OpenAI, but a narrower set — the four above are all Grok reads, with no eagerness, idle_timeout_ms, interrupt_response or create_response — and a different default: server_vad here, semantic_vad on OpenAI.

Available Voices

eve, ara, rex, sal, leo

Native Tools

xAI supports native web_search and x_search tools alongside standard function tools:

channel = RealtimeVoiceChannel(
    "voice",
    provider=provider,
    transport=transport,
    tools=[
        {"type": "web_search"},                # Search the web
        {"type": "x_search"},                  # Search X (Twitter)
        weather_tool,                          # Standard function tool
    ],
)

Input Transcription

# Override transcription model via provider_config
provider_config = {
    "transcription_model": "grok-2-audio",
}

Environment Variables (Example)

XAI_API_KEY=xai-...  uv run python examples/realtime_voice_local_xai.py
XAI_MODEL=grok-3-fast  # Model override
XAI_VOICE=ara           # Voice override

Mid-session reconfiguration

Tool Search, a skill activation and a handoff reconfigure a live session. xAI does it in band, as OpenAI does, with a partial session.update carrying only what changes (instructions, voice, tools, temperature): the conversation survives. A provider_config key the live session cannot take (turn detection, transcription, audio format) is named in a warning and applies from the next session.

PersonaPlex and Anam take their prompt and persona only when a session opens, so they declare supports_mid_session_reconfigure = False: the channel does not reconnect them for Tool Search or a skill, and their skills are delivered inline.

---

ElevenLabs Conversational AI

Server-orchestrated speech-to-speech using ElevenLabs agents. STT, LLM, TTS, VAD, and turn-taking are all handled server-side — the provider sends and receives audio, and bridges the agent's client tool calls into RoomKit's tool path. A call keeps the service's id: the SDK folds it into the call's parameters, where a parameter the model writes under tool_call_id would replace it, so the provider keeps the model's parameters apart and the id stands; a tool_call_id the model wrote reaches your handler as an argument, judged against the tool's schema as any.

from __future__ import annotations

from roomkit.providers.elevenlabs.config import ElevenLabsRealtimeConfig
from roomkit.providers.elevenlabs.realtime import ElevenLabsRealtimeProvider

config = ElevenLabsRealtimeConfig(
    api_key="xi-...",
    agent_id="agent_abc123",           # From ElevenLabs dashboard
)
provider = ElevenLabsRealtimeProvider(config)

Agent Setup

ElevenLabs agents are pre-configured on the ElevenLabs dashboard with an LLM, voice, knowledge base, and tools. The agent_id identifies which agent to connect to. Runtime overrides for system prompt, voice, and temperature are applied at connection time. The agent's TTS model, Eleven v4 Turbo included, is chosen in its dashboard configuration; RoomKit does not override it.

Audio Format

The ElevenLabs SDK reads and writes 16 kHz mono 16-bit PCM, and neither side of that is negotiable. The provider therefore rejects a channel configured for anything else at connect() time rather than accepting a clock it cannot honour:

channel = RealtimeVoiceChannel(
    "voice",
    provider=provider,
    transport=transport,
    input_sample_rate=16000,      # required
    output_sample_rate=16000,     # required — the 24000 default raises ValueError
)

A transport running at another rate is still fine: set transport_sample_rate and the channel resamples around the provider.

connect() also waits until the SDK has installed its audio-input callback before reporting the session active, so speech arriving in the first moments of a call is buffered and delivered instead of dropped on the floor.

Configuration Overrides

Override agent defaults via channel parameters. Provider-specific settings (language, first message, dynamic variables) are passed via session metadata["provider_config"]:

channel = RealtimeVoiceChannel(
    "voice",
    provider=provider,
    transport=transport,
    system_prompt="You are a helpful assistant.",   # Override agent prompt
    voice="voice-id-from-elevenlabs",               # Override agent voice
    temperature=0.7,                                # Override LLM temperature
)

# Provider-specific overrides via session metadata
session = await channel.start_session(
    room_id="room-1",
    participant_id="user-1",
    metadata={
        "provider_config": {
            "language": "fr",                       # Language code
            "first_message": "Bonjour!",            # Agent's opening message
            "dynamic_variables": {                  # Template variables for the prompt
                "user_name": "Alice",
                "account_id": "12345",
            },
        },
    },
)

Parameter	Where	Description
system_prompt	Channel	Override the agent's system prompt
voice	Channel	ElevenLabs voice ID (overrides agent default)
temperature	Channel	LLM sampling temperature
language	metadata["provider_config"]	Language code (e.g. en, fr, ja, es)
first_message	metadata["provider_config"]	Agent greeting message
dynamic_variables	metadata["provider_config"]	Dict of variables for prompt templates

Authentication

Two authentication modes:

# Direct API key (server-to-server) — default
config = ElevenLabsRealtimeConfig(
    api_key="xi-...",
    agent_id="agent_abc123",
    requires_auth=False,              # API key sent as header
)

# Signed URL (client-facing deployments)
config = ElevenLabsRealtimeConfig(
    api_key="xi-...",
    agent_id="agent_abc123",
    requires_auth=True,               # Fetches signed URL via SDK
)

Regional Endpoints

# EU (GDPR)
config = ElevenLabsRealtimeConfig(
    api_key="xi-...",
    agent_id="agent_abc123",
    base_url="wss://api.eu.residency.elevenlabs.io",
)

Tool Calling

Tool calling takes a different route than on OpenAI or Gemini, and it takes both halves to work:

1. The agent declares the tool as a client tool (ElevenLabs dashboard, or the Agents API), with its name, description and JSON schema. This is what the LLM sees, and no schema ever travels over the WebSocket.
2. The channel declares the same names in tools, so the provider registers a handler for each one on the SDK's ClientTools registry.

The agent then invokes the tool, the provider dispatches through the standard on_tool_call callback, and the value your tool_handler returns is sent back as the tool result — hooks, gates and result truncation all apply exactly as on the other providers.

async def handle_tool(name: str, arguments: dict) -> str:
    if name == "check_order":
        return json.dumps({"status": "shipped", "eta": "Tomorrow"})
    return json.dumps({"error": f"Unknown tool: {name}"})


channel = RealtimeVoiceChannel(
    "voice",
    provider=provider,
    transport=transport,
    # Names must match the client tools declared on the agent.
    tools=[{"name": "check_order", "description": "Order status", "parameters": {}}],
    tool_handler=handle_tool,
)

Names must match on both sides

A tool the agent knows but the channel did not declare reaches the channel all the same, as on every other provider. A channel that declares tools refuses it as undeclared: the agent reads the error, and ON_TOOL_CALL hears of it. A channel that declares none has no list to hold it against, so its gate judges the call alone and its tool_handler runs it, under any name the agent sends. A tool the channel declared but the agent does not know is never called. The names are case-sensitive.

A call is bounded by the channel (tool_timeout_seconds, 10 s by default on RealtimeVoiceChannel, and tool_timeouts per tool): past it the handler is cancelled and the agent reads the failure rather than its turn hanging. A refused, failed, timed-out or unserved call reaches the agent as a tool error (is_error), not as a result it could take for a success. tool_timeout_s is off by default; set it only to cap the channel's bound for this provider: a call cut there is reported cancelled, and the channel cancels its handler. A call still waiting when the conversation ends is reported the same way.

No mid-session reconfigure

supports_mid_session_reconfigure is False: ConvAI takes its overrides once, in the initiation message, and reconnecting would start a different conversation server-side — losing the transcript and every pending tool_call_id. Tool and skill surfaces are therefore fixed for the life of the session. Per-conversation tool sets are possible the ElevenLabs way, by creating the tools through the Agents API and passing their tool_ids in the prompt override (which the agent's Security settings must allow).

Session Failures

start_session only spawns the task that opens the WebSocket, so a rejected key, an unknown agent_id or a dropped connection surfaces asynchronously. The provider supervises the session and reports these through on_error (connection_failed, session_ended) with the session marked ENDED, instead of leaving a silent session that looks active.

Audio Format

ElevenLabs uses 16-bit PCM mono at 16 kHz by default.

Supported formats: pcm_8000, pcm_16000, pcm_22050, pcm_24000, pcm_44100, pcm_48000, ulaw_8000.

The format lives on the agent, not on the session

Input and output formats are part of the agent's configuration and cannot be overridden per conversation, so input_sample_rate / output_sample_rate on the channel must match what the agent is configured for. A mismatch is not an error — it is pitched-up or pitched-down audio.

Turn Boundaries

ConvAI sends its agent text before the synthesis and marks no end of audio, so the provider opens a response on the first audio chunk and closes it once the audio stream has been quiet for response_idle_ms (default 800 ms). A tool call in flight holds the turn open, since the agent resumes speaking on the same turn once it has the result.

Environment Variables (Example)

ELEVENLABS_API_KEY=xi-...  ELEVENLABS_AGENT_ID=agent_abc123 \
    uv run python examples/realtime_voice_local_elevenlabs.py

# Optional overrides
ELEVENLABS_VOICE_ID=voice-id    # Voice override
SYSTEM_PROMPT="Be concise."     # System prompt override
LANGUAGE=fr                     # Language code

---

Deepgram Voice Agent

Deepgram assembles an agent from three stages you choose independently — listen (Nova/Flux transcription), think (the LLM) and speak (an Aura voice, or another vendor's — see Non-Deepgram voices) — instead of one end-to-end model. That is the reason to reach for it: you can put a self-hosted LLM behind a hosted STT/TTS pair, or an ElevenLabs voice behind Deepgram's turn-taking, without touching the rest.

from __future__ import annotations

from roomkit.providers.deepgram import DeepgramAgentConfig, DeepgramAgentProvider

config = DeepgramAgentConfig(
    api_key="...",
    listen_model="nova-3",              # STT
    think_model="gpt-4o-mini",          # LLM (served by Deepgram)
    speak_model="aura-2-thalia-en",     # TTS voice
    greeting="Hi! What can I do for you?",
)
provider = DeepgramAgentProvider(config)

The provider first waits for Deepgram's server-side Welcome, then sends one Settings message and waits for SettingsApplied before the session goes ACTIVE, so a rejected key or a bad model fails in connect() rather than silently later.

Configuration

Field	Default	Description
api_key	required	Sent as Authorization: Token <key>
base_url	wss://agent.deepgram.com/v1/agent/converse	Use wss://api.eu.deepgram.com/v1/agent/converse for EU processing
listen_model	nova-3	Transcription model
listen_version	None	Required by Flux ("v2"); leave unset for Nova
think_provider / think_model	open_ai / gpt-4o-mini	LLM stage
speak_model	aura-2-thalia-en	Aura voice id
speak_provider	None	Full agent.speak.provider dict, sent verbatim — selects a non-Deepgram TTS vendor with that vendor's own fields. Takes precedence over speak_model/speak_language
speak_endpoint	None	agent.speak.endpoint (URL + auth headers) — Deepgram requires one for every non-Deepgram speak_provider
greeting	None	Line the agent speaks as soon as the session opens
keepalive_interval	8.0	Seconds between KeepAlive messages — Deepgram closes silent sockets
max_prompt_chars	25_000	Warn when the system prompt exceeds this — Deepgram's documented cap for managed LLMs, past which it truncates (PROMPT_TOO_LONG). None disables; never fires with a think_endpoint (no cap there)

Everything above can be overridden per session through provider_config, alongside a few keys with no config equivalent:

Key	Description
listen_model, listen_version, listen_language, keyterms, smart_format	STT stage
think_provider, think_model, think_endpoint, context_length	LLM stage — think_endpoint points at your own OpenAI-compatible server
speak_model, speak_language, speak_provider, speak_endpoint	TTS stage — speak_provider/speak_endpoint swap in a non-Deepgram vendor (see below)
greeting, tags	Session greeting; dashboard labels
max_prompt_chars	Per-session override of the prompt-size warning threshold
input_encoding, output_encoding, output_container, output_bitrate	Audio codecs (see below)
settings	Deep-merged into the final Settings payload, last — escape hatch for fields the provider does not model. A nested dict naming a different provider type replaces the built one instead of merging into it — fields from two vendors never blend

Non-Deepgram Voices (ElevenLabs, Cartesia…)

The speak stage accepts five provider types — deepgram, eleven_labs, cartesia, open_ai and aws_polly — and each has its own field shape (model vs model_id/voice_id vs voice objects). speak_provider therefore carries the full agent.speak.provider dict verbatim rather than modelling every vendor:

config = DeepgramAgentConfig(
    api_key="...",
    speak_provider={
        "type": "eleven_labs",
        "model_id": "eleven_turbo_v2_5",
    },
    speak_endpoint={
        "url": "wss://api.elevenlabs.io/v1/text-to-speech/<voice-id>/multi-stream-input",
        "headers": {"xi-api-key": "..."},
    },
)

Or per session — including a mid-session vendor swap through reconfigure():

provider_config = {
    "speak_provider": {
        "type": "cartesia",
        "model_id": "sonic-2",
        "voice": {"mode": "id", "id": "a167e0f3-df7e-4d52-a9c3-f949145efdab"},
    },
}

BYO-key vendors need an endpoint

ElevenLabs runs on your own key: speak_endpoint carries the vendor URL (the ElevenLabs voice id rides in the URL, not the provider dict) plus your API key in its headers. Deepgram-managed vendors — Cartesia above — need no endpoint. Check Deepgram's TTS provider docs for each vendor's exact shape; RoomKit passes both dicts through verbatim.

speak_provider takes precedence over voice and speak_model, which name Aura voices: a voice argument passed while another vendor holds the stage is ignored with a warning, since that vendor's voice belongs in its own fields.

Telephony

Deepgram accepts mulaw at 8 kHz in both directions, so a SIP or Twilio transport needs no resampling on either leg:

session = await channel.start_session(
    room_id="room-1",
    participant_id="caller-1",
    metadata={
        "provider_config": {
            "input_encoding": "mulaw",
            "output_encoding": "mulaw",
        },
    },
)

Set input_sample_rate=8000 and output_sample_rate=8000 on the channel to match.

Tool Calling

Tools declared on the channel are sent as think.functions and come back through the standard on_tool_call path — no dashboard setup, unlike ElevenLabs:

channel = RealtimeVoiceChannel(
    "voice",
    provider=provider,
    transport=transport,
    tools=[{"name": "check_order", "description": "Order status", "parameters": {...}}],
    tool_handler=handle_tool,
)

A function carrying an endpoint key is called by Deepgram server-side and never reaches RoomKit; without one it arrives as a client-side call for your tool_handler to answer.

When a Gemini thinking model attaches a thought_signature to a function call, RoomKit preserves it and returns it unchanged with the function result. No special handling is required in the tool handler.

Runnable end to end in examples/realtime_deepgram_tools.py.

Mid-Session Reconfigure

reconfigure() is overridden to patch the live session instead of reconnecting: UpdateThink carries the new prompt, model and functions in one message, UpdateSpeak swaps the voice. An agent handoff therefore keeps the WebSocket and the conversation context.

Both messages replace the whole Think/Speak block on Deepgram's side, so the patch is computed from the live session state rather than from provider defaults: a prompt-only handoff preserves the functions, the per-session model, endpoint, temperature and context length already in force. A provider_config-only change (think_model, think_endpoint, context_length, speak_model, speak_language) is applied on its own, without also needing a new prompt or voice.

Barge-In

Deepgram signals barge-in the other way round from most providers: there is no client-side interrupt message. When the caller speaks over the agent, Deepgram sends UserStartedSpeaking, which the provider surfaces as on_speech_start — and that is what makes the channel flush playback, reset the outbound resampler and send clear_audio to the client. interrupt() therefore only clears local state; it sends nothing.

Deepgram also has no "user stopped speaking" event: the user's ConversationText transcript is the end of their turn, and the provider fires on_speech_end on it so idle detection keeps working.

Known Limitations

• Turn detection is always Deepgram's. server_vad=False is not supported; the provider logs a warning and ignores it.
• Transcriptions are final-only. ConversationText carries no interim results.
• Sessions are capped at two hours. A MAXIMUM_SESSION_LENGTH_APPROACHING warning arrives at 1 h 55 and a terminal error at 2 h, both surfaced through on_error.
• Silent injection rewrites the prompt. Deepgram has no message that adds to the conversation without a reply, so inject_text(..., silent=True) appends the text to the system prompt via UpdatePrompt — additive, and only a role="system" text joins the instructions: any other intent is content, appended in a <context> block of its own so it never reads as more of them. UpdatePrompt never starts a turn, so a non-silent role="system" instruction travels as InjectUserMessage, which the agent acts on at once; a standing instruction ("speak more slowly from now on") is role="system", silent=True. role="assistant" is InjectAgentMessage: the agent says the line verbatim.
• Managed-LLM prompts are capped at 25,000 characters. Past the cap Deepgram truncates the prompt and keeps the session alive (a non-fatal PROMPT_TOO_LONG warning). RoomKit warns client-side before sending — at connect, on reconfigure(), and as silent injections grow the prompt — governed by max_prompt_chars. Bring-your-own think_endpoint sessions have no cap and are never warned.

Local speaker and microphone example

The local Deepgram example enables WebRTC AEC by default so the microphone can stay open while the agent speaks:

DEEPGRAM_API_KEY=... \
    uv run python examples/realtime_voice_local_deepgram.py

Its audio-specific environment variables are:

Variable	Default	Description
AEC	webrtc	webrtc, speex, or 0 to disable echo cancellation
DENOISE	webrtc	webrtc, rnnoise, sherpa, or 0 to disable continuous noise suppression
AEC_DELAY_MS	0	Fixed speaker-to-microphone delay in milliseconds; 0 lets the local backend give AEC3 the delay its devices' clocks report, on every captured frame
AUDIO_PREBUFFER_MS	240	Local speaker jitter buffer used by this Deepgram example
MUTE_MIC	automatic	0 keeps capture open; 1 mutes capture during playback. Automatic mode mutes only when AEC is unavailable

These variables are conveniences implemented by the example, not global RoomKit settings. In application code, their equivalents are WebRTCAECProvider(stream_delay_ms=...) and WebRTCNoiseSuppressorProvider(...), plus LocalAudioBackend(rt_prebuffer_ms=...).

Start with AEC_DELAY_MS=0. If the speaker stream reports underruns, increase AUDIO_PREBUFFER_MS; lowering it reduces startup latency but gives bursty network audio less protection. The library default for rt_prebuffer_ms is 120 ms, while this Deepgram example uses 240 ms because Aura audio commonly arrives in network bursts.

Voices

DeepgramAgentProvider.available_voices() returns the full Aura-2 catalog — English, Spanish, Dutch, French, German, Italian and Japanese — plus the twelve Aura-1 voices flagged deprecated=True. There is no live voices endpoint, so list_voices() returns the same curated list.

Audio Transports

Realtime channels need a transport to carry audio between the client and server.

WebSocket Transport

from __future__ import annotations

from roomkit.voice.realtime.ws_transport import WebSocketRealtimeTransport

transport = WebSocketRealtimeTransport(
    authenticate=my_auth_callback,             # Optional auth
    audio_format="binary",                     # default; "base64_json" for legacy clients
)

Audio is PCM16 little-endian mono: the client sends it at the channel's input rate and receives it at the output rate (16 kHz in, 24 kHz out with Gemini Live). By default both directions carry it as raw binary frames; with audio_format="base64_json" the server wraps it in JSON, and a client may always send the JSON form.

Client messages: binary audio frames, or

{"type": "audio", "data": "<base64 PCM>"}

Server messages: binary audio frames (or {"type": "audio", "data": "<base64>"} with base64_json), and

{"type": "session_started"}
{"type": "transcription", "text": "Hello", "role": "user", "is_final": true}
{"type": "speaking", "speaking": true, "who": "assistant"}
{"type": "clear_audio"}
{"type": "session_ended"}

The transport iterates a websockets server connection; see examples/realtime_voice_gemini.py for a minimal server and browser page.

FastRTC WebRTC Transport

Browser-based WebRTC with low latency:

from __future__ import annotations

from roomkit.voice.realtime.fastrtc_transport import (
    FastRTCRealtimeTransport,
    mount_fastrtc_realtime,
)

transport = FastRTCRealtimeTransport(
    input_sample_rate=16000,
    output_sample_rate=24000,
)

# Mount on FastAPI app
mount_fastrtc_realtime(
    app,
    transport,
    path="/rtc-realtime",
    auth=my_auth_callback,                     # Optional
)

Audio codec: mu-law (8-bit) over WebRTC DataChannel.

Auto-session: Clients connect via WebRTC, transport fires on_client_connected callback.

SIP Transport

Bridge SIP calls to realtime AI:

from __future__ import annotations

from roomkit.voice.realtime.sip_transport import SIPRealtimeTransport

transport = SIPRealtimeTransport(backend=sip_backend)

Sample rate: Negotiated via SIP codec (G.711 @ 8kHz, G.722 @ 16kHz). The channel auto-resamples.

Audio pacing: Built-in OutboundAudioPacer with ~80ms pre-buffer and jitter absorption.

Local Audio (Development)

Use system mic/speakers for testing:

from __future__ import annotations

from roomkit.voice.backends.local import LocalAudioBackend

transport = LocalAudioBackend(
    input_sample_rate=24000,   # match the provider output rate for AEC
    output_sample_rate=24000,
    rt_prebuffer_ms=120,
    aec=aec_provider,
)

Audio pacing: rt_prebuffer_ms (default 120ms) primes the speaker buffer before playback starts and re-primes after an underrun — the local-speaker analogue of the SIP pacer's pre-buffer. It paces streamed TTS the same way: every response plays through the backend's one speaker stream. Set it to 0 to play from the first byte. The rt_underruns property counts mid-response starvations.

When aec is supplied, the local backend uses the blocks actually written to the speaker as the reference, including silence inserted after playback has started. This keeps the render and microphone timelines aligned across an underrun and avoids the agent's output being transcribed back as user speech.

---

Tool Calling

Via Tool Objects (Recommended)

Pass Tool objects directly — the channel extracts definitions and handlers automatically:

from __future__ import annotations

import json

from roomkit import RealtimeVoiceChannel, Tool


async def get_weather(city: str) -> str:
    return json.dumps({"temperature": 72, "condition": "sunny", "city": city})


weather_tool = Tool(
    name="get_weather",
    description="Get current weather for a city",
    parameters={
        "type": "object",
        "properties": {
            "city": {"type": "string", "description": "City name"},
        },
        "required": ["city"],
    },
    handler=get_weather,
)

channel = RealtimeVoiceChannel(
    "voice",
    provider=provider,
    transport=transport,
    tools=[weather_tool],
    mute_on_tool_call=True,          # Prevent barge-in during tool execution
)

Via tool_handler Callback (Advanced)

For MCP integration, auditing wrappers, or custom dispatch logic, use tool_handler directly:

async def handle_tool(name, arguments):
    if name == "get_weather":
        city = arguments.get("city", "Unknown")
        return json.dumps({"temperature": 72, "condition": "sunny", "city": city})
    return json.dumps({"error": f"Unknown tool: {name}"})


channel = RealtimeVoiceChannel(
    "voice",
    provider=provider,
    transport=transport,
    tools=[
        {
            "type": "function",
            "name": "get_weather",
            "description": "Get current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name"},
                },
                "required": ["city"],
            },
        },
    ],
    tool_handler=handle_tool,
    mute_on_tool_call=True,
)

Via Hooks

The unified ON_TOOL_CALL hook fires from both AIChannel and RealtimeVoiceChannel. Use event.channel_type to distinguish the source. Return the result via HookResult.metadata["result"].

from __future__ import annotations

from roomkit import HookTrigger, RoomKit
from roomkit.models.hook import HookResult

kit = RoomKit()


@kit.hook(HookTrigger.ON_TOOL_CALL)
async def on_tool_call(event, ctx):
    if event.name == "get_weather":
        result = await fetch_weather(event.arguments["city"])
        return HookResult(action="allow", metadata={"result": result})
    return HookResult.allow()

Mute on Tool Call

Set mute_on_tool_call=True when using Gemini Live — it prevents the user from speaking during tool execution, which can cause Gemini to drop tool results.

---

When the Model Speaks the Call Instead of Issuing It

Realtime models sometimes say a tool call rather than emit it through the function calling API — Gemini Live most often, usually under load or after a long turn:

call:get_weather{city:Montreal}

tool_recovery (on by default) recognises that shape in an assistant transcription, parses the arguments, runs the tool, and suppresses the text so the caller never hears it. Speech before the call is kept and still reaches the room; a transcription that is only a call produces no room event.

Only a call said as a sentence of its own that ends the utterance is one: at the start of the transcription or of a line, or after a sentence's end (Let me check. call:get_weather{city:Montreal}), with nothing after its closing brace but a final stop. A sentence that mentions the form (You can type call:get_weather{city:Montreal} to ask me.) is speech, and so is a call followed by more speech.

The recovered call is a tool call like any other:

• it passes the same pre-execution gate — the declared catalogue, the tool policy, skill gating, the argument schema and BEFORE_TOOL_USE, so a hook that denies it prevents the side effect rather than reporting it afterwards;
• it fires ON_TOOL_CALL through the same dispatch, so a serving hook or a handler answers it normally;
• its outcome — result or refusal — returns as injected context, never as a tool result, because the model issued no call and has no FunctionResponse waiting on it. It reads the outcome on its next turn. A call that nothing served (no handler, no hook) reads as a failure, never as a success.

Two limits follow from the text being text:

Case	Outcome
call:lookup{city:Paris,limit:3} on a tool declaring limit as an integer	3 is coerced to an integer
call:lookup{city:Paris,limit:many}	refused — many is not an integer, and guessing would be worse
a tool declaring an array or object parameter	not invocable this way — only boolean/integer/number are coerced
call:lookup{city:Paris,country:FR} on a closed schema	refused by name — country is undeclared
call:note{text:see you at 3:30}	intact — a colon inside a value is not a key boundary

Set tool_recovery=False to disable the whole path and let a spoken call stay speech.

channel = RealtimeVoiceChannel(
    "voice",
    provider=provider,
    transport=transport,
    tools=[...],
    tool_recovery=False,
)

---

Tool policy

tool_policy= takes the same ToolPolicy as an AIChannel. A tool it denies is never declared to the session (at connection, at a reconfiguration, in a Tool Search reveal or a skill activation), is not in the catalogue a reasoning backend receives, is never named by find_tools, and is refused if a call names it anyway, whichever way the call arrived: from the provider, spoken as call:name{args}, or from a reasoning backend. The refusal reaches ON_TOOL_CALL's observers with is_error. On a provider whose declarations are fixed for the session, Tool Search's call_tool stays declared under an allow list, and the policy applies to the tool it names.

Role overrides apply to the session's participant. Its role is read from the store when the session starts and again at each call, so demoting a caller during the call takes effect at their next tool call; the tools already declared to the session change at its next reconfiguration. Add the participant to the store before start_session(): one the store does not hold has no role, and only the base rules apply to it.

from roomkit import RealtimeVoiceChannel, RoleOverride, ToolPolicy

voice = RealtimeVoiceChannel(
    "voice",
    provider=provider,
    transport=transport,
    tools=[lookup_account, close_account],
    tool_handler=accounts,
    # Everyone may use every tool, but an observer may not close anything.
    tool_policy=ToolPolicy(role_overrides={"observer": RoleOverride(deny=["close_*"])}),
)

A conference's provider takes one too, as ConferenceRealtimeConfig(tool_policy=...); the mix names no participant, so only the base rules apply there, and the channel logs the role overrides it ignores. See examples/realtime_tool_policy.py.

Session Lifecycle

from __future__ import annotations

from roomkit import RealtimeVoiceChannel

# Start a session
session = await channel.start_session(
    room_id="room-1",
    participant_id="user-1",
    connection=webrtc_id,                      # Transport-specific connection handle
    metadata={
        "system_prompt": "Override prompt",     # Per-session override
        "voice": "echo",                        # Per-session voice
    },
)

# Reconfigure mid-conversation
await channel.reconfigure_session(
    session,
    system_prompt="New prompt",
    voice="shimmer",
    tools=new_tools,
    temperature=0.5,
)

# End session
await channel.end_session(session.id)

State machine: CONNECTING → ACTIVE → ENDED

Auto-resampling: If transport_sample_rate differs from input_sample_rate or output_sample_rate, per-session resamplers are created automatically.

---

Hooks

Realtime-specific hooks fired during voice sessions:

Hook	Type	Description
ON_TRANSCRIPTION	Sync	Transcription received (can block/modify)
ON_TOOL_CALL	Sync	Tool call from any channel (return result via metadata)
ON_TOOL_CALL	Async	Audit of every call: served, refused, failed, or abandoned by the model (is_error, cancelled)
ON_SPEECH_START	Async	User started speaking
ON_SPEECH_END	Async	User stopped speaking
ON_SESSION_STARTED	Async	Voice session activated
ON_INPUT_AUDIO_LEVEL	Async	Input audio level (~10/sec)
ON_OUTPUT_AUDIO_LEVEL	Async	Output audio level (~10/sec)

from __future__ import annotations

from roomkit import HookTrigger, RoomKit

kit = RoomKit()


@kit.hook(HookTrigger.ON_TRANSCRIPTION)
async def on_transcription(event, ctx):
    print(f"[{event.metadata['role']}] {event.content.body}")


@kit.hook(HookTrigger.ON_SPEECH_START)
async def on_speech_start(event, ctx):
    print("User started speaking")

---

Access Control

The channel enforces ChannelBinding permissions (RFC Section 7.5):

• Access revoked: Audio from the client is silently dropped
• Muted: Audio from the client is silently dropped
• Active: Audio flows normally

Permissions are checked on every audio frame — changes take effect immediately.

---

Complete Example: FastRTC + Gemini

from __future__ import annotations

import json

from fastapi import FastAPI

from roomkit import RealtimeVoiceChannel, RoomKit, Tool
from roomkit.providers.gemini.realtime import GeminiLiveProvider
from roomkit.voice.realtime.fastrtc_transport import (
    FastRTCRealtimeTransport,
    mount_fastrtc_realtime,
)

app = FastAPI()
kit = RoomKit()

provider = GeminiLiveProvider(
    api_key="your-gemini-key",
    model="gemini-3.8-live",
)

transport = FastRTCRealtimeTransport(
    input_sample_rate=16000,
    output_sample_rate=24000,
)


async def lookup_order(order_id: str) -> str:
    return json.dumps({"status": "shipped", "eta": "Tomorrow"})


order_tool = Tool(
    name="lookup_order",
    description="Look up an order by ID",
    parameters={
        "type": "object",
        "properties": {"order_id": {"type": "string"}},
        "required": ["order_id"],
    },
    handler=lookup_order,
)

channel = RealtimeVoiceChannel(
    "voice",
    provider=provider,
    transport=transport,
    system_prompt="You are a customer service agent. Be helpful and concise.",
    voice="Aoede",
    tools=[order_tool],
    mute_on_tool_call=True,
    emit_transcription_events=True,
)

kit.register_channel(channel)

# Auto-create session on WebRTC connect
transport.on_client_connected = lambda webrtc_id: channel.start_session(
    room_id="room-1",
    participant_id="caller",
    connection=webrtc_id,
)

mount_fastrtc_realtime(app, transport, path="/rtc")

Testing with Mocks

from __future__ import annotations

from roomkit import RealtimeVoiceChannel
from roomkit.voice.realtime.mock import MockRealtimeProvider, MockRealtimeTransport

provider = MockRealtimeProvider()
transport = MockRealtimeTransport()

channel = RealtimeVoiceChannel("voice-test", provider=provider, transport=transport)

session = await channel.start_session("room-1", "user-1")

# Simulate provider events
await provider.simulate_transcription(session, "Hello", role="user", is_final=True)
await provider.simulate_audio(session, b"\x00\x01" * 100)
await provider.simulate_tool_call(session, "call-1", "get_weather", {"city": "NYC"})

# Assert transport received audio
assert len(transport.sent_audio) > 0

Sharing a transport between channels

When each caller needs a separate provider or prompt, channels can share one FastRTCRealtimeTransport. Pass owns_transport=False to each channel:

channel = RealtimeVoiceChannel(
    "voice-user-123",
    provider=provider,
    transport=shared_transport,
    owns_transport=False,
)

await channel.close() ends that channel's sessions, closes its provider and removes its transport callbacks. Other channels remain connected. The service that created the shared transport calls await shared_transport.close() once at shutdown. The default owns_transport=True preserves exclusive ownership.

FastRTC and SIP realtime audio/disconnect registrations return an idempotent unsubscribe function. Invoke it when disposing a custom subscriber. SIP adapters use SIPVoiceBackend.subscribe_audio_received() to listen alongside the primary pipeline callback; await adapter.close() detaches the adapter without closing the SIP listener. Custom shared backends must likewise support unsubscribe functions for their audio, playback and disconnect registrations.

FastRTC declares capture and playback rates independently in session metadata (transport_sample_rate and transport_output_sample_rate). The channel resamples each direction to its provider's configured format. SIP declares the negotiated codec rate. Configure provider rates using the provider's accepted formats; for OpenAI PCM, use 24 kHz rather than Gemini's usual 16 kHz input.

await channel.wait_idle(room_id) waits for generation to finish and audio to reach the transport. A tool call holds it until the response that continues its result; a call that owes no result (cancelled by the model, or spared by the reconnect its own handler caused) stops holding it when it ends. A queued SIP transport may still be playing: before a conversational hangup, also wait until backend.is_playing(carrier_session) is false. Bound this wait and stop it when the call disconnects. SIP reports RTP emission and an estimated playback boundary; remote speaker playback cannot be observed directly.