Quickstart

pip install assemblyai sounddevice

import os

import sounddevice as sd
from assemblyai.streaming.v3 import (
    BeginEvent,
    StreamingClient,
    StreamingClientOptions,
    StreamingError,
    StreamingEvents,
    StreamingParameters,
    TerminationEvent,
    TurnEvent,
)

SAMPLE_RATE = 16000


def on_begin(client: StreamingClient, event: BeginEvent):
    print(f"Session started: {event}")
    print("Connected. Speak into your microphone; press Ctrl+C to stop.")


def on_turn(client: StreamingClient, event: TurnEvent):
    print(event.transcript)


def on_terminated(client: StreamingClient, event: TerminationEvent):
    print(f"Session terminated: {event.audio_duration_seconds}s of audio processed")


def on_error(client: StreamingClient, error: StreamingError):
    print(f"Error: {error}")


def mic_stream():
    # sounddevice bundles PortAudio in its wheel — no system install needed.
    with sd.RawInputStream(
        samplerate=SAMPLE_RATE, channels=1, dtype="int16", blocksize=800
    ) as mic:
        while True:
            frames, _ = mic.read(800)  # ~50 ms of audio
            yield bytes(frames)


def main():
    client = StreamingClient(
        StreamingClientOptions(api_key=os.environ["ASSEMBLYAI_API_KEY"])
    )

    client.on(StreamingEvents.Begin, on_begin)
    client.on(StreamingEvents.Turn, on_turn)
    client.on(StreamingEvents.Termination, on_terminated)
    client.on(StreamingEvents.Error, on_error)

    client.connect(
        StreamingParameters(speech_model="universal-3-5-pro", sample_rate=SAMPLE_RATE)
    )

    try:
        client.stream(mic_stream())
    except KeyboardInterrupt:
        pass
    finally:
        client.disconnect(terminate=True)


if __name__ == "__main__":
    main()

pip install sounddevice websocket-client

import json
import os
import threading
from urllib.parse import urlencode

import sounddevice as sd
import websocket

API_KEY = os.environ["ASSEMBLYAI_API_KEY"]
SAMPLE_RATE = 16000
CONNECTION_PARAMS = {"speech_model": "universal-3-5-pro", "sample_rate": SAMPLE_RATE}
API_ENDPOINT = f"wss://streaming.assemblyai.com/v3/ws?{urlencode(CONNECTION_PARAMS)}"

stop = threading.Event()


def on_open(ws):
    print("Connected. Speak into your microphone; press Ctrl+C to stop.")

    def stream_audio():
        # sounddevice bundles PortAudio in its wheel — no system install needed.
        with sd.RawInputStream(
            samplerate=SAMPLE_RATE, channels=1, dtype="int16", blocksize=800
        ) as mic:
            while not stop.is_set():
                frames, _ = mic.read(800)  # ~50 ms of audio
                ws.send(bytes(frames), websocket.ABNF.OPCODE_BINARY)

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


def on_message(ws, message):
    data = json.loads(message)
    if data.get("type") == "Turn":
        print(data.get("transcript", ""), end="\n" if data.get("end_of_turn") else "\r")


def on_error(ws, error):
    # On a normal shutdown, websocket-client hands the server's close frame to
    # on_error; ignore it and let on_close report the disconnect. Real failures
    # arrive as exceptions, not close frames.
    if isinstance(error, websocket.ABNF) and error.opcode == websocket.ABNF.OPCODE_CLOSE:
        return
    print(f"\nError: {error}")
    stop.set()


def on_close(ws, status, msg):
    stop.set()
    print("\nDisconnected.")


def main():
    ws = websocket.WebSocketApp(
        API_ENDPOINT,
        header={"Authorization": API_KEY},
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close,
    )

    ws_thread = threading.Thread(target=ws.run_forever, daemon=True)
    ws_thread.start()

    try:
        while ws_thread.is_alive():
            ws_thread.join(0.1)
    except KeyboardInterrupt:
        stop.set()
        if ws.sock and ws.sock.connected:
            ws.send(json.dumps({"type": "Terminate"}))  # close the session
        ws.close()


if __name__ == "__main__":
    main()