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 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:

1import json, threading, time, websocket, pyaudio, signal
2from urllib.parse import urlencode
3
4DG_KEY = "YOUR_DG_API_KEY"
5
6PARAMS = {
7 "model": "nova-3",
8 "encoding": "linear16",
9 "sample_rate": "16000",
10 "channels": "1",
11 "punctuate": "true",
12 "interim_results": "true",
13}
14
15EP = f"wss://api.deepgram.com/v1/listen?{urlencode(PARAMS)}"
16
17FRAMES = 800
18CHANNELS = 1
19FORMAT = pyaudio.paInt16
20SAMPLE_RATE = int(PARAMS["sample_rate"])
21
22audio = stream = ws_app = None
23audio_thread = ws_thread = None
24stop_event = threading.Event()
25
26# ── WebSocket callbacks ───────────────────────────────────────────────
27def on_open(ws):
28 def mic_loop():
29 while not stop_event.is_set():
30 ws.send(stream.read(FRAMES, exception_on_overflow=False),
31 websocket.ABNF.OPCODE_BINARY)
32 global audio_thread
33 audio_thread = threading.Thread(target=mic_loop, daemon=True)
34 audio_thread.start()
35
36def on_message(ws, msg):
37 global final_metadata
38 d = json.loads(msg)
39 if d.get("type") == "Results":
40 txt = d["channel"]["alternatives"][0]["transcript"]
41 if d["is_final"]:
42 print(" " * 80, end="\r")
43 print(txt)
44 else:
45 print(txt, end="\r")
46 elif d.get("type") == "Metadata":
47 final_metadata = d # save the metadata for later
48
49def on_error(ws, error):
50 print(f"\nWebSocket error: {error}")
51 stop_event.set()
52
53def on_close(ws, *args):
54 stop_event.set()
55 if final_metadata:
56 print("\nFinal Metadata from Deepgram:")
57 print(json.dumps(final_metadata, indent=2))
58 else:
59 print("\nNo metadata received before close.")
60
61
62 def graceful_shutdown(signum, frame):
63 print("\nCtrl+C received → shutting down …")
64 stop_event.set()
65
66 # Send CloseStream before closing WebSocket
67 if ws_app and ws_app.sock and ws_app.sock.connected:
68 try:
69 ws_app.send(json.dumps({"type": "CloseStream"}))
70 time.sleep(0.5) # Allow time for final metadata to come through
71 except Exception as e:
72 print("Error sending CloseStream:", e)
73 ws_app.close()
74
75 if ws_thread and ws_thread.is_alive():
76 ws_thread.join(timeout=2.0)
77
78signal.signal(signal.SIGINT, graceful_shutdown)
79
80def run():
81 global audio, stream, ws_app, ws_thread
82
83 # 1. open microphone
84 audio = pyaudio.PyAudio()
85 try:
86 stream = audio.open(format=FORMAT, channels=CHANNELS,
87 rate=SAMPLE_RATE, input=True,
88 frames_per_buffer=FRAMES)
89 print("Microphone stream opened. Press Ctrl+C to stop.")
90 except Exception as e:
91 print(f"Error opening mic: {e}")
92 audio.terminate()
93 return
94
95 # 2. create WebSocket
96 ws_app = websocket.WebSocketApp(
97 EP,
98 header={"Authorization": f"Token {DG_KEY}"},
99 on_open=on_open,
100 on_message=on_message,
101 on_error=on_error,
102 on_close=on_close,
103 )
104
105 # 3. start WS thread (only once!)
106 ws_thread = threading.Thread(target=ws_app.run_forever, daemon=True)
107 ws_thread.start()
108
109 # 4. block main thread until WS thread ends
110 ws_thread.join()
111
112 # 5. cleanup
113 if stream and stream.is_active():
114 stream.stop_stream()
115 if stream:
116 stream.close()
117 if audio:
118 audio.terminate()
119 print("Cleanup complete. Exiting.")
120
121if __name__ == "__main__":
122 run()

Authentication

1import json, threading, time, websocket, pyaudio, signal
2from urllib.parse import urlencode
3
4DG_KEY = "YOUR_DG_API_KEY"

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

Get your API key from your AssemblyAI dashboard

Protect Your API Key

For improved security, store your API key as an environment variable.

Connection Parameters & Microphone Setup

1PARAMS = {
2 "model": "nova-3",
3 "encoding": "linear16",
4 "sample_rate": "16000",
5 "channels": "1",
6 "punctuate": "true",
7 "interim_results": "true",
8}
9
10EP = f"wss://api.deepgram.com/v1/listen?{urlencode(PARAMS)}"
11
12FRAMES = 800
13CHANNELS = 1
14FORMAT = pyaudio.paInt16
15SAMPLE_RATE = int(PARAMS["sample_rate"])
16
17audio = stream = ws_app = None
18audio_thread = ws_thread = None
19stop_event = threading.Event()

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

  • One universal model – no model param needed - Just connect to wss://streaming.assemblyai.com/v3/ws. The live endpoint always uses our latest, best-performing model, so you can drop Deepgram’s model="nova-3" line entirely.

  • format_turns=True ≈ Deepgram’s punctuate=true - When the flag is set to True, every Final message arrives with smart punctuation & casing. Set it to False (or omit it) to get raw, lowercase text — useful if you do your own formatting.

  • Partials are always on - like Deepgram’s interim_results=true - AssemblyAI streams immutable interim results automatically. There’s no switch to toggle. Expect fast, token-level updates that refine until each Final is emitted.

Opening the WebSocket

1def on_open(ws):
2 def mic_loop():
3 while not stop_event.is_set():
4 ws.send(stream.read(FRAMES, exception_on_overflow=False),
5 websocket.ABNF.OPCODE_BINARY)
6 global audio_thread
7 audio_thread = threading.Thread(target=mic_loop, daemon=True)
8 audio_thread.start()

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.

Receiving Messsages from the WebSocket

1def on_message(ws, msg):
2 global final_metadata
3 d = json.loads(msg)
4 if d.get("type") == "Results":
5 txt = d["channel"]["alternatives"][0]["transcript"]
6 if d["is_final"]:
7 print(" " * 80, end="\r")
8 print(txt)
9 else:
10 print(txt, end="\r")
11 elif d.get("type") == "Metadata":
12 final_metadata = d # save the metadata for later

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.

  • Formatted vs. raw finals – Each "Turn" object includes a boolean turn_is_formatted. When you set format_turns to True, punctuation/casing appears in the Final Transcript, so you can toggle display styles on the fly.

Handling Errors

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

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

Closing the WebSocket

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

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

1global audio, stream, ws_app, ws_thread
2
3# 1. open microphone
4audio = pyaudio.PyAudio()
5try:
6 stream = audio.open(format=FORMAT, channels=CHANNELS,
7 rate=SAMPLE_RATE, input=True,
8 frames_per_buffer=FRAMES)
9 print("Microphone stream opened. Press Ctrl+C to stop.")
10except Exception as e:
11 print(f"Error opening mic: {e}")
12 audio.terminate()
13 return
14
15# 2. create WebSocket
16ws_app = websocket.WebSocketApp(
17 EP,
18 header={"Authorization": f"Token {DG_KEY}"},
19 on_open=on_open,
20 on_message=on_message,
21 on_error=on_error,
22 on_close=on_close,
23)
24
25ws_thread = threading.Thread(target=ws_app.run_forever, daemon=True)
26ws_thread.start()
27ws_thread.join()

Session Shutdown

1def graceful_shutdown(signum, frame):
2 print("\nCtrl+C received → shutting down …")
3 stop_event.set()
4
5 # Send CloseStream before closing WebSocket
6 if ws_app and ws_app.sock and ws_app.sock.connected:
7 try:
8 ws_app.send(json.dumps({"type": "CloseStream"}))
9 time.sleep(0.5) # Allow time for final metadata to come through
10 except Exception as e:
11 print("Error sending CloseStream:", e)
12 ws_app.close()
13
14 if ws_thread and ws_thread.is_alive():
15 ws_thread.join(timeout=2.0)
16
17signal.signal(signal.SIGINT, graceful_shutdown)
18
19def run():
20 # Open microphone and create WebSocket
21
22 # cleanup
23 if stream and stream.is_active():
24 stream.stop_stream()
25 if stream:
26 stream.close()
27 if audio:
28 audio.terminate()
29 print("Cleanup complete. Exiting.")

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" }

  • Server handles idle timeouts - If you forget to send "Terminate", AssemblyAI will close the socket after an idle window of one minute.

  • 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: