Migrating from the OpenAI Realtime API to AssemblyAI

If you are running a voice agent on the OpenAI Realtime API and considering a move, here is the concrete result. One speech-to-speech session becomes three layers you can see: Universal-3.6 Pro Realtime turns the caller's audio into text and decides when they've finished, an LLM writes the reply (through LLM Gateway on your AssemblyAI key, or your existing OpenAI model), and the AssemblyAI Text-to-Speech API speaks it. The configuration you used to send in session.update splits across the three, your tools carry over almost unchanged, and barge-in becomes something you can make exact, down to the word the caller heard last.

This guide walks the migration in the order you will actually do it: what maps to what, the session you are porting from, the agent rebuilt layer by layer, what changes in your bill, and where the honest tradeoffs are. The code is Python on raw WebSockets, the same shape as a typical Realtime integration, so you can port it handler by handler.

What actually maps to what

The Realtime API runs listening, thinking and speaking inside one session, and your client talks to it through one event stream. On AssemblyAI each layer has its own interface: a streaming speech-to-text WebSocket, an OpenAI-compatible chat completions call, and a text-to-speech WebSocket. Almost every Realtime event has a direct counterpart. The ones that don't are the events that existed only because the model was speaking for you.

Two rows do most of the work in a migration. Turn detection moves into the speech-to-text model, so there are no VAD thresholds to port. And the conversation now lives in a messages list you own, which is what makes the truncation row precise.

Before: the session you are porting

A typical Realtime agent configures everything in one session.update. Here is a support agent with one tool, in the current GA session shape:

{
  "type": "session.update",
  "session": {
    "type": "realtime",
    "model": "gpt-realtime-2.1",
    "output_modalities": ["audio"],
    "instructions": "You are a support agent for an online retailer. Keep every reply to one or two short sentences. Lead with the answer.",
    "audio": {
      "input": {
        "format": { "type": "audio/pcm", "rate": 24000 },
        "turn_detection": { "type": "semantic_vad" }
      },
      "output": {
        "format": { "type": "audio/pcm", "rate": 24000 },
        "voice": "marin"
      }
    },
    "tools": [
      {
        "type": "function",
        "name": "lookup_order",
        "description": "Get status and ship date for an order. Use whenever the caller gives an order number. Returns status (string) and ships_on (ISO-8601 date).",
        "parameters": {
          "type": "object",
          "properties": {
            "order_number": { "type": "string", "description": "Digits only, e.g. 41820933." }
          },
          "required": ["order_number"]
        }
      }
    ],
    "tool_choice": "auto"
  }
}

Every field here has a home in the new stack. instructions becomes the system message. tools goes into the LLM request. audio.input becomes query parameters on the speech-to-text URL, and turn_detection disappears into the model. audio.output becomes query parameters on the text-to-speech URL. model becomes a model string on LLM Gateway, or your own OpenAI model if you keep it.

After: the same agent on AssemblyAI

Install three packages with pip install websockets openai sounddevice and set ASSEMBLYAI_API_KEY. That one key runs speech-to-text, LLM Gateway and text-to-speech. The agent is two files: tools.py and agent.py, which is shown below in five parts. Paste them together in order and it runs.

Configuration: where each session field goes

# agent.py: the Realtime agent rebuilt as speech-to-text -> LLM -> text-to-speech
import asyncio
import base64
import json
import os
import string
import threading
from urllib.parse import urlencode

import sounddevice as sd
import websockets
from openai import AsyncOpenAI

from tools import TOOLS, run_tool

API_KEY = os.environ["ASSEMBLYAI_API_KEY"]
AAI_HEADERS = {"Authorization": API_KEY}  # the raw key, no "Bearer" prefix

RATE = 24000                       # the 24 kHz PCM16 your Realtime client already handles
MIC_CHUNK = 1200                   # 50 ms of 24 kHz audio per frame
BYTES_PER_MS = RATE * 2 // 1000    # 16-bit mono

# Was session.instructions
INSTRUCTIONS = (
    "You are a support agent for an online retailer. Keep every reply to one "
    "or two short sentences. Lead with the answer. No markdown: everything "
    "you write is read aloud."
)
GREETING = "Hey, what can I do for you?"

# Was session.audio.input: format and turn detection
STT_URL = "wss://streaming.assemblyai.com/v3/ws?" + urlencode({
    "speech_model": "universal-3-6-pro",
    "sample_rate": RATE,           # pcm_s16le is the default encoding
    "agent_context": GREETING,     # what the agent says first
})

# Was session.audio.output: voice and format
TTS_URL = "wss://tts.us.assemblyai.com/v1/ws?" + urlencode({
    "voice_id": "49c443a4-8a1d-5f7e-b70b-98bdbd9823c3",  # Audrey
    "language": "english",
    "sample_rate": RATE,
    "encoding": "pcm_s16le",
})

# Was session.model. LLM Gateway speaks the OpenAI API, so the OpenAI SDK
# works as is. To keep calling OpenAI directly, set LLM_BASE_URL to
# https://api.openai.com/v1, LLM_API_KEY to your OpenAI key and LLM_MODEL.
llm = AsyncOpenAI(
    base_url=os.environ.get("LLM_BASE_URL", "https://llm-gateway.assemblyai.com/v1"),
    api_key=os.environ.get("LLM_API_KEY", API_KEY),
)
LLM_MODEL = os.environ.get("LLM_MODEL", "claude-haiku-4-5-20251001")
MAX_TOOL_ROUNDS = 3

Universal-3.6 Pro Realtime accepts any sample rate from 8000 to 96000 Hz, so the 24 kHz PCM16 your Realtime client already captures goes in unchanged. agent_context seeds the model with the agent's greeting, which helps it transcribe a short or spelled-out first reply. LLM Gateway is OpenAI-compatible, so the client is the OpenAI SDK you already use, with a different base_url. Pick any model from the available models list; the three environment variables at the end let you keep calling OpenAI directly instead.

Tools: carried over, one format change

# tools.py: your Realtime tools, carried over as they are
import json

# The same array you sent in session.tools, in the Realtime API's flat format.
TOOLS = [
    {
        "type": "function",
        "name": "lookup_order",
        "description": (
            "Get status and ship date for an order. Use whenever the caller gives "
            "an order number. Returns status (string) and ships_on (ISO-8601 date)."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "order_number": {
                    "type": "string",
                    "description": "Digits only, e.g. 41820933.",
                }
            },
            "required": ["order_number"],
        },
    }
]

# Stand-in for your order database.
ORDERS = {"41820933": {"status": "shipped", "ships_on": "2026-10-30"}}


async def lookup_order(order_number: str) -> dict:
    """Your existing tool. Unchanged by the migration."""
    row = ORDERS.get(order_number)
    if row is None:
        return {
            "error": (
                f"No order found for {order_number}. The customer lookup "
                "succeeded. Ask the caller to re-read just the order number."
            )
        }
    return row


async def run_tool(name: str, arguments: str) -> dict:
    # Arguments arrive as a JSON string, as they did in function_call items.
    args = json.loads(arguments or "{}")
    if name == "lookup_order":
        return await lookup_order(**args)
    return {"error": f"Unknown tool: {name}"}

Your tool implementations port unchanged, and the Realtime session.tools array can be pasted as it is. The one difference is the envelope: the Realtime API takes tools flat (type, name, description, parameters), while chat completions nests them under a function key. The chat_tool() helper in the next part does that conversion, so the array above stays the single source of truth. Arguments still arrive as a JSON string, as they did in Realtime function_call items.

Worth designing deliberately: the error text. The model reads it verbatim. “Lookup failed” makes the agent re-ask for everything. Naming the failing field, saying what did succeed, and telling the agent what to ask for next gets a clean recovery, which is why the stub returns a full sentence.

Listening: turn detection lives in the model

