How to stream text to speech in Python over a WebSocket

Search for "text to speech python" and you’ll find a lot of ways to turn a sentence into an MP3. That’s fine for a script that reads a notification aloud. It falls apart the moment you build a voice agent, because a voice agent has three problems a saved file doesn’t solve: the reply is still being written by an LLM when the caller should already be hearing it, the caller interrupts, and the conversation runs for minutes on a connection that has to stay healthy the whole time.

This tutorial covers all three in Python. You’ll start with a one-shot WAV file, then move to a single WebSocket session that speaks a whole conversation, streams LLM tokens into speech as they arrive, and stops the instant the caller barges in. Every sample uses the AssemblyAI Python SDK and is based on the official docs.

What you’ll build

• A WAV file from one call to TextToSpeech.synthesize.
• A multi-sentence reply spoken over one TTSSession, saved with a proper WAV header.
• Live playback of raw PCM as it streams in.
• An LLM-to-speech loop that flushes at each sentence and cancels on barge-in.
• A long-lived session that survives quiet stretches with keep_alive().
• The same thing with no SDK, over the raw WebSocket.

What is the AssemblyAI Text-to-Speech API?

The AssemblyAI Text-to-Speech API turns text into speech with the AssemblyAI-TTS-1 model (assemblyai-tts-1), which was trained on millions of hours of audio, including real conversational data. It’s built for conversational text to speech: short turns, names and numbers read the way they’re written, and audio that starts while the reply is still being generated. It delivers frontier-level quality, runs faster and costs less than other production voice agent TTS models, and has unlimited concurrency.

You can reach it three ways:

• HTTP (POST https://tts.us.assemblyai.com/v1/speech): one whole text in, audio back. Good for prompts, previews, and anything you can cache.
• WebSocket (wss://tts.us.assemblyai.com/v1/ws): one session per conversation, one request per turn. This is the interface for a voice agent.
• OpenAI-compatible speech endpoint, if you already have code written against that shape. See OpenAI compatibility.

It speaks six languages (English, Spanish, French, German, Italian and Portuguese) with thousands of voices to choose from, and every voice can speak all six. The choose an interface guide walks through which route fits how your text arrives.

Install the SDK and set your API key

Text to speech is in the Python SDK from release 1.8.2. Install or upgrade it, and set your API key as an environment variable:

pip install -U assemblyai
export ASSEMBLYAI_API_KEY=<your-key>

Every sample below reads the key from ASSEMBLYAI_API_KEY, so nothing secret ends up in your source.

Generate your first audio file in Python

Start with the simplest call there is: one text in, one WAV file out. TextToSpeech.synthesize calls the HTTP API and returns the whole file. Save this as hello.py:

import os

import assemblyai as aai

tts = aai.TextToSpeech(api_key=os.environ["ASSEMBLYAI_API_KEY"])

response = tts.synthesize(
    "Hi, thanks for calling. How can I help you today?",
    aai.TTSConfig(
        voice_id="49c443a4-8a1d-5f7e-b70b-98bdbd9823c3",  # Audrey
        container=aai.TTSContainer.wav,
    ),
)

with open("hello.wav", "wb") as f:
    f.write(response.audio)

print(f"voice {response.voice_id}, model {response.model}, "
      f"{response.sample_rate} Hz {response.encoding}")
print(f"wrote hello.wav: {response.audio_duration_ms} ms of audio, "
      f"{response.billed_characters} characters billed")

Run python hello.py and hello.wav plays in any audio player. A few things worth knowing about that call:

• Only the text is required. The code sets voice_id explicitly to Audrey, by her ID. Set it in your own code too, so a change to the default voice never changes what your callers hear. The voices page explains how to find a voice ID and set the language.
• The response tells you what it ran with. voice_id, model, sample_rate and encoding come back on the result. Read the sample rate from there rather than assuming it when you set up a player.
• Billing counts characters. billed_characters is what you pay for, at $15 per 1M characters, and audio_duration_ms varies a little from run to run because synthesis isn’t deterministic.

To speak another language, add language="spanish" (or "french", "german", "italian", "portuguese") to the TTSConfig. The API doesn’t detect language from your text, so set it to match.

Speak a whole conversation over one WebSocket session

A voice agent says many short things, one per turn. Opening a fresh HTTP request or WebSocket per sentence pays the connection setup every time, and it burns through your limit on new sessions per minute. So the pattern is one TTSSession per conversation, with each sentence as a request on it.

This program opens a session, speaks a four-sentence reply, and writes it to reply.wav. Save it as reply.py:

import os
import wave

import assemblyai as aai

config = aai.TTSConfig(
    voice_id="49c443a4-8a1d-5f7e-b70b-98bdbd9823c3",  # Audrey
    language="english",
)
reply = [
    "Hi, thanks for calling.",
    "You have reached the accounts team.",
    "There is one open invoice from last month.",
    "Would you like to pay it today?",
]

with aai.TTSSession(config, api_key=os.environ["ASSEMBLYAI_API_KEY"]) as session:
    print(f"session {session.begin.id}")
    rate = session.begin.configuration["sample_rate"]  # read it back, don't assume

    with wave.open("reply.wav", "wb") as out:
        out.setnchannels(1)  # mono
        out.setsampwidth(2)  # 16-bit PCM
        out.setframerate(rate)
        for sentence in reply:
            n = 0
            for chunk in session.speak(sentence):  # raw 16-bit mono PCM, no header
                out.writeframes(chunk)
                n += len(chunk)
            print(f"{sentence!r}: {n} bytes")

summary = session.termination  # None if the summary did not arrive in time
if summary:
    print(f"{summary.audio_duration_seconds} s of audio "
          f"for {summary.total_input_char_length} characters")

Here’s what the session does for you:

• The server speaks first. Entering the with block waits for the Begin frame, which carries the session id and the configuration you actually got. A bad key or an unknown voice raises right there.
• Each sentence is one request. speak() sends the text, flushes it, and yields that request’s audio until it’s done. That’s why the loop can count bytes per sentence.
• Leaving the block ends the session cleanly. It sends Terminate, speaks anything not yet flushed, and stores the summary on session.termination.

Play raw PCM and write a WAV header

The session streams raw audio: 16-bit little-endian PCM, mono, at the sample rate in Begin (24 kHz unless you ask for another). There’s no header, which is what makes it cheap to stream, and also why a file of those bytes won’t open in a normal player.

You have three options.

Write a WAV header. That’s what reply.py does with Python’s built-in wave module: set one channel, a sample width of 2 bytes, and the frame rate from Begin, then write the chunks as frames. wave fills in the RIFF header and the data length when the file closes. The TTS quickstart shows the same header built by hand with struct, if you’d rather see the bytes.

Tell the player the format. If you dump the chunks straight to a .pcm file, SoX or ffmpeg can play it as long as you pass the format:

play -t raw -r 24000 -e signed -b 16 -c 1 reply.pcm           # SoX
ffplay -f s16le -ar 24000 -ac 1 reply.pcm                      # ffmpeg

Play it as it arrives. For an agent, you don’t want a file at all. sounddevice (pip install sounddevice) opens a raw output stream that takes the bytes as they are:

import os

import assemblyai as aai
import sounddevice as sd

config = aai.TTSConfig(
    voice_id="49c443a4-8a1d-5f7e-b70b-98bdbd9823c3",  # Audrey
    language="english",
)

with aai.TTSSession(config, api_key=os.environ["ASSEMBLYAI_API_KEY"]) as session:
    rate = session.begin.configuration["sample_rate"]
    with sd.RawOutputStream(samplerate=rate, channels=1, dtype="int16") as speaker:
        for chunk in session.speak("Sure. Your order shipped this morning."):
            speaker.write(chunk)  # plays while the rest is still arriving

speaker.write() blocks until there’s room in the device buffer, so playback paces itself. Audio arrives faster than it plays, which is what you want: the buffer stays full and the caller hears no gaps. What the caller does hear first is set by time to first audio, which is worth measuring on your own side of the connection.

Phone calls are the exception to 24 kHz. Twilio and most carriers want 8 kHz mu-law, so ask for it in the config with encoding=aai.TTSEncoding.pcm_mulaw and sample_rate=8000, and pass the bytes straight to the media stream.

Stream LLM tokens into speech with generate, flush and audio

So far every text was complete before you sent it. An LLM doesn’t work that way: it writes a token at a time, and waiting for the whole reply before speaking adds the LLM’s entire generation time to every turn. In a chained STT-LLM-TTS pipeline, that wait is the easiest one to remove.

TTSSession splits speak() into its three parts so you can feed it as text arrives:

• generate(text) appends text to the current request. Pieces are joined exactly as sent, so keep the spaces.
• flush() concludes the request and returns its flush_id.
• audio(flush_id) yields that request’s audio until it’s done.

The server starts speaking once it holds enough complete sentences, even before you flush. Flushing at the end of each sentence starts that sentence’s audio right away, and it keeps requests short enough to cancel cleanly. Don’t flush per token, though: each fragment would be synthesized on its own, which sounds choppy, and Flush counts against a control-frame rate limit.

The program below runs the writer in a thread, so text goes out while audio comes back. The llm_tokens() generator stands in for your LLM; swap in your client’s streaming iterator. If you route your model through LLM Gateway, that’s the token stream you’d put here. It’s the same idea as streaming LLM responses sentence by sentence, with the TTS side filled in.

import os
import queue
import threading
import time

import assemblyai as aai
import sounddevice as sd

config = aai.TTSConfig(
    voice_id="49c443a4-8a1d-5f7e-b70b-98bdbd9823c3",  # Audrey
    language="english",
)


def llm_tokens():
    """Stand-in for a streaming LLM. Swap in your client's token stream."""
    reply = ("Sure, I can help with that. Your last order shipped on Tuesday. "
             "It should arrive by Friday. Is there anything else I can check?")
    for word in reply.split(" "):
        time.sleep(0.05)  # pretend tokens trickle in
        yield word + " "  # tokens are joined exactly as sent: keep the spaces


# Stand-in for barge-in. In a real agent your speech-to-text sets this
# the moment the caller starts talking. Here it fires after 3 seconds.
barge_in = threading.Event()
threading.Timer(3.0, barge_in.set).start()


def write_tokens(session, flush_ids, stop):
    """Send tokens as they arrive; flush at the end of every sentence."""
    pending = False
    for token in llm_tokens():
        if stop.is_set():
            break
        session.generate(token)
        pending = True
        if token.rstrip().endswith((".", "!", "?")):
            flush_ids.put(session.flush())  # this sentence starts speaking now
            pending = False
    if pending and not stop.is_set():
        flush_ids.put(session.flush())  # whatever is left after the last period
    flush_ids.put(None)  # no more requests


with aai.TTSSession(config, api_key=os.environ["ASSEMBLYAI_API_KEY"]) as session:
    rate = session.begin.configuration["sample_rate"]
    flush_ids = queue.Queue()
    stop = threading.Event()
    writer = threading.Thread(target=write_tokens, args=(session, flush_ids, stop))
    writer.start()

    with sd.RawOutputStream(samplerate=rate, channels=1, dtype="int16") as speaker:
        interrupted = False
        while not interrupted:
            flush_id = flush_ids.get()
            if flush_id is None:
                break
            for chunk in session.audio(flush_id):  # requests come back in order
                if barge_in.is_set():
                    interrupted = True
                    break
                speaker.write(chunk)

        if interrupted:
            stop.set()      # stop sending LLM text
            writer.join()
            session.cancel()  # drop the request in flight and anything queued
            speaker.abort()   # and drop the audio already buffered locally
            print("caller barged in, reply cancelled")
        else:
            writer.join()

Handle barge-in with cancel

The last block of that program is the barge-in. When the caller starts talking over the agent, three things have to stop, in this order:

1. The LLM text. stop.set() tells the writer thread to quit, and join() waits for it, so no new text slips in after the cancel.
2. The server. session.cancel() stops the request being spoken and drops every request queued behind it, along with any text sent but not flushed. The audio() iteration of a cancelled request ends at once, and the SDK discards any of its audio still in transit.
3. Your speaker. Audio you already handed to the device keeps playing unless you drop it. speaker.abort() does that for sounddevice.

The session stays open after a cancel. The next reply is just the next generate() and flush(), and it gets a new flush_id (ids skip one after a cancel, so read them from flush() rather than counting).

In the sample, a timer fakes the interruption. In a real agent, the signal comes from your speech-to-text: the moment it reports the caller has started a turn, set the event. Turn detection decides how quickly and how reliably that happens, so it’s worth getting right before you tune anything on the TTS side. If you need to know exactly which words the caller heard before cutting in, open the session with word_boundaries=True and read the per-word timings described in word boundaries.

Keep the session alive: keep_alive and session limits

A conversation has quiet stretches. The caller is thinking, or talking, and the agent has nothing to say. The session has two clocks you need to respect:

• Idle timeout. The server closes a session after two minutes with no frame in either direction. Send keep_alive() at least that often while you wait. It restarts the idle timer and does nothing else, so sending it more often doesn’t make anything faster, and a tight loop of them trips the control-frame limit.
• Session lifetime. A session lasts at most one hour. Begin carries the exact deadline in expires_at. For a longer call, open the replacement in a pause before then, move new turns to it, and terminate the old one.

Two more limits matter in practice. A single generate() takes at most 2,000 characters, and speak() splits longer text at fixed character positions, which can land mid-sentence, so split long text at sentence ends yourself. And rate limits count new sessions per minute; sessions held open aren’t counted, which is one more reason to keep one session per conversation. The full list is in limits.

This program opens the session up front, the way an agent should when a call starts, and speaks whatever you type. After a minute of silence it sends a keep-alive instead:

import os
import queue
import threading

import assemblyai as aai
import sounddevice as sd

config = aai.TTSConfig(
    voice_id="49c443a4-8a1d-5f7e-b70b-98bdbd9823c3",  # Audrey
    language="english",
)

# Stand-in for your agent: every line you type becomes a reply.
replies = queue.Queue()


def read_stdin():
    while True:
        replies.put(input("say> "))


threading.Thread(target=read_stdin, daemon=True).start()

# Open the session when the call starts, not when the first reply is ready.
session = aai.TTSSession(config, api_key=os.environ["ASSEMBLYAI_API_KEY"])
begin = session.connect()  # waits for Begin
print(f"session {begin.id} expires at {begin.expires_at}")
rate = begin.configuration["sample_rate"]

try:
    with sd.RawOutputStream(samplerate=rate, channels=1, dtype="int16") as speaker:
        while True:
            try:
                reply = replies.get(timeout=60)  # your agent; empty after 60 s of silence
            except queue.Empty:
                session.keep_alive()  # restarts the idle timer, nothing else
                continue
            for chunk in session.speak(reply):
                speaker.write(chunk)
except KeyboardInterrupt:
    pass
finally:
    session.terminate()

Without the with block, you call connect() and terminate() yourself. connect() returns the Begin event, and opening early also gives the server time to load the voice before the first reply. If your agent is built on asyncio, aai.AsyncTTSSession has the same methods as awaitables.

Using the raw WebSocket in Python

Not using the SDK? The protocol is small enough to drive by hand with the websockets package (version 14 or newer). Put your API key in the Authorization header and the voice and audio options in the query string, then:

1. Read Begin (or Error) first. The server always speaks first.
2. Send {"type": "Generate", "text": ...}, then {"type": "Flush"}.
3. Collect Audio frames, whose audio field is base64, until FlushDone.
4. Send {"type": "Terminate"} and wait for Termination.

import asyncio
import base64
import json
import os
import wave

import websockets

URL = (
    "wss://tts.us.assemblyai.com/v1/ws"
    "?voice_id=49c443a4-8a1d-5f7e-b70b-98bdbd9823c3&language=english"  # Audrey
    "&sample_rate=24000&encoding=pcm_s16le"
)


async def main():
    headers = {"Authorization": os.environ["ASSEMBLYAI_API_KEY"]}

    async with websockets.connect(URL, additional_headers=headers) as ws:
        # The server speaks first: Begin, or Error for a bad key or parameter.
        first = json.loads(await ws.recv())
        if first["type"] == "Error":
            raise SystemExit(f"{first['error_code']}: {first['error']}")
        config = first["configuration"]
        print(f"voice {config['voice_id']}, model {config['model']}, "
              f"{config['sample_rate']} Hz {config['encoding']}")

        await ws.send(json.dumps({
            "type": "Generate",
            "text": "Hi, thanks for calling. How can I help you today?",
        }))
        await ws.send(json.dumps({"type": "Flush"}))

        pcm = bytearray()
        async for raw in ws:
            message = json.loads(raw)
            kind = message.get("type")

            if kind == "Audio":
                pcm += base64.b64decode(message["audio"])
            elif kind == "FlushDone":
                print(f"flush {message['flush_id']}: "
                      f"{message['audio_duration_ms']} ms of audio")
                await ws.send(json.dumps({"type": "Terminate"}))
            elif kind == "Termination":
                break
            elif kind == "Error":
                raise SystemExit(f"{message['error_code']}: {message['error']}")
            # Ignore other frames, such as WordBoundaries.

    with wave.open("out.wav", "wb") as f:
        f.setnchannels(1)
        f.setsampwidth(2)
        f.setframerate(config["sample_rate"])
        f.writeframes(bytes(pcm))

    print(f"wrote out.wav ({len(pcm)} bytes of PCM)")


asyncio.run(main())

Two things the SDK hides that this shows. Word boundaries are on by default on the wire, so you’ll see WordBoundaries frames after each FlushDone; the loop ignores them, and you should ignore any frame type you don’t handle. And barge-in on the wire is {"type": "Cancel"}, answered by a Cancelled frame, with any Audio frames that arrive in between belonging to the cancelled request.

Where TTS fits in a Python voice agent

Text to speech is the last leg of the loop. The full stack is streaming speech-to-text, an LLM, and TTS. On AssemblyAI that’s Universal-3.6 Pro Realtime (universal-3-6-pro) to hear the caller and decide when they’ve finished, LLM Gateway or your own model to decide what to say, and the Text-to-Speech API to say it, all on one API key.

If you’d rather not run the orchestration yourself, both major open-source frameworks have it built in. The LiveKit plugin and the Pipecat service reuse one session across replies and cancel in-flight audio on barge-in without reconnecting. When you’re comparing voices for production, how to evaluate TTS for voice agents covers the tests that matter: names, numbers and interruptions, on phone-quality audio. Text-to-Speech costs $15 per 1M characters; see the pricing page.

The part most Python TTS tutorials skip

Getting audio out of a TTS API in Python takes four lines. What decides whether a voice agent sounds natural is everything around those lines: when you flush, how fast you can stop, and whether the connection is already open when the caller finishes talking. Those are session-design decisions, and they’re the same ones whichever framework you end up in.

So build the loop once by hand, the way this tutorial does. Watch where the flushes land, interrupt it mid-sentence, leave it idle for a few minutes. Once you’ve seen those behaviors in a hundred lines of Python, the settings in LiveKit or Pipecat stop being magic, and you’ll know which one to reach for when a call sounds wrong.

Frequently asked questions

What is the best way to do text to speech in Python?

For a voice agent or any live conversation, the best way to do text to speech in Python is a streaming TTS API over a WebSocket, with one session held open for the whole conversation. That lets audio start while an LLM is still writing the reply and lets you cancel playback when the user interrupts. For a one-off file, a single HTTP call such as the AssemblyAI SDK’s TextToSpeech.synthesize is enough.

How do I convert text to speech in Python with an API?

Install the AssemblyAI SDK with pip install -U assemblyai, create aai.TextToSpeech with your API key, and call synthesize() with your text and a TTSConfig that sets voice_id and container=aai.TTSContainer.wav. The result’s audio attribute holds the WAV bytes, ready to write to a file. The result also reports the sample rate, encoding, duration and billed characters.

How do I stream text to speech in Python while an LLM is still generating?

Open an aai.TTSSession, call generate() with each token as the LLM produces it, and call flush() at the end of each sentence to start that sentence’s audio. Read the audio with audio(flush_id), ideally from a separate thread so sending and playback overlap. Avoid flushing per token, which makes speech choppy and counts against the control-frame rate limit.

How do I stop text-to-speech audio when the user interrupts?

Call cancel() on the session: it stops the request being spoken, drops queued requests and unflushed text, and ends the current audio() iteration. Also stop sending new LLM text first, and clear any audio already buffered in your local player, or it keeps playing. The session stays open for the next reply.

Should I use an offline Python TTS library or a text-to-speech API?

An offline library is a reasonable choice for local scripts, accessibility tools and prototypes that don’t need natural conversational speech. A text-to-speech API fits production voice agents, where you need streaming input from an LLM, low time to first audio, barge-in handling and voices that read names and numbers correctly. Many teams prototype offline and move to an API once real users are on the line.

What audio format does the AssemblyAI TTS API return, and which languages does it support?

WebSocket sessions return raw 16-bit PCM at 24 kHz by default with no header, and you can request other sample rates or 8 kHz mu-law and A-law for telephony. The HTTP API can also return a ready-to-play WAV file. The model speaks English, Spanish, French, German, Italian and Portuguese, set with the language parameter, and billing is based on input characters, at $15 per 1M characters.