Realtime API
WebSocket-based bidirectional streaming for low-latency voice agents and live conversational AI.
Overview
The Realtime API is a stateful WebSocket connection that streams audio and text in both directions with sub-200ms first-byte latency. It is the right fit for:
- Voice agents (phone bots, ChatGPT-style voice mode)
- Live captioning with low latency
- Streaming tool execution while audio is still flowing
- Interruption-aware conversations (talk over the assistant to redirect it)
For text-only use cases, prefer the standard Chat Completions endpoint with stream: true -- it's simpler and equally fast for text.
Connection
Connect to:
wss://api.allroutes.ai/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01
Authenticate by setting the Authorization header on the WebSocket upgrade request, or by sending a ?token=... query parameter (use this only over WSS, never WS).
# wscat (CLI)
wscat -c "wss://api.allroutes.ai/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01" \
-H "Authorization: Bearer allroutes_sk_..."
import websockets
import json
async with websockets.connect(
"wss://api.allroutes.ai/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01",
extra_headers={"Authorization": "Bearer allroutes_sk_..."},
) as ws:
...
const ws = new WebSocket(
"wss://api.allroutes.ai/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01",
{ headers: { Authorization: `Bearer ${process.env.ALLROUTES_API_KEY}` } }
);
Supported Models
| Model | Provider | Modalities |
|---|---|---|
gpt-4o-realtime-preview-2024-10-01 | OpenAI | text, audio |
gemini-2.0-flash-realtime | text, audio, video | |
claude-3-5-realtime-preview | Anthropic | text, audio |
Event Protocol
The protocol is JSON message-passing. Every message has a type and (optionally) an event_id. There are roughly 30 event types; the most important ones below.
Client -> Server Events
| Event | Purpose |
|---|---|
session.update | Set system prompt, voice, modalities, tools |
input_audio_buffer.append | Append base64-encoded PCM16 audio chunk |
input_audio_buffer.commit | Mark end of user turn |
conversation.item.create | Insert text or tool-call result |
response.create | Trigger model response (manual mode) |
response.cancel | Interrupt the in-flight response |
Server -> Client Events
| Event | Purpose |
|---|---|
session.created | Initial session state |
input_audio_buffer.speech_started | VAD detected user speech |
input_audio_buffer.speech_stopped | VAD detected end of speech |
response.audio.delta | Incremental output audio (base64 PCM16) |
response.audio_transcript.delta | Incremental transcript of the assistant's audio |
response.text.delta | Incremental text output |
response.function_call_arguments.delta | Incremental tool-call arguments |
response.done | Response completed |
error | Recoverable error; session continues |
Minimal Voice Loop (Python)
import asyncio
import base64
import json
import websockets
import sounddevice as sd
import numpy as np
URL = "wss://api.allroutes.ai/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01"
async def main():
async with websockets.connect(URL, extra_headers={"Authorization": f"Bearer {KEY}"}) as ws:
# Configure session
await ws.send(json.dumps({
"type": "session.update",
"session": {
"modalities": ["text", "audio"],
"voice": "alloy",
"instructions": "You are a helpful voice assistant. Be concise.",
"turn_detection": {"type": "server_vad"},
},
}))
async def send_mic():
with sd.InputStream(samplerate=24000, channels=1, dtype="int16") as stream:
while True:
data, _ = stream.read(2400) # 100ms chunks
pcm16 = data.tobytes()
await ws.send(json.dumps({
"type": "input_audio_buffer.append",
"audio": base64.b64encode(pcm16).decode(),
}))
async def play_speaker():
with sd.OutputStream(samplerate=24000, channels=1, dtype="int16") as stream:
async for raw in ws:
msg = json.loads(raw)
if msg["type"] == "response.audio.delta":
pcm16 = base64.b64decode(msg["delta"])
stream.write(np.frombuffer(pcm16, dtype=np.int16))
await asyncio.gather(send_mic(), play_speaker())
asyncio.run(main())
Server-Side VAD vs. Manual Mode
Two turn-detection modes:
server_vad(default) -- the gateway runs a voice-activity detector and triggers responses automatically when the user stops speaking. Use this for hands-free voice agents.none-- the client decides when to commit the buffer (input_audio_buffer.commit) and trigger generation (response.create). Use this for push-to-talk UIs.
Switch modes by setting session.turn_detection.type.
Tool Calling
Tools work the same way as in chat completions, but arguments stream incrementally via response.function_call_arguments.delta. After response.done, send the result back as a conversation.item.create of type function_call_output.
{
"type": "session.update",
"session": {
"tools": [{
"type": "function",
"name": "get_weather",
"description": "Get current weather",
"parameters": {
"type": "object",
"properties": {"location": {"type": "string"}},
"required": ["location"]
}
}]
}
}
Interruptions
To barge in on the assistant mid-response, send response.cancel. The model immediately stops generating, and the unspoken portion of audio is discarded. Most production voice agents wire this to either a "press to interrupt" button or VAD that detects user speech overlapping output.
Pricing
Realtime models bill audio tokens at a higher rate than text:
| Model | Audio input ($/1M tok) | Audio output ($/1M tok) |
|---|---|---|
gpt-4o-realtime-preview | $100 | $200 |
gemini-2.0-flash-realtime | $0.50 | $2.00 |
Use the Models API for live numbers.
Troubleshooting
| Symptom | Cause | Fix |
|---|---|---|
| Connection drops every 60s | Idle timeout | Send input_audio_buffer.append keepalives or upgrade plan |
| Audio is choppy | Network jitter | Increase chunk size from 100ms to 200-500ms |
| VAD never triggers | Mic gain too low | Test with silence_duration_ms: 200 and louder input |
| Cannot interrupt | Forgot response.cancel | Wire button or VAD to send the cancel event |
See Also
- Audio API -- non-realtime TTS and STT
- Chat Completions -- text-only streaming alternative
- Streaming -- SSE-based unidirectional streaming
- Observability -- trace WebSocket sessions in Langfuse/Datadog