The previous lesson laid out the architectural case for native speech-to-speech: fewer hand-offs, better prosody, less transcript-level control. This lesson makes that concrete by building against OpenAI’s Realtime API, the production entry point for gpt-realtime, walking through session setup, bidirectional audio streaming, and function calling inside a single persistent connection, the same three concerns that shaped every cascaded pipeline earlier in this course, now handled by one model instead of three.
Opening the session
The Realtime API is a WebSocket protocol at wss://api.openai.com/v1/realtime, authenticated the same way as other OpenAI API calls, that stays open for the duration of a conversation rather than issuing one request per turn. Everything, audio input, audio output, function calls, and control events, flows as JSON events over that single connection.
import asyncio
import json
import base64
import websockets
REALTIME_URL = "wss://api.openai.com/v1/realtime?model=gpt-realtime"
async def open_session(api_key: str):
headers = {"Authorization": f"Bearer {api_key}", "OpenAI-Beta": "realtime=v1"}
ws = await websockets.connect(REALTIME_URL, extra_headers=headers)
# Configure voice, turn detection, and available tools before
# streaming any audio. This is the single most important event
# in the session's lifecycle.
await ws.send(json.dumps({
"type": "session.update",
"session": {
"modalities": ["audio", "text"],
"voice": "verse",
"turn_detection": {
"type": "server_vad",
"threshold": 0.5,
"silence_duration_ms": 500,
},
"tools": [
{
"type": "function",
"name": "check_order_status",
"description": "Look up the current status of a customer order by order ID.",
"parameters": {
"type": "object",
"properties": {"order_id": {"type": "string"}},
"required": ["order_id"],
},
}
],
},
}))
return wssession.update is where every product-level decision from earlier lessons gets encoded: which voice the model speaks in, how sensitive server-side voice activity detection is (tuned via threshold and silence_duration_ms, directly relevant to the barge-in behavior from Module 5), and which functions the model is allowed to call mid-conversation. Skipping this step means running on whatever defaults OpenAI has set, which is rarely the right choice for a specific product.
Streaming audio in both directions
Once the session is configured, audio flows as base64-encoded deltas in both directions: the client appends chunks of the user’s microphone audio to an input buffer, and the server emits chunks of generated response audio as they’re produced, mirroring the streaming synthesis pattern from Module 4 but now happening natively inside the model rather than as a separate TTS call.
async def stream_microphone(ws, mic_audio_chunks):
"""Push user audio into the session as it's captured."""
async for chunk in mic_audio_chunks:
await ws.send(json.dumps({
"type": "input_audio_buffer.append",
"audio": base64.b64encode(chunk).decode("utf-8"),
}))
# With server_vad enabled, the API detects turn boundaries itself;
# committing manually is only needed in manual turn-detection mode.
async def receive_events(ws, playback, tool_registry):
async for raw in ws:
event = json.loads(raw)
etype = event["type"]
if etype == "response.audio.delta":
audio_chunk = base64.b64decode(event["delta"])
await playback.enqueue(audio_chunk)
elif etype == "response.audio_transcript.delta":
# Useful for logging/observability even in a native pipeline.
log.debug("assistant_partial_transcript", text=event["delta"])
elif etype == "input_audio_buffer.speech_started":
# The user has started talking -- this is the barge-in signal.
await playback.cancel_current_response()
elif etype == "response.function_call_arguments.done":
await handle_function_call(ws, event, tool_registry)Notice response.audio_transcript.delta: even though the model reasons over audio natively, the API still emits a text transcript of its own output alongside the audio, which is a meaningful concession to the observability gap discussed in the previous lesson. It’s not a substitute for full cascaded-pipeline transparency (you still can’t inspect an intermediate ASR transcript of the user’s audio the way you can in a cascaded system), but it does give you a loggable record of what the assistant said, which matters for the same compliance and debugging reasons raised earlier in this module.
Function calling inside a live session
Tool use follows a familiar request-execute-respond shape, adapted to a persistent connection instead of a single request-response call. The model signals intent to call a registered function; your application executes the actual logic (hitting a database, calling an internal service); and the result is sent back into the same session so the model can continue speaking with that information incorporated.
async def handle_function_call(ws, event, tool_registry):
call_id = event["call_id"]
name = event["name"]
args = json.loads(event["arguments"])
fn = tool_registry[name]
result = await fn(**args) # e.g. actually query the orders database
await ws.send(json.dumps({
"type": "conversation.item.create",
"item": {
"type": "function_call_output",
"call_id": call_id,
"output": json.dumps(result),
},
}))
# Ask the model to continue generating a response now that it
# has the function's result available.
await ws.send(json.dumps({"type": "response.create"}))The response.create call at the end matters: adding the function output to the conversation doesn’t automatically trigger the model to speak again, it has to be explicitly told to continue. This is a small but easy-to-miss detail that trips up teams porting function-calling patterns from a text-based chat API to the realtime session shape for the first time, since the text API’s request-response cycle makes this step implicit in a way the persistent WebSocket connection does not.