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 during tool execution
`tool_result_max_length`	`16384`	Max chars for tool results

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 the catalogue exceeds tool_search_threshold (20 by default). Set tool_search=True to enable it for a smaller catalogue or False to expose the catalogue directly. tool_search_pinned keeps selected tools directly callable alongside discovery.

Providers that support mid-session reconfiguration receive native declarations for the tools matched by find_tools. 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 catalogue overview. 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.

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.

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 seed AEC3 from actual PortAudio latencies
`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 remains active while AEC is bypassed between turns.

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 now obtains both actual stream latencies from PortAudio and supplies their sum to AEC3 before playback.

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.

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",
}

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-2.5-flash-native-audio-preview-12-2025",
)

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,

    # Proactive audio (AI speaks without prompt)
    "proactive_audio": True,

    # Affective dialog (emotional responses)
    "enable_affective_dialog": True,

    # Extended thinking
    "thinking_budget": 1024,

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

    # Non-interruptible mode
    "no_interruption": True,

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

Feature	Description
Proactive audio	AI can initiate speech without user prompt
Affective dialog	Emotional, expressive responses
Thinking budget	Extended reasoning before responding
Session resumption	Preserves context across reconfiguration
Non-interruptible	Prevent user barge-in during responses

Available Voices¶

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

Session Resumption¶

Gemini preserves conversation context when reconfigured — useful for agent handoff:

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

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

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

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.

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.

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:

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.
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 comes back to the agent as an error; a tool the channel declared but the agent does not know is never called. The names are case-sensitive.

A call left unanswered for tool_timeout_s (default 30 s) is reported to the agent as an error rather than hanging its turn.

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, but it lands as an instruction rather than as a turn.
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` leaves WebRTC delay estimation enabled
`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="base64_json",                # or "binary"
)

Client messages:

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

Server messages:

{"type": "audio", "data": "<base64>"}
{"type": "transcription", "text": "Hello", "role": "user", "is_final": true}
{"type": "speaking", "speaking": true, "who": "assistant"}
{"type": "clear_audio"}

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

The recovered call is a tool call like any other:

it passes the same pre-execution gate — the declared catalogue, the argument schema, skill gating 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.

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,
)

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.id,
    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_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-2.5-flash-native-audio-preview-12-2025",
)

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

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

Realtime Voice (Speech-to-Speech)¶

How It Differs from VoiceChannel¶

Quick Start¶

RealtimeVoiceChannel Configuration¶

Tool Search with fixed declarations¶

Passing provider_config¶

Which model is speaking¶

OpenAI Realtime API¶

Images¶

VAD Configuration¶

Local speaker and microphone example¶

WebSocket interruption context¶

Available Voices¶

Input Transcription¶

Google Gemini Live¶

Advanced Configuration¶

Available Voices¶

Session Resumption¶

xAI Grok Realtime¶

VAD Configuration¶

Available Voices¶

Native Tools¶

Input Transcription¶

Environment Variables (Example)¶

ElevenLabs Conversational AI¶

Agent Setup¶

Audio Format¶

Configuration Overrides¶

Authentication¶

Regional Endpoints¶

Tool Calling¶

Session Failures¶

Audio Format¶

Turn Boundaries¶

Environment Variables (Example)¶

Deepgram Voice Agent¶

Configuration¶

Non-Deepgram Voices (ElevenLabs, Cartesia…)¶

Telephony¶

Tool Calling¶

Mid-Session Reconfigure¶

Barge-In¶

Known Limitations¶

Local speaker and microphone example¶

Voices¶

Audio Transports¶

WebSocket Transport¶

FastRTC WebRTC Transport¶

SIP Transport¶

Local Audio (Development)¶

Tool Calling¶

Via Tool Objects (Recommended)¶

Via tool_handler Callback (Advanced)¶

Via Hooks¶

When the Model Speaks the Call Instead of Issuing It¶

Session Lifecycle¶

Hooks¶

Access Control¶

Complete Example: FastRTC + Gemini¶

Testing with Mocks¶

Sharing a transport between channels¶

Passing `provider_config`¶