> ## Documentation Index
> Fetch the complete documentation index at: https://assemblyai.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Streaming Migration Guide: Deepgram to AssemblyAI

This guide walks through the process of migrating from Deepgram to AssemblyAI for transcribing streaming audio.

### Get Started

Before we begin, make sure you have an AssemblyAI account and an API key. You can [sign up](https://assemblyai.com/dashboard/signup) for a free account and get your API key from your dashboard.

## Side-By-Side Code Comparison

Below is a side-by-side comparison of a basic snippet to transcribe live audio by Deepgram and AssemblyAI using a microphone:

<Tabs groupId="language">
  <Tab language="dg" title="Deepgram">
    ```python expandable theme={null}
    import json, threading, time, websocket, pyaudio, signal
    from urllib.parse import urlencode

    DG_KEY = "YOUR_DG_API_KEY"

    PARAMS = {
        "model": "nova-3",
        "encoding": "linear16",
        "sample_rate": "16000",
        "channels": "1",
        "punctuate": "true",
        "interim_results": "true",
    }

    EP = f"wss://api.deepgram.com/v1/listen?{urlencode(PARAMS)}"

    FRAMES      = 800
    CHANNELS    = 1
    FORMAT      = pyaudio.paInt16
    SAMPLE_RATE = int(PARAMS["sample_rate"])

    audio = stream = ws_app = None
    audio_thread = ws_thread = None
    stop_event = threading.Event()

    # ── WebSocket callbacks ───────────────────────────────────────────────
    def on_open(ws):
        def mic_loop():
            while not stop_event.is_set():
                ws.send(stream.read(FRAMES, exception_on_overflow=False),
                        websocket.ABNF.OPCODE_BINARY)
        global audio_thread
        audio_thread = threading.Thread(target=mic_loop, daemon=True)
        audio_thread.start()

    def on_message(ws, msg):
        global final_metadata
        d = json.loads(msg)
        if d.get("type") == "Results":
            txt = d["channel"]["alternatives"][0]["transcript"]
            if d["is_final"]:
                print(" " * 80, end="\r")
                print(txt)
            else:
                print(txt, end="\r")
        elif d.get("type") == "Metadata":
            final_metadata = d  # save the metadata for later

    def on_error(ws, error):
        print(f"\nWebSocket error: {error}")
        stop_event.set()

    def on_close(ws, *args):
        stop_event.set()
        if final_metadata:
            print("\nFinal Metadata from Deepgram:")
            print(json.dumps(final_metadata, indent=2))
        else:
            print("\nNo metadata received before close.")


        def graceful_shutdown(signum, frame):
            print("\nCtrl+C received → shutting down …")
            stop_event.set()

            # Send CloseStream before closing WebSocket
            if ws_app and ws_app.sock and ws_app.sock.connected:
                try:
                    ws_app.send(json.dumps({"type": "CloseStream"}))
                    time.sleep(0.5)  # Allow time for final metadata to come through
                except Exception as e:
                    print("Error sending CloseStream:", e)
                ws_app.close()

            if ws_thread and ws_thread.is_alive():
                ws_thread.join(timeout=2.0)

    signal.signal(signal.SIGINT, graceful_shutdown)

    def run():
        global audio, stream, ws_app, ws_thread

        # 1. open microphone
        audio = pyaudio.PyAudio()
        try:
            stream = audio.open(format=FORMAT, channels=CHANNELS,
                                rate=SAMPLE_RATE, input=True,
                                frames_per_buffer=FRAMES)
            print("Microphone stream opened. Press Ctrl+C to stop.")
        except Exception as e:
            print(f"Error opening mic: {e}")
            audio.terminate()
            return

        # 2. create WebSocket
        ws_app = websocket.WebSocketApp(
            EP,
            header={"Authorization": f"Token {DG_KEY}"},
            on_open=on_open,
            on_message=on_message,
            on_error=on_error,
            on_close=on_close,
        )

        # 3. start WS thread (only once!)
        ws_thread = threading.Thread(target=ws_app.run_forever, daemon=True)
        ws_thread.start()

        # 4. block main thread until WS thread ends
        ws_thread.join()

        # 5. cleanup
        if stream and stream.is_active():
            stream.stop_stream()
        if stream:
            stream.close()
        if audio:
            audio.terminate()
        print("Cleanup complete. Exiting.")

    if __name__ == "__main__":
        run()
    ```
  </Tab>

  <Tab language="aai" title="AssemblyAI">
    ```python expandable theme={null}
    import pyaudio
    import websocket
    import json
    import threading
    import time
    from urllib.parse import urlencode
    from datetime import datetime

    # --- Configuration ---
    YOUR_API_KEY = "YOUR-API-KEY"  # Replace with your actual API key

    CONNECTION_PARAMS = {
        "sample_rate": 16000,
        "speech_model": "u3-rt-pro",
    }
    API_ENDPOINT_BASE_URL = "wss://streaming.assemblyai.com/v3/ws"
    API_ENDPOINT = f"{API_ENDPOINT_BASE_URL}?{urlencode(CONNECTION_PARAMS)}"

    # Audio Configuration
    FRAMES_PER_BUFFER = 800  # 50ms of audio (0.05s * 16000Hz)
    SAMPLE_RATE = CONNECTION_PARAMS["sample_rate"]
    CHANNELS = 1
    FORMAT = pyaudio.paInt16

    # Global variables for audio stream and websocket
    audio = None
    stream = None
    ws_app = None
    audio_thread = None
    stop_event = threading.Event()  # To signal the audio thread to stop

    # --- WebSocket Event Handlers ---

    def on_open(ws):
        """Called when the WebSocket connection is established."""
        print("WebSocket connection opened.")
        print(f"Connected to: {API_ENDPOINT}")

        # Start sending audio data in a separate thread
        def stream_audio():
            global stream
            print("Starting audio streaming...")
            while not stop_event.is_set():
                try:
                    audio_data = stream.read(FRAMES_PER_BUFFER, exception_on_overflow=False)
                    # Send audio data as binary message
                    ws.send(audio_data, websocket.ABNF.OPCODE_BINARY)
                except Exception as e:
                    print(f"Error streaming audio: {e}")
                    # If stream read fails, likely means it's closed, stop the loop
                    break
            print("Audio streaming stopped.")

        global audio_thread
        audio_thread = threading.Thread(target=stream_audio)
        audio_thread.daemon = (
            True  # Allow main thread to exit even if this thread is running
        )
        audio_thread.start()

    def on_message(ws, message):
        try:
            data = json.loads(message)
            msg_type = data.get('type')

            if msg_type == "Begin":
                session_id = data.get('id')
                expires_at = data.get('expires_at')
                print(f"\nSession began: ID={session_id}, ExpiresAt={datetime.fromtimestamp(expires_at)}")
            elif msg_type == "Turn":
                transcript = data.get('transcript', '')
                end_of_turn = data.get('end_of_turn', False)

                # Print final end-of-turn transcript
                if end_of_turn:
                    print('\r' + ' ' * 80 + '\r', end='')
                    print(transcript)
                else:
                    print(f"\r{transcript}", end='')
            elif msg_type == "Termination":
                audio_duration = data.get('audio_duration_seconds', 0)
                session_duration = data.get('session_duration_seconds', 0)
                print(f"\nSession Terminated: Audio Duration={audio_duration}s, Session Duration={session_duration}s")
        except json.JSONDecodeError as e:
            print(f"Error decoding message: {e}")
        except Exception as e:
            print(f"Error handling message: {e}")

    def on_error(ws, error):
        """Called when a WebSocket error occurs."""
        print(f"\nWebSocket Error: {error}")
        # Attempt to signal stop on error
        stop_event.set()

    def on_close(ws, close_status_code, close_msg):
        """Called when the WebSocket connection is closed."""
        print(f"\nWebSocket Disconnected: Status={close_status_code}, Msg={close_msg}")
        # Ensure audio resources are released
        global stream, audio
        stop_event.set()  # Signal audio thread just in case it's still running

        if stream:
            if stream.is_active():
                stream.stop_stream()
            stream.close()
            stream = None
        if audio:
            audio.terminate()
            audio = None
        # Try to join the audio thread to ensure clean exit
        if audio_thread and audio_thread.is_alive():
            audio_thread.join(timeout=1.0)

    # --- Main Execution ---
    def run():
        global audio, stream, ws_app

        # Initialize PyAudio
        audio = pyaudio.PyAudio()

        # Open microphone stream
        try:
            stream = audio.open(
                input=True,
                frames_per_buffer=FRAMES_PER_BUFFER,
                channels=CHANNELS,
                format=FORMAT,
                rate=SAMPLE_RATE,
            )
            print("Microphone stream opened successfully.")
            print("Speak into your microphone. Press Ctrl+C to stop.")
        except Exception as e:
            print(f"Error opening microphone stream: {e}")
            if audio:
                audio.terminate()
            return  # Exit if microphone cannot be opened

        # Create WebSocketApp
        ws_app = websocket.WebSocketApp(
            API_ENDPOINT,
            header={"Authorization": YOUR_API_KEY},
            on_open=on_open,
            on_message=on_message,
            on_error=on_error,
            on_close=on_close,
        )

        # Run WebSocketApp in a separate thread to allow main thread to catch KeyboardInterrupt
        ws_thread = threading.Thread(target=ws_app.run_forever)
        ws_thread.daemon = True
        ws_thread.start()

        try:
            # Keep main thread alive until interrupted
            while ws_thread.is_alive():
                time.sleep(0.1)
        except KeyboardInterrupt:
            print("\nCtrl+C received. Stopping...")
            stop_event.set()  # Signal audio thread to stop

            # Send termination message to the server
            if ws_app and ws_app.sock and ws_app.sock.connected:
                try:
                    terminate_message = {"type": "Terminate"}
                    print(f"Sending termination message: {json.dumps(terminate_message)}")
                    ws_app.send(json.dumps(terminate_message))
                    # Give a moment for messages to process before forceful close
                    time.sleep(5)
                except Exception as e:
                    print(f"Error sending termination message: {e}")

            # Close the WebSocket connection (will trigger on_close)
            if ws_app:
                ws_app.close()

            # Wait for WebSocket thread to finish
            ws_thread.join(timeout=2.0)

        except Exception as e:
            print(f"\nAn unexpected error occurred: {e}")
            stop_event.set()
            if ws_app:
                ws_app.close()
            ws_thread.join(timeout=2.0)

        finally:
            # Final cleanup (already handled in on_close, but good as a fallback)
            if stream and stream.is_active():
                stream.stop_stream()
            if stream:
                stream.close()
            if audio:
                audio.terminate()
            print("Cleanup complete. Exiting.")

    if __name__ == "__main__":
        run()
    ```
  </Tab>
</Tabs>

## Authentication

<Tabs groupId="language">
  <Tab language="dg" title="Deepgram">
    ```python theme={null}
    import json, threading, time, websocket, pyaudio, signal
    from urllib.parse import urlencode

    DG_KEY = "YOUR_DG_API_KEY"
    ```
  </Tab>

  <Tab language="aai" title="AssemblyAI">
    ```python theme={null}
    import pyaudio
    import websocket
    import json
    import threading
    import time
    from urllib.parse import urlencode
    from datetime import datetime

    YOUR_API_KEY = "YOUR-API-KEY"
    ```
  </Tab>
</Tabs>

When migrating from Deepgram to AssemblyAI, you'll first need to handle authentication:

Get your API key from your [AssemblyAI dashboard](https://www.assemblyai.com/app/api-keys)

<Note>
  **Protect Your API Key**

  For improved security, store your API key as an environment variable.
</Note>

## Connection Parameters & Microphone Setup

<Tabs groupId="language">
  <Tab language="dg" title="Deepgram">
    ```python theme={null}
    PARAMS = {
        "model": "nova-3",
        "encoding": "linear16",
        "sample_rate": "16000",
        "channels": "1",
        "punctuate": "true",
        "interim_results": "true",
    }

    EP = f"wss://api.deepgram.com/v1/listen?{urlencode(PARAMS)}"

    FRAMES = 800
    CHANNELS = 1
    FORMAT = pyaudio.paInt16
    SAMPLE_RATE = int(PARAMS["sample_rate"])

    audio = stream = ws_app = None
    audio_thread = ws_thread = None
    stop_event = threading.Event()
    ```
  </Tab>

  <Tab language="aai" title="AssemblyAI">
    ```python theme={null}
    CONNECTION_PARAMS = {
        "sample_rate": 16000,
        "speech_model": "u3-rt-pro",
    }
    API_ENDPOINT_BASE_URL = "wss://streaming.assemblyai.com/v3/ws"
    API_ENDPOINT = f"{API_ENDPOINT_BASE_URL}?{urlencode(CONNECTION_PARAMS)}"

    # Audio Configuration
    FRAMES_PER_BUFFER = 800  # 50ms of audio (0.05s * 16000Hz)
    SAMPLE_RATE = CONNECTION_PARAMS["sample_rate"]
    CHANNELS = 1
    FORMAT = pyaudio.paInt16

    # Global variables for audio stream and websocket
    audio = None
    stream = None
    ws_app = None
    audio_thread = None
    stop_event = threading.Event()  # To signal the audio thread to stop
    ```
  </Tab>
</Tabs>

Here are helpful things to know about connecting our streaming model:

* **Universal-3 Pro model** — Connect to `wss://streaming.assemblyai.com/v3/ws` with `speech_model=u3-rt-pro` to use our latest, highest-accuracy streaming model. Unlike Deepgram’s `model="nova-3"`, you use a single `speech_model` parameter.

* **Built-in formatting** — Universal-3 Pro always returns formatted transcripts with smart punctuation & casing, similar to Deepgram’s `punctuate=true`. No extra parameter is needed.

* **Partials are always on — like Deepgram’s `interim_results=true`** — AssemblyAI streams interim results automatically. Universal-3 Pro emits partials during periods of silence, with at most one partial per silence period.

## Opening the WebSocket

<Tabs groupId="language">
  <Tab language="dg" title="Deepgram">
    ```python theme={null}
    def on_open(ws):
        def mic_loop():
            while not stop_event.is_set():
                ws.send(stream.read(FRAMES, exception_on_overflow=False),
                        websocket.ABNF.OPCODE_BINARY)
        global audio_thread
        audio_thread = threading.Thread(target=mic_loop, daemon=True)
        audio_thread.start()
    ```
  </Tab>

  <Tab language="aai" title="AssemblyAI">
    ```python expandable theme={null}
    def on_open(ws):
        """Called when the WebSocket connection is established."""
        print("WebSocket connection opened.")
        print(f"Connected to: {API_ENDPOINT}")

        # Start sending audio data in a separate thread
        def stream_audio():
            global stream
            print("Starting audio streaming...")
            while not stop_event.is_set():
                try:
                    audio_data = stream.read(FRAMES_PER_BUFFER, exception_on_overflow=False)
                    # Send audio data as binary message
                    ws.send(audio_data, websocket.ABNF.OPCODE_BINARY)
                except Exception as e:
                    print(f"Error streaming audio: {e}")
                    # If stream read fails, likely means it's closed, stop the loop
                    break
            print("Audio streaming stopped.")

        global audio_thread
        audio_thread = threading.Thread(target=stream_audio)
        audio_thread.daemon = (
            True  # Allow main thread to exit even if this thread is running
        )
        audio_thread.start()
    ```
  </Tab>
</Tabs>

<Note>
  **Tip:** Adding error-handling and log lines (as in the AssemblyAI snippet) lets you see exactly when the socket opens, audio starts, or a read fails—catching issues early saving time debugging silent failures.
</Note>

## Receiving Messsages from the WebSocket

<Tabs groupId="language">
  <Tab language="dg" title="Deepgram">
    ```python theme={null}
    def on_message(ws, msg):
        global final_metadata
        d = json.loads(msg)
        if d.get("type") == "Results":
            txt = d["channel"]["alternatives"][0]["transcript"]
            if d["is_final"]:
                print(" " * 80, end="\r")
                print(txt)
            else:
                print(txt, end="\r")
        elif d.get("type") == "Metadata":
            final_metadata = d  # save the metadata for later
    ```
  </Tab>

  <Tab language="aai" title="AssemblyAI">
    ```python expandable theme={null}
    def on_message(ws, message):
        try:
            data = json.loads(message)
            msg_type = data.get(‘type’)

            if msg_type == "Begin":
                session_id = data.get(‘id’)
                expires_at = data.get(‘expires_at’)
                print(f"\nSession began: ID={session_id}, ExpiresAt={datetime.fromtimestamp(expires_at)}")
            elif msg_type == "Turn":
                transcript = data.get(‘transcript’, ‘’)
                end_of_turn = data.get(‘end_of_turn’, False)

                # Print final end-of-turn transcript
                if end_of_turn:
                    print(‘\r’ + ‘ ‘ * 80 + ‘\r’, end=’’)
                    print(transcript)
                else:
                    print(f"\r{transcript}", end=’’)
            elif msg_type == "Termination":
                audio_duration = data.get(‘audio_duration_seconds’, 0)
                session_duration = data.get(‘session_duration_seconds’, 0)
                print(f"\nSession Terminated: Audio Duration={audio_duration}s, Session Duration={session_duration}s")
        except json.JSONDecodeError as e:
            print(f"Error decoding message: {e}")
        except Exception as e:
            print(f"Error handling message: {e}")
    ```
  </Tab>
</Tabs>

Helpful things to know about AssemblyAI’s message payloads:

* **Clear message types** – Instead of checking `is_final`, you’ll receive explicit `"Begin"`, `"Turn"`, and `"Termination"` events, making your logic simpler and more readable.

* **Session metadata up-front** – The first `"Begin"` message delivers a `session_id` and expiry timestamp. You can log or surface these for tracing or billing.

* **End-of-turn detection** – Each `"Turn"` object includes an `end_of_turn` boolean. When `end_of_turn` is `true`, the transcript is a final, formatted result. When `false`, it is a partial transcript. Universal-3 Pro always returns formatted transcripts with smart punctuation & casing built in.

## Handling Errors

<Tabs groupId="language">
  <Tab language="dg" title="Deepgram">
    ```python theme={null}
    def on_error(ws, error):
        print(f"\nWebSocket error: {error}")
        stop_event.set()
    ```
  </Tab>

  <Tab language="aai" title="AssemblyAI">
    ```python theme={null}
    def on_error(ws, error):
        """Called when a WebSocket error occurs."""
        print(f'\nWebSocket Error: {error}')

        # Attempt to signal stop on error
        stop_event.set()
    ```
  </Tab>
</Tabs>

Capture and log any errors emitted by the WebSocket connection to streamline troubleshooting and maintain smooth operation.

## Closing the WebSocket

<Tabs groupId="language">
  <Tab language="dg" title="Deepgram">
    ```python theme={null}
    def on_close(ws, *args):
        stop_event.set()
        if final_metadata:
            print("\nFinal Metadata from Deepgram:")
            print(json.dumps(final_metadata, indent=2))
        else:
            print("\nNo metadata received before close.")
    ```
  </Tab>

  <Tab language="aai" title="AssemblyAI">
    ```python theme={null}
    def on_close(ws, close_status_code, close_msg):
        """Called when the WebSocket connection is closed."""
        print(f"\nWebSocket Disconnected: Status={close_status_code}, Msg={close_msg}")
        # Ensure audio resources are released
        global stream, audio
        stop_event.set()  # Signal audio thread just in case it's still running

        if stream:
            if stream.is_active():
                stream.stop_stream()
            stream.close()
            stream = None
        if audio:
            audio.terminate()
            audio = None
        # Try to join the audio thread to ensure clean exit
        if audio_thread and audio_thread.is_alive():
            audio_thread.join(timeout=1.0)
    ```
  </Tab>
</Tabs>

Helpful things to know about AssemblyAI’s WebSocket Closure:

* **Connection diagnostics on tap** - If the socket closes unexpectedly, AssemblyAI supplies both a status code and a reason message (close\_status\_code, close\_msg), so you know immediately whether the server timed out, refused auth, or encountered another error.

* **Metadata arrives at session start, not at close** - Deepgram sends its final metadata only when the socket closes. AssemblyAI delivers session information up front in the initial "Begin" message, so you can log IDs and expiry times right away.

## Opening the Microphone Stream & Creating a WebSocket

<Tabs groupId="language">
  <Tab language="dg" title="Deepgram">
    ```python expandable theme={null}
    global audio, stream, ws_app, ws_thread

    # 1. open microphone
    audio = pyaudio.PyAudio()
    try:
        stream = audio.open(format=FORMAT, channels=CHANNELS,
                            rate=SAMPLE_RATE, input=True,
                            frames_per_buffer=FRAMES)
        print("Microphone stream opened. Press Ctrl+C to stop.")
    except Exception as e:
        print(f"Error opening mic: {e}")
        audio.terminate()
        return

    # 2. create WebSocket
    ws_app = websocket.WebSocketApp(
        EP,
        header={"Authorization": f"Token {DG_KEY}"},
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close,
    )

    ws_thread = threading.Thread(target=ws_app.run_forever, daemon=True)
    ws_thread.start()
    ws_thread.join()
    ```
  </Tab>

  <Tab language="aai" title="AssemblyAI">
    ```python expandable theme={null}
    global audio, stream, ws_app
    # Initialize PyAudio
    audio = pyaudio.PyAudio()

    # Open microphone stream
    try:
        stream = audio.open(
            input=True,
            frames_per_buffer=FRAMES_PER_BUFFER,
            channels=CHANNELS,
            format=FORMAT,
            rate=SAMPLE_RATE,
        )
        print("Microphone stream opened successfully.")
        print("Speak into your microphone. Press Ctrl+C to stop.")
    except Exception as e:
        print(f"Error opening microphone stream: {e}")
        if audio:
            audio.terminate()
        return  # Exit if microphone cannot be opened

    # Create WebSocketApp
    ws_app = websocket.WebSocketApp(
        API_ENDPOINT,
        header={"Authorization": YOUR_API_KEY},
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close,
    )
    # Run WebSocketApp in a separate thread to allow main thread to catch KeyboardInterrupt
    ws_thread = threading.Thread(target=ws_app.run_forever)
    ws_thread.daemon = True
    ws_thread.start()
    ```
  </Tab>
</Tabs>

## Session Shutdown

<Tabs groupId="language">
  <Tab language="dg" title="Deepgram">
    ```python expandable theme={null}
    def graceful_shutdown(signum, frame):
        print("\nCtrl+C received → shutting down …")
        stop_event.set()

        # Send CloseStream before closing WebSocket
        if ws_app and ws_app.sock and ws_app.sock.connected:
            try:
                ws_app.send(json.dumps({"type": "CloseStream"}))
                time.sleep(0.5)  # Allow time for final metadata to come through
            except Exception as e:
                print("Error sending CloseStream:", e)
            ws_app.close()

        if ws_thread and ws_thread.is_alive():
            ws_thread.join(timeout=2.0)

    signal.signal(signal.SIGINT, graceful_shutdown)

    def run():
        # Open microphone and create WebSocket

        # cleanup
        if stream and stream.is_active():
            stream.stop_stream()
        if stream:
            stream.close()
        if audio:
            audio.terminate()
        print("Cleanup complete. Exiting.")
    ```
  </Tab>

  <Tab language="aai" title="AssemblyAI">
    ```python expandable theme={null}
    try:
        # Keep main thread alive until interrupted
        while ws_thread.is_alive():
            time.sleep(0.1)
    except KeyboardInterrupt:
        print("\nCtrl+C received. Stopping...")
        stop_event.set()  # Signal audio thread to stop

        # Send termination message to the server
        if ws_app and ws_app.sock and ws_app.sock.connected:
            try:
                terminate_message = {"type": "Terminate"}
                print(f"Sending termination message: {json.dumps(terminate_message)}")
                ws_app.send(json.dumps(terminate_message))
                # Give a moment for messages to process before forceful close
                time.sleep(5)
            except Exception as e:
                print(f"Error sending termination message: {e}")

        # Close the WebSocket connection (will trigger on_close)
        if ws_app:
            ws_app.close()

        # Wait for WebSocket thread to finish
        ws_thread.join(timeout=2.0)

    except Exception as e:
        print(f"\nAn unexpected error occurred: {e}")
        stop_event.set()
        if ws_app:
            ws_app.close()
        ws_thread.join(timeout=2.0)

    finally:
        # Final cleanup (already handled in on_close, but good as a fallback)
        if stream and stream.is_active():
            stream.stop_stream()
        if stream:
            stream.close()
        if audio:
            audio.terminate()
        print("Cleanup complete. Exiting.")
    ```
  </Tab>
</Tabs>

Helpful things to know about AssemblyAI’s shutdown:

* **JSON payload difference** - When closing the stream with AssemblyAI, your JSON payload will be `{"type: "Terminate" }` instead of `{"type: "CloseStream" }`

* **No metadata race condition** - Because AssemblyAI already provided session info at "Begin" and doesn’t append extra data at shutdown, you don’t have to sleep (time.sleep(0.5)) to wait for “final metadata” before closing—making the exit faster and less error-prone.

## Resources

For additional information about using AssemblyAI's Streaming Speech-To-Text API you can also refer to:

* [Universal-3 Pro Streaming](/streaming/getting-started/transcribe-streaming-audio)
* [Streaming Guide](/streaming/getting-started/transcribe-streaming-audio)