Real-Time TTS with WebSockets

Why Use WebSockets for TTS?

WebSockets provide a continuous audio stream flowing directly to the playback device without saving files to disk. This approach is essential for voice agents and conversational AI that require minimal latency and natural-sounding speech.

Key benefits include low latency, which allows audio playback to begin as soon as the first data chunk arrives, continuous streaming that maintains a persistent connection for rapid audio delivery, and efficient processing by streaming audio directly to playback devices.

WebSocket Implementation Examples

The following examples demonstrate how to implement real-time TTS using Deepgram’s WebSocket API:

# For more Python SDK migration guides, visit:
# https://github.com/deepgram/deepgram-python-sdk/tree/main/docs
import sounddevice as sd
import numpy as np
import time
from deepgram import DeepgramClient
from deepgram.core.events import EventType
from deepgram.speak.v1.types import SpeakV1Text
TTS_TEXT = "Hello, this is a text to speech example using Deepgram."
def main():
    try:
        # Create a Deepgram client using the API key from environment variables
        deepgram = DeepgramClient()
        # Create a websocket connection to Deepgram
        with deepgram.speak.v1.connect(
            model="aura-2-thalia-en",
            encoding="linear16",
            sample_rate=48000
        ) as dg_connection:
            def on_message(message) -> None:
                if isinstance(message, bytes):
                    print("Received audio chunk")
                    # Convert binary data to audio format playback devices understand
                    array = np.frombuffer(message, dtype=np.int16)
                    # Play the audio immediately upon receiving each chunk
                    sd.play(array, 48000)
                    sd.wait()
                else:
                    msg_type = getattr(message, "type", "Unknown")
                    print(f"Received {msg_type} event")
            dg_connection.on(EventType.OPEN, lambda _: print("Connection opened"))
            dg_connection.on(EventType.MESSAGE, on_message)
            dg_connection.on(EventType.CLOSE, lambda _: print("Connection closed"))
            dg_connection.on(EventType.ERROR, lambda error: print(f"Error: {error}"))
            dg_connection.start_listening()
            # Send text to be converted to speech
            dg_connection.send_text(SpeakV1Text(text=TTS_TEXT))
            # Send control messages
            dg_connection.send_flush()
            # Allow time for playback
            time.sleep(5)
            dg_connection.send_close()
            print("TTS stream completed")
    except Exception as e:
        print(f"An error occurred: {e}")
if __name__ == "__main__":
    main()

For optimal text handling, see our guide on Text Chunking for TTS.