# agent.py (continued): listening
# "yes" and "no" are left out on purpose: they are real answers.
BACKCHANNELS = frozenset({
    "mhm", "mm", "mmhm", "uh", "huh", "um", "hmm", "ah", "oh",
    "yeah", "yep", "okay", "ok", "right", "gotcha",
})


def is_backchannel(text):
    words = text.lower().translate(str.maketrans("", "", string.punctuation)).split()
    return len(words) < 2 or all(w in BACKCHANNELS for w in words)


async def stream_microphone(stt):
    """Was input_audio_buffer.append: raw binary frames, no JSON, no base64."""
    loop = asyncio.get_running_loop()
    frames = asyncio.Queue()

    def on_audio(indata, frame_count, time_info, status):
        loop.call_soon_threadsafe(frames.put_nowait, bytes(indata))

    with sd.RawInputStream(samplerate=RATE, channels=1, dtype="int16",
                           blocksize=MIC_CHUNK, callback=on_audio):
        while True:
            await stt.send(await frames.get())


async def listen(stt, agent):
    """Turn detection now lives in the speech-to-text model."""
    async for raw in stt:
        msg = json.loads(raw)
        kind = msg.get("type")

        if kind == "Begin":
            print(f"[stt] session {msg['id']} on {msg['configuration']['model']}")
        elif kind == "SpeechStarted":
            pass  # was input_audio_buffer.speech_started; barge-in waits for words
        elif kind == "Turn":
            text = msg.get("transcript", "").strip()
            if not text:
                continue
            if agent.speaking:
                if is_backchannel(text):
                    continue  # "mhm" while the agent talks is not an interruption
                await agent.interrupt()
            if msg.get("end_of_turn"):
                # Was ...input_audio_transcription.completed plus the
                # response.create that server VAD sent for you
                print(f"[user] {text}")
                agent.turns.put_nowait(text)
        elif kind == "Error":
            raise RuntimeError(f"speech-to-text error: {msg}")
        elif kind == "Termination":
            return

Audio goes up as binary frames of about 50 ms, with no JSON and no base64. The server answers with Turn messages: partials while the caller talks, then one with end_of_turn: true when they've finished. That final turn replaces both the transcription-completed event and the automatic response.create that server VAD used to fire for you.

Barge-in waits for words. SpeechStarted arrives before the first Turn of each turn, but a caller saying “mhm” while the agent talks is listening, so the agent only stops when the transcript holds real words. The voice agent guide covers this backchannel filter in more depth. Leave the turn-silence parameters at their defaults: tune them only after you've measured a specific problem.

Thinking: the LLM call and tool round trips

# agent.py (continued): thinking
def chat_tool(tool):
    """Realtime tools are flat. Chat Completions nests them under "function"."""
    return {
        "type": "function",
        "function": {key: tool[key] for key in ("name", "description", "parameters")},
    }


CHAT_TOOLS = [chat_tool(tool) for tool in TOOLS]


async def think(messages):
    """One reply. Was response.create plus the function_call round trips."""
    for _ in range(MAX_TOOL_ROUNDS):
        completion = await llm.chat.completions.create(
            model=LLM_MODEL,
            messages=messages,
            tools=CHAT_TOOLS,
            max_tokens=300,
        )
        # Some providers return text and tool calls in separate choices
        text = " ".join(c.message.content for c in completion.choices if c.message.content).strip()
        calls = [call for c in completion.choices for call in (c.message.tool_calls or [])]
        if not calls:
            messages.append({"role": "assistant", "content": text})
            return text

        messages.append({
            "role": "assistant",
            "content": text or None,
            "tool_calls": [
                {"id": call.id, "type": "function",
                 "function": {"name": call.function.name, "arguments": call.function.arguments}}
                for call in calls
            ],
        })
        for call in calls:  # was response.function_call_arguments.done
            result = await run_tool(call.function.name, call.function.arguments)
            print(f"[tool] {call.function.name}{call.function.arguments} -> {result}")
            messages.append({  # was a function_call_output item
                "role": "tool",
                "tool_call_id": call.id,
                "content": json.dumps(result),
            })
    fallback = "Sorry, I lost track of that. Could you say it again?"
    messages.append({"role": "assistant", "content": fallback})
    return fallback

This is the part of a Realtime integration that used to be events and is now a function call. The model sees the history, either answers or asks for tools, and loops until it has an answer. Tool calls come back in message.tool_calls, results go back as role: "tool" messages with the matching tool_call_id, and the next request continues the conversation. No event ordering to track.

The code waits for the whole reply before speaking. To start speaking sooner, stream the completion and send each text delta to the text-to-speech socket with Generate as it arrives: the server starts speaking once it holds complete sentences, and Flush finishes the request. LLM Gateway streams responses from OpenAI models; see chat completions.

Speaking: barge-in and truncation

# agent.py (continued): speaking, barge-in and truncation
class Speaker:
    """Plays 16-bit PCM as it arrives. clear() drops what hasn't played yet."""

    def __init__(self):
        self._buf = bytearray()
        self._lock = threading.Lock()
        self._stream = sd.RawOutputStream(samplerate=RATE, channels=1,
                                          dtype="int16", callback=self._fill)
        self._stream.start()

    def _fill(self, outdata, frame_count, time_info, status):
        n = len(outdata)
        with self._lock:
            chunk = bytes(self._buf[:n])
            del self._buf[:n]
        outdata[:len(chunk)] = chunk
        outdata[len(chunk):] = b"\x00" * (n - len(chunk))

    def play(self, pcm):
        with self._lock:
            self._buf += pcm

    def clear(self):
        with self._lock:
            self._buf.clear()

    @property
    def buffered(self):
        return len(self._buf)

    def close(self):
        self._stream.stop()
        self._stream.close()


class Agent:
    def __init__(self, stt, tts):
        self.stt, self.tts = stt, tts
        self.speaker = Speaker()
        self.turns = asyncio.Queue()
        self.messages = [{"role": "system", "content": INSTRUCTIONS}]
        self.in_flight = False   # flushed, and its FlushDone hasn't arrived
        self.dropping = 0        # Cancels sent whose Cancelled hasn't arrived
        self.received = 0        # TTS audio bytes received this session
        self.flush_id = None     # the request now playing
        self.words = {}          # flush_id -> WordBoundaries words

    @property
    def speaking(self):
        return self.in_flight or self.speaker.buffered > 0

    async def say(self, text):
        """Was the audio half of response.create."""
        self.in_flight = True
        await self.tts.send(json.dumps({"type": "Generate", "text": text}))
        await self.tts.send(json.dumps({"type": "Flush"}))

    async def interrupt(self):
        """Was response.cancel plus conversation.item.truncate."""
        heard_ms = (self.received - self.speaker.buffered) / BYTES_PER_MS
        self.speaker.clear()
        self.in_flight = False
        self.dropping += 1
        await self.tts.send(json.dumps({"type": "Cancel"}))
        self.truncate(heard_ms)

    def truncate(self, heard_ms):
        """Keep only the words the caller actually heard in the history."""
        words = self.words.get(self.flush_id)
        last = self.messages[-1]
        if not words or last["role"] != "assistant":
            return  # timings not in yet: keep the full reply
        heard = [w["word"] for w in words if w["end"] <= heard_ms]
        last["content"] = " ".join(heard) + " [interrupted]"
        print(f"[agent] truncated to: {last['content']}")


async def speak(agent):
    """Handle every TTS frame after Begin."""
    async for raw in agent.tts:
        msg = json.loads(raw)
        kind = msg.get("type")

        if kind == "Audio":  # was response.output_audio.delta
            pcm = base64.b64decode(msg["audio"])
            agent.received += len(pcm)
            if agent.dropping:
                continue  # the tail of a cancelled reply
            agent.flush_id = msg["flush_id"]
            agent.speaker.play(pcm)
        elif kind == "FlushDone":  # was response.done
            agent.in_flight = False
        elif kind == "WordBoundaries":  # per-word timings on the session timeline
            agent.words[msg["flush_id"]] = msg["words"]
        elif kind == "Cancelled":
            agent.dropping -= 1
        elif kind == "Error":
            raise RuntimeError(f"text-to-speech {msg['error_code']}: {msg['error']}")
        elif kind == "Termination":
            return

The AssemblyAI Text-to-Speech API brings frontier-level quality and unlimited concurrency, and it is faster and cheaper than other production voice agent TTS models. One text-to-speech session lasts the whole call. Each reply is one request: Generate with the text, then Flush. Audio arrives as base64 Audio frames, and FlushDone marks the end of a request's audio, the way response.done did.

Barge-in is Cancel. It stops the synthesis in flight and discards text you sent but the caller hasn't heard; a short tail of audio already on the wire can still arrive before Cancelled, which is why the reader drops Audio frames while a cancel is outstanding. Clearing the local buffer stops playback immediately.

Truncation is where the cascade does more than the Realtime API. OpenAI's conversation.item.truncate cuts the model's audio at a timestamp, and OpenAI's docs note that it doesn't give you a truncated transcript. On the raw WebSocket, AssemblyAI text-to-speech sends a WordBoundaries frame for each request with per-word start and end times on the session's audio timeline. Compare those with how much audio has actually played, and you know the exact words the caller heard before cutting in. The agent rewrites its last message to just those words, so the LLM's next turn starts from what was really said. See word boundaries for the timing details.

The loop

# agent.py (continued): the conversation loop
async def converse(agent):
    """Each final turn becomes one LLM reply and one TTS request."""
    while True:
        text = await agent.turns.get()
        agent.messages.append({"role": "user", "content": text})
        answer = await think(agent.messages)
        print(f"[agent] {answer}")
        await agent.say(answer)
        # Tell speech-to-text what the agent just asked, for a better next turn
        await agent.stt.send(json.dumps({
            "type": "UpdateConfiguration",
            "agent_context": answer[-1750:],
        }))


async def keep_tts_alive(agent):
    """A TTS session closes after two idle minutes."""
    while True:
        await asyncio.sleep(60)
        if not agent.speaking:
            await agent.tts.send(json.dumps({"type": "KeepAlive"}))


async def wait_for_termination(ws):
    async for raw in ws:
        if json.loads(raw).get("type") == "Termination":
            return


async def main():
    async with websockets.connect(STT_URL, additional_headers=AAI_HEADERS) as stt, \
               websockets.connect(TTS_URL, additional_headers=AAI_HEADERS) as tts:
        begin = json.loads(await tts.recv())  # was session.created
        if begin["type"] == "Error":
            raise SystemExit(f"text-to-speech {begin['error_code']}: {begin['error']}")

        agent = Agent(stt, tts)
        agent.messages.append({"role": "assistant", "content": GREETING})
        await agent.say(GREETING)

        tasks = [asyncio.create_task(job) for job in (
            stream_microphone(stt), listen(stt, agent), speak(agent),
            converse(agent), keep_tts_alive(agent),
        )]
        print("Listening. Use headphones; press Ctrl+C to stop.")
        try:
            await asyncio.gather(*tasks)
        except asyncio.CancelledError:
            pass
        finally:
            for task in tasks:
                task.cancel()
            agent.speaker.close()
            for ws in (stt, tts):  # end both sessions cleanly
                try:
                    await ws.send(json.dumps({"type": "Terminate"}))
                    await asyncio.wait_for(wait_for_termination(ws), timeout=5)
                except (websockets.ConnectionClosed, asyncio.TimeoutError):
                    pass


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

After each reply the agent sends UpdateConfiguration with agent_context set to what it just said, so the next turn is transcribed in context. A short answer like “yes”, an email address or an order number is easier to transcribe when the model knows the question. In a speech-to-speech model this happened implicitly; here it's one message. On shutdown, Terminate ends both sessions cleanly.

Already using OpenAI's speech API in a pipeline?

Some Realtime migrations aren't from speech-to-speech at all: the agent already runs a cascade and calls OpenAI's speech endpoint for the voice. The AssemblyAI Text-to-Speech API answers the OpenAI speech API's request shape, POST /audio/speech, under https://tts.us.assemblyai.com/v1/openai. Point the OpenAI SDK at that base URL with your AssemblyAI key:

# speak.py: an existing OpenAI speech call, pointed at AssemblyAI
import os

from openai import OpenAI

client = OpenAI(
    base_url="https://tts.us.assemblyai.com/v1/openai",
    api_key=os.environ["ASSEMBLYAI_API_KEY"],
)

with client.audio.speech.with_streaming_response.create(
    model="assemblyai-tts-en",  # OpenAI names such as tts-1 are refused
    voice="alloy",              # an OpenAI voice name maps to an AssemblyAI voice
    input="Your order shipped this morning.",
    response_format="pcm",      # raw 16-bit mono PCM at 24 kHz
) as response:
    response.stream_to_file("reply.pcm")

Three details to get right. The model is an AssemblyAI name, such as assemblyai-tts-en or another language's code; OpenAI model names like tts-1 are refused with model_not_found. OpenAI voice names map to AssemblyAI voices (alloy speaks as aai-jane in English), or you can pass an AssemblyAI voice slug or ID. And ask for pcm: it plays from the first bytes. This route returns no word timings, and each reply is its own HTTP request, so for a new voice agent the WebSocket session above is the better fit; it gives you Cancel and WordBoundaries on one connection. The OpenAI compatibility page lists every field, including drop-in settings for LiveKit's and Pipecat's OpenAI TTS plugins.

What changes in your bill

The Realtime API bills tokens, and audio tokens are priced well above text: for gpt-realtime-2.1, OpenAI lists $32 per million audio input tokens and $64 per million audio output tokens, against $4 and $24 for text (OpenAI pricing). A conversation's cost depends on how long the caller talks, how much history you carry, how much the model says and how many tool round trips a turn takes.

The cascade bills each layer in its own unit. Streaming speech-to-text on Universal-3.6 Pro Realtime is $0.45/hr of audio. The LLM bills text tokens only, at the rate of the model you pick on LLM Gateway or your own provider, because audio never reaches it. Text-to-speech is $15 per 1M characters, billed on the text you send. You can see which layer drives cost and change one without touching the others. Current rates for all three are on the pricing page.

One habit to carry over: keep the system prompt and tool descriptions tight. In the cascade they are text tokens on every LLM call, the same as before, just without the audio multiplier.

What you gain on accuracy

In a speech-to-speech session, transcription is a side channel you can't tune. In the cascade it's the input to everything, and its inputs are yours to set. prompt takes up to 1,750 characters describing the conversation, keyterms_prompt takes up to 100 exact terms to boost, and agent_context carries the agent's last reply. All three can be updated mid-call with UpdateConfiguration, so you can narrow them as the conversation narrows: pair keyterms with a lookup tool so the transcript doesn't mangle a product name before the tool ever sees it. See prompting and keyterms.

On AssemblyAI's English voice-agent benchmark, Universal-3.6 Pro Realtime posts a mean word error rate of 5.19% and an entity error rate of 14.4%, and on the Pipecat open benchmark it has a 307 ms median time to final transcript. If your agent takes an order number, an address or a callback number, the entity figure is the one to read: aggregate word error rate tells you how the transcript reads, and entity error rate tells you whether the transaction completed. Methodology is on the benchmarks page.

What you give up, and when not to migrate

Being straight about this is more useful than a feature table.

• More connections to own. One session becomes two WebSockets and an HTTPS call, and the glue between them is your code: turn handling, barge-in, keep-alives. The program above is that glue, and frameworks like LiveKit and Pipecat take most of it off your hands, but it is still more surface than one socket.
• The LLM reads text. A speech-to-speech model hears tone and inflection directly. In the cascade the LLM sees the transcript. For most support, booking and intake agents the transcript is what matters; if your agent depends on reacting to how something was said, test that before you commit.
• Six voice languages. Universal-3.6 Pro Realtime transcribes 32 languages with native code-switching, but the Text-to-Speech API speaks six: English, Spanish, French, German, Italian and Portuguese. An agent that must answer in another language needs a different voice provider for that language.
• Streaming from the LLM depends on the model. LLM Gateway streams responses from OpenAI models. With other providers, the agent waits for the full reply before it starts speaking, as the code above does.

A migration plan that does not require a big-bang cutover

1. Map your session. Take your current session.update and assign every field a new home with the table above. Anything without a home is a decision to make now.
2. Port the tools first. Keep the Realtime array, convert it with chat_tool(), and run the LLM layer on its own against typed transcripts. Spend the time you save on tool descriptions and error text.
3. Pick the LLM. Start with the OpenAI model you have, or point the client at LLM Gateway and try others by changing one string.
4. Build the minimum loop. Final turn in, one LLM call, Generate and Flush out. Get one full turn working before adding barge-in, truncation and context updates.
5. Leave turn detection at its defaults. Don't port VAD thresholds. Change the turn-silence parameters only after you have measured a specific problem.
6. Re-tune the system prompt for voice. Prompts rarely transfer verbatim between models. Keep replies to a sentence or two and drop markdown entirely: the voice reads what you send.
7. Set prompt and keyterms before you benchmark. Comparing a tuned incumbent against an untuned replacement gives you the wrong answer.
8. Shadow a slice of traffic. Run both agents on the same calls, compare transcripts on the entities that matter to your business, and measure end-to-end turn latency the same way on both.
9. Cut over by segment. One workflow, one region or one hour of the day at a time.

The pages to keep open during the port: the speech-to-text message sequence, LLM Gateway tool calling, and the text-to-speech quickstart. If you are still deciding whether to leave the Realtime API at all, our roundup of OpenAI Realtime API alternatives compares the options.

Frequently asked questions

How hard is it to migrate from the OpenAI Realtime API to AssemblyAI?

Most of the work is moving configuration; the logic carries over. Each field of your session.update gets a new home: instructions become the system message, tools go into the LLM request, input audio settings become speech-to-text connection parameters, and the voice becomes a text-to-speech connection parameter. Tool implementations port unchanged, turn detection moves into Universal-3.6 Pro Realtime, and the event handlers become one speech-to-text reader, one LLM call and one text-to-speech reader.

Can I keep using my OpenAI model?

Yes. The LLM layer is an ordinary chat completions call, so you can keep calling OpenAI with your own key, or point the same OpenAI SDK at LLM Gateway at https://llm-gateway.assemblyai.com/v1 and use your AssemblyAI key. LLM Gateway lets you change models by changing one string.

What replaces response.cancel and conversation.item.truncate?

Cancel on the text-to-speech WebSocket stops the reply and discards text the caller hasn't heard. For truncation, the WordBoundaries frame gives per-word timings for each reply, so you can compare them with how much audio has played and rewrite the assistant message in your history to the exact words the caller heard.

Do I need to port my turn-detection settings?

No. Universal-3.6 Pro Realtime detects the end of each turn itself and sends a Turn message with end_of_turn set to true. Leave the turn-silence parameters at their defaults and only tune them after you have measured a specific problem.

Can I keep my existing OpenAI text-to-speech code?

If it calls OpenAI's speech API, yes: point the OpenAI SDK at https://tts.us.assemblyai.com/v1/openai with your AssemblyAI key and use an AssemblyAI model name such as assemblyai-tts-en. OpenAI voice names map to AssemblyAI voices. For a live voice agent, the WebSocket API is the better fit, because it adds barge-in and word timings on one session.

How is the cascaded stack billed compared to the Realtime API?

The Realtime API bills audio and text tokens. On AssemblyAI each layer bills on its own: streaming speech-to-text is $0.45/hr on Universal-3.6 Pro Realtime, the LLM bills text tokens at the rate of the model you choose, and text-to-speech is $15 per 1M characters. Current rates are on the pricing page.