Recipe: Realtime Voice Session

Goal

Hold a speech-to-speech conversation with a model: stream audio in, consume audio and transcripts as they are produced, and read the session's cost when it ends.

This is the session-oriented surface. For one-shot text-to-speech or transcription, use Multimedia instead — you do not need a live connection for those.

Prerequisites

• OPENAI_API_KEY for gpt-realtime, or GOOGLE_GEMINI_API_KEY for gemini-3.1-flash-live-preview. Keys are read from .env at the repo root.
• No extra dependency: both adapters go through their vendor SDK's own realtime client.

The examples below name the two long-standing rows. Both now carry a deprecation record — gpt-realtime retires 2027-01-20 in favour of gpt-realtime-2.1, and gemini-3.1-flash-live-preview is a legacy preview whose recommended replacement is gemini-3.8-live — and every example works unchanged on the newer rows, which is the point of a registry-driven adapter. Check the model table before pinning one.

One turn, start to finish

The smallest complete session — open, speak, listen, close:

import asyncio
from doc_processing_toolkit.realtime import open_realtime_session

async def main():
    async with await open_realtime_session(
        "gpt-realtime",
        voice="marin",
        instructions="You are terse. Answer in at most three words.",
        output_format="audio/pcm",
    ) as session:
        await session.send_text("Say the word hello, and nothing else.")

        audio = bytearray()
        transcript = []
        async for event in session.events():
            if event.type == "audio_delta":
                audio.extend(event.audio or b"")
            elif event.type == "transcript_delta" and event.role == "assistant":
                transcript.append(event.text or "")
            elif event.type == "error":
                raise RuntimeError(event.error)
            elif event.type == "response_complete":
                break

    print("".join(transcript))            # "Hello"
    print(len(audio), "bytes of PCM")
    print(f"${session.cost:.4f}")         # final after the context manager closes

asyncio.run(main())

open_realtime_session is a coroutine that returns an already-connected session, which is why the context manager reads async with await .... The adapter is chosen from the model's provider in the registry, so you never name one.

Always bound your event loop

session.events() runs until the session closes. Break on response_complete when you want one turn, and consider a maximum-event ceiling so a stalled stream fails loudly instead of hanging — the live tests use 400.

response_complete means "the model finished answering" on every model, including one that called a tool first. On gemini-3.8-live-extended-thinking the turn boundary arrives separately, as a turn_complete event, because that model keeps reasoning past it — see Realtime Voice.

Give the session tools

Pass tools= and the toolkit runs the model's tool calls for you — same loader, same MCP path, same hook events as an ordinary agent run — and answers them on the socket:

session = await open_realtime_session(
    "gpt-realtime-2.1-mini",
    tools=["tools/my_tools.py"],
    max_tool_rounds=8,          # per session, default 16
    tool_timeout_seconds=20,    # per call, default 30
)

You will see tool_call and tool_result events go by; you can ignore both if you only want the answer. Every ceiling, veto, timeout and tool crash is answered with an error result, because a realtime model left waiting on an unanswered tool slot simply goes quiet. See Tools in a realtime session for the per-provider mechanics.

Save the reply as a playable WAV

Both providers emit raw 24 kHz mono PCM16 — headerless bytes, not a playable file. Wrap them with the stdlib:

import wave
from doc_processing_toolkit.realtime import GOOGLE_LIVE_OUTPUT_SAMPLE_RATE

def write_wav(path: str, pcm: bytes, sample_rate: int = GOOGLE_LIVE_OUTPUT_SAMPLE_RATE) -> None:
    with wave.open(path, "wb") as fh:
        fh.setnchannels(1)
        fh.setsampwidth(2)              # 16-bit
        fh.setframerate(sample_rate)
        fh.writeframes(pcm)

write_wav("reply.wav", bytes(audio))

Streaming microphone audio in

send_audio takes little-endian 16-bit mono PCM. Feed it in chunks as they arrive from your capture device; with the provider's default server-side voice activity detection, the model decides when your turn ended and answers on its own.

from doc_processing_toolkit.realtime import (
    OPENAI_REALTIME_SAMPLE_RATE,       # 24000
    GOOGLE_LIVE_INPUT_SAMPLE_RATE,     # 16000
)

session = await open_realtime_session("gpt-realtime", voice="marin")
consumer = asyncio.create_task(drain_events(session))

for chunk in microphone_chunks():        # bytes, 20-100 ms each works well
    await session.send_audio(chunk)

The two providers want different input rates

OpenAI Realtime expects 24 kHz in; Gemini Live expects 16 kHz in. send_audio forwards your bytes unchanged and neither provider resamples for you, so audio captured at the wrong rate is transcribed as gibberish rather than rejected. Read the rate from the constants above, or from the registry's input_audio_sample_rate when you support both models.

Run the event consumer as a separate task, as above — sending and receiving are independent directions, and awaiting events() inline would block your capture loop.

Show the model a picture

send_image attaches a still to the conversation; send_video_frame offers a frame from a continuous capture and is rate-gated so a capture loop cannot multiply your bill.

await session.send_image(chart_png)                     # never throttled
await session.send_text("What is the trend in this chart?")

async for frame in screen_capture():                    # 30 fps in
    await session.send_video_frame(frame)               # ~1 fps on the wire
print(session.frames_sent, session.frames_dropped)

Pass encoded bytes, not raw pixels — the container is identified from the payload itself and an unidentifiable one is refused rather than guessed at. A frame does not ask the model to answer; it is context for whatever you say next. max_video_fps (default 1, 0 to disable the gate) is settable through open_realtime_session or configure. The full rules, including what happens to OpenAI's detail hint on Gemini, are in Showing the model what you are looking at.

Transcribe the caller's own audio

By default only the model's speech is transcribed. Ask the provider to transcribe your side too, and it arrives on the same event stream tagged role="user":

session.configure(input_transcription=True)

async for event in session.events():
    if event.type == "transcript_delta":
        who = "you" if event.role == "user" else "model"
        print(f"[{who}] {event.text}")

On OpenAI, input_transcription=True uses the row's default_input_transcription_model, which is gpt-transcribe on every conversational row. To steer that model, pass a dict of guidance fields. The dict is merged over the default, and language and languages are alternatives:

session.configure(
    input_transcription={"language": "de", "keywords": ["IBAN", "Rechnungsnummer"]},
    noise_reduction="far_field",      # laptop microphone; "near_field" for a headset, False for off
)

Manual turn control

Disable turn detection when your application decides where turns end — push-to-talk, for example. On OpenAI you then close each input turn yourself:

session = await open_realtime_session(
    "gpt-realtime",
    turn_detection={"disabled": True},
)

for chunk in push_to_talk_chunks():
    await session.send_audio(chunk)

await session.commit_audio()      # ends the turn and requests a response

commit_audio(create_response=False) closes the turn without asking for an answer yet. With server VAD left on, you never call this — the provider commits turns itself.

The same loop works on Gemini Live. The first send_audio() of a turn opens it with activity_start, and commit_audio() sends activity_end, which makes the model answer:

session = await open_realtime_session(
    "gemini-3.1-flash-live-preview",
    turn_detection={"disabled": True},
)

for chunk in push_to_talk_chunks_16k():
    await session.send_audio(chunk)   # activity_start goes out before the first chunk

await session.commit_audio()          # activity_end -> the model answers this turn

With Gemini's automatic detection left on, commit_audio() sends audio_stream_end instead. Call it when the microphone pauses so the server flushes what it has buffered.

Captions instead of a conversation

When you want the words and not a spoken reply, open a transcription session rather than muting the model. There is no model turn on that socket at all:

from doc_processing_toolkit.realtime import open_transcription_session

session = await open_transcription_session("gpt-transcribe")
try:
    for chunk in microphone_chunks():
        await session.send_audio(chunk)
    await session.commit_audio()
    async for event in session.events():
        if event.type == "transcript_complete":
            print(event.item_id, event.text)
finally:
    await session.close()       # drains what is still in flight first

Correlate on item_id, not on arrival order — items complete out of order. On gemini-3.5-transcribe-live the interim events are the additions to a cumulative snapshot and the settled caption only arrives once commit_audio() (or close()) has marked the end of the audio, so call it at each utterance boundary. See Live transcription.

Finer turn_detection settings are a neutral dict that is translated per provider, not forwarded: a key the target provider cannot express is dropped with a warning naming the nearest control it does have. The full mapping is in Realtime Voice.

Interpret a speaker into another language

A translation session is an interpreter, not a conversation: no voice to choose, no instructions, no tools, no turns. You stream speech in and get translated speech plus both transcripts back while the speaker is still talking.

from doc_processing_toolkit.realtime import open_translation_session

session = await open_translation_session(
    "gemini-3.5-live-translate-preview",     # 16 kHz in; OpenAI's row wants 24 kHz
    target_language="en",                    # required — there is no safe default
)
consumer = asyncio.create_task(drain_events(session))   # audio_delta -> speaker
try:
    for chunk in japanese_microphone():
        await session.send_audio(chunk)
finally:
    await session.close()                    # flushes, then drains the tail

print(session.source_transcript)             # what the speaker said
print(session.translated_transcript)         # the interpretation
print(session.detected_source_languages)     # ['ja'] — detected, never set

Streaming a file? Keep sending after the last sample

Both transcripts advance only while input audio keeps arriving. A loop that pushes a clip and stops sending loses the tail of both transcripts — and, on Gemini, the last usage report. Follow the clip with a second or two of paced silence, exactly as a microphone would, before you close:

silence = b"\x00\x00" * 16_000              # 1 s of 16 kHz mono PCM16
for chunk in clip_chunks():
    await session.send_audio(chunk)
    await asyncio.sleep(0.02)               # pace it; do not blast the socket
for _ in range(2):
    await session.send_audio(silence)
    await asyncio.sleep(1.0)

The target language is validated against the model's own published list before the socket opens, in the vendor's own spelling, so "PT-br" goes out as pt-BR and a wrong code fails immediately with a "did you mean" rather than being ignored. source_language= is accepted and ignored with a warning — neither provider has a field for it. Closing drains what the server is still flushing; on OpenAI that took 3.9 seconds in a live run, which is why that close has its own 10-second budget (provider_options={"close_drain_seconds": ...}). Full behaviour: Live translation.

GPT-Live, with a reasoning backend

gpt-live-1 is a different protocol on /v1/live/*, not another realtime model. Three habits from the sections above do not carry over: there is no audio commit, no response_complete to break on, and no item_id on a transcript delta.

from doc_processing_toolkit.realtime_gpt_live import open_live_session

session = await open_live_session(
    "gpt-live-1",
    voice="marin",
    instructions="Be concise. Delegate order questions to the backend.",
    tools=["tools/order_tools.py"],                          # run locally, for the backend
    provider_options={"backend_model": "gpt-5.6-luna"},      # without this, tools= is refused
)
try:
    async for event in session.events():
        if event.type == "audio_delta":
            speaker.write(event.audio)
        elif event.type == "transcript_delta":
            print(event.role, event.text)                    # event.item_id is always None
        elif event.type == "session_closed":
            break                                            # the only terminal event
finally:
    await session.close()

print(session.billed_audio_seconds, session.cost)
print(session.cost_excludes)        # why that number is not the whole bill
print(session.delegated_usage)      # backend responses, for you to price

Drive the loop from your playback queue or the transcript and end the session when your application decides — nothing on this protocol marks the end of a spoken answer. The bill is wall clock at $0.05/minute, silences and backend thinking included, so bound the session yourself. Steer a running conversation with session.append_instructions(...) (there is no cancel — the model is full duplex and interrupts itself), and note that session.cost covers the voice session only: a delegated backend is billed separately on its own rates. Full behaviour: GPT-Live.

Interrupt, and tell the model what was heard

await session.cancel_response()                    # stop mid-answer
await session.truncate_response(played_ms=1_800)   # what the listener actually heard

Without the truncation the provider still believes it said the whole answer, and the next turn is grounded in words nobody heard. session.emitted_audio_ms is the toolkit's own count of the audio it handed you for the current assistant item and is used when you pass no played_ms — pass your own if you buffer before playback. OpenAI has both controls; Gemini Live has no truncation message, and its cancel exists only on the 3.8 rows.

Gemini reconnects are already handled

Gemini Live caps session duration server-side (15 minutes, 2 with video). The adapter hides that: it enables session resumption, records each handle the server issues, reacts to the server's goAway warning, and reconnects with the stored handle. Your events() iterator does not stop — you just see one extra event at the seam:

async for event in session.events():
    if event.type == "session_resumed":
        logger.info("Reconnected transparently; conversation continues.")
        continue

You only need to handle it if you want to observe the seam. Bound the retries with provider_options={"max_reconnects": 3} if the default of 10 is more patience than your application wants; reconnects back off exponentially, capped at 8 seconds.

Export the session and seed the next one

import json

record = session.export()                      # JSON-clean; final after close()
json.dump(record, open("call-1.json", "w"))
print(record["transcript"])                    # [{"role": "user", "text": "...", ...}, {"role": "assistant", ...}]

seed = [{"role": e["role"], "text": e["text"]} for e in record["transcript"]]
async with await open_realtime_session(
    "gpt-realtime",
    voice="marin",
    output_format="audio/pcm",
    provider_options={"initial_items": seed},  # the model starts knowing the earlier turns
) as resumed:
    await resumed.send_text("What was my name again?")
    async for event in resumed.events():
        if event.type == "response_complete":
            break

The same option works on a Gemini Live row; GPT-Live takes provider_options={"initial_input": [{"role": ..., "content": ...}]} for the same purpose. See Exporting a session and seeding the next one.

Read the cost

Usage accumulates per modality in both directions and is priced under the registry's audio_realtime schema. Both are live during the session and final after close():

usage = session.usage
print(usage.input_audio_tokens, usage.output_audio_tokens)
print(usage.input_image_tokens)      # stills and frames, billed at the row's input_image rate
print(usage.cached_input_tokens)     # a SUBSET of total_input_tokens, never an addition
print(session.usage.as_dict())       # every counter plus derived totals
print(session.cost, session.cost_estimated)

response_complete events also carry a cumulative usage dict and cost float, which is the cheapest way to drive a running cost display without touching the session object.

For reference, the live smoke tests bill roughly $0.0045 for a one-word gpt-realtime turn (158,400 bytes of PCM) and $0.0010 for the same turn on gemini-3.1-flash-live-preview (43,710 bytes). Audio tokens dominate — text is a rounding error at these sizes.

Not every session is token-priced. gpt-live-1 and gpt-realtime-translate are billed by the clock, so session.usage stays empty and the number to watch is session.billed_audio_seconds; read session.cost_excludes before treating session.cost as the whole charge. See Sessions billed by the second.

Validate before you connect

Options are checked against the model's registry row before any network activity, so a bad voice fails immediately and names what the model does support:

from doc_processing_toolkit.realtime import RealtimeValidationError

try:
    await open_realtime_session("gpt-realtime", voice="Kore")   # a Gemini voice
except RealtimeValidationError as exc:
    print(exc)      # names the supported OpenAI voices

Voice names are opaque provider strings — gpt-realtime has 10, gemini-3.1-flash-live-preview has 30, and there is no cross-provider enum. Errors the provider reports mid-session arrive as error events instead, so your consumer loop decides whether to continue or break.

Testing without an API key

The unit suite drives both adapters over mock transports (tests/unit/test_realtime.py), covering lifecycle, event mapping, usage accumulation, cost, resumption handles, and reconnects. The live sessions live in tests/integration/test_realtime_live.py and skip without keys. See Testing.

Related

• Realtime Voice — the full event vocabulary and provider behavior
• Multimedia — one-shot TTS, STT, and image generation
• Cost Tracking — pricing schemas and the cached-tokens convention