fix(websockets): switch socket clients to construct_type for unknown message tolerance

Replace parse_obj_as with construct_type in all socket clients so unknown
message types are silently coerced instead of raising ValidationError.
Also fix v2 listen example to actually send audio.

Author: lukeocodes

examples/14-transcription-live-websocket-v2.py

@@ -5,6 +5,10 @@
 with contextual turn detection.
 """
 
+import os
+import threading
+import time
+from pathlib import Path
 from typing import Union
 
 from dotenv import load_dotenv
@@ -14,42 +18,59 @@
 from deepgram import DeepgramClient
 from deepgram.core.events import EventType
 from deepgram.listen.v2.types import (
+    ListenV2CloseStream,
     ListenV2Connected,
     ListenV2FatalError,
     ListenV2TurnInfo,
 )
 
 ListenV2SocketClientResponse = Union[ListenV2Connected, ListenV2TurnInfo, ListenV2FatalError]
 
-client = DeepgramClient()
+client = DeepgramClient(api_key=os.environ.get("DEEPGRAM_API_KEY"))
 
 try:
-    with client.listen.v2.connect(model="flux-general-en", encoding="linear16", sample_rate=16000) as connection:
+    with client.listen.v2.connect(
+        model="flux-general-en",
+        encoding="linear16",
+        sample_rate="16000",
+    ) as connection:
 
         def on_message(message: ListenV2SocketClientResponse) -> None:
-            msg_type = getattr(message, "type", "Unknown")
-            print(f"Received {msg_type} event")
+            msg_type = getattr(message, "type", type(message).__name__)
+            print(f"Received {msg_type} event ({type(message).__name__})")
 
             # Extract transcription from TurnInfo events
             if isinstance(message, ListenV2TurnInfo):
-                print(f"Turn transcript: {message.transcript}")
-                print(f"Turn event: {message.event}")
-                print(f"Turn index: {message.turn_index}")
+                print(f"  transcript: {message.transcript}")
+                print(f"  event: {message.event}")
+                print(f"  turn_index: {message.turn_index}")
 
         connection.on(EventType.OPEN, lambda _: print("Connection opened"))
         connection.on(EventType.MESSAGE, on_message)
         connection.on(EventType.CLOSE, lambda _: print("Connection closed"))
-        connection.on(EventType.ERROR, lambda error: print(f"Error: {error}"))
+        connection.on(EventType.ERROR, lambda error: print(f"Error: {type(error).__name__}: {error}"))
 
-        # Start listening - this blocks until the connection closes
-        # In production, you would send audio data here using connection.send_media()
-        connection.start_listening()
+        # Send audio in a background thread so start_listening can process responses
+        def send_audio():
+            audio_path = Path(__file__).parent / "fixtures" / "audio.wav"
+            with open(audio_path, "rb") as f:
+                audio = f.read()
+
+            # Send in chunks
+            chunk_size = 4096
+            for i in range(0, len(audio), chunk_size):
+                connection.send_media(audio[i : i + chunk_size])
+                time.sleep(0.01)  # pace the sending
+
+            # Signal end of audio
+            time.sleep(2)
+            connection.send_close_stream(ListenV2CloseStream(type="CloseStream"))
 
-    # For async version:
-    # from deepgram import AsyncDeepgramClient
-    # async with client.listen.v2.connect(...) as connection:
-    #     # ... same event handlers ...
-    #     await connection.start_listening()
+        sender = threading.Thread(target=send_audio, daemon=True)
+        sender.start()
+
+        # This blocks until the connection closes
+        connection.start_listening()
 
 except Exception as e:
-    print(f"Error: {e}")
+    print(f"Error: {type(e).__name__}: {e}")

src/deepgram/agent/v1/socket_client.py

@@ -7,7 +7,7 @@
 import websockets
 import websockets.sync.connection as websockets_sync_connection
 from ...core.events import EventEmitterMixin, EventType
-from ...core.pydantic_utilities import parse_obj_as
+from ...core.unchecked_base_model import construct_type
 from .types.agent_v1agent_audio_done import AgentV1AgentAudioDone
 from .types.agent_v1agent_started_speaking import AgentV1AgentStartedSpeaking
 from .types.agent_v1agent_thinking import AgentV1AgentThinking
@@ -64,7 +64,7 @@ async def __aiter__(self):
             if isinstance(message, bytes):
                 yield message
             else:
-                yield parse_obj_as(V1SocketClientResponse, json.loads(message))  # type: ignore
+                yield construct_type(type_=V1SocketClientResponse, object_=json.loads(message))  # type: ignore
 
     async def start_listening(self):
         """
@@ -83,7 +83,7 @@ async def start_listening(self):
                     parsed = raw_message
                 else:
                     json_data = json.loads(raw_message)
-                    parsed = parse_obj_as(V1SocketClientResponse, json_data)  # type: ignore
+                    parsed = construct_type(type_=V1SocketClientResponse, object_=json_data)  # type: ignore
                 await self._emit_async(EventType.MESSAGE, parsed)
         except Exception as exc:
             await self._emit_async(EventType.ERROR, exc)
@@ -154,7 +154,7 @@ async def recv(self) -> V1SocketClientResponse:
         if isinstance(data, bytes):
             return data  # type: ignore
         json_data = json.loads(data)
-        return parse_obj_as(V1SocketClientResponse, json_data)  # type: ignore
+        return construct_type(type_=V1SocketClientResponse, object_=json_data)  # type: ignore
 
     async def _send(self, data: typing.Any) -> None:
         """
@@ -181,7 +181,7 @@ def __iter__(self):
             if isinstance(message, bytes):
                 yield message
             else:
-                yield parse_obj_as(V1SocketClientResponse, json.loads(message))  # type: ignore
+                yield construct_type(type_=V1SocketClientResponse, object_=json.loads(message))  # type: ignore
 
     def start_listening(self):
         """
@@ -200,7 +200,7 @@ def start_listening(self):
                     parsed = raw_message
                 else:
                     json_data = json.loads(raw_message)
-                    parsed = parse_obj_as(V1SocketClientResponse, json_data)  # type: ignore
+                    parsed = construct_type(type_=V1SocketClientResponse, object_=json_data)  # type: ignore
                 self._emit(EventType.MESSAGE, parsed)
         except Exception as exc:
             self._emit(EventType.ERROR, exc)
@@ -271,7 +271,7 @@ def recv(self) -> V1SocketClientResponse:
         if isinstance(data, bytes):
             return data  # type: ignore
         json_data = json.loads(data)
-        return parse_obj_as(V1SocketClientResponse, json_data)  # type: ignore
+        return construct_type(type_=V1SocketClientResponse, object_=json_data)  # type: ignore
 
     def _send(self, data: typing.Any) -> None:
         """

src/deepgram/listen/v1/socket_client.py

@@ -7,7 +7,7 @@
 import websockets
 import websockets.sync.connection as websockets_sync_connection
 from ...core.events import EventEmitterMixin, EventType
-from ...core.pydantic_utilities import parse_obj_as
+from ...core.unchecked_base_model import construct_type
 from .types.listen_v1close_stream import ListenV1CloseStream
 from .types.listen_v1finalize import ListenV1Finalize
 from .types.listen_v1keep_alive import ListenV1KeepAlive
@@ -34,7 +34,7 @@ async def __aiter__(self):
             if isinstance(message, bytes):
                 yield message
             else:
-                yield parse_obj_as(V1SocketClientResponse, json.loads(message))  # type: ignore
+                yield construct_type(type_=V1SocketClientResponse, object_=json.loads(message))  # type: ignore
 
     async def start_listening(self):
         """
@@ -53,7 +53,7 @@ async def start_listening(self):
                     parsed = raw_message
                 else:
                     json_data = json.loads(raw_message)
-                    parsed = parse_obj_as(V1SocketClientResponse, json_data)  # type: ignore
+                    parsed = construct_type(type_=V1SocketClientResponse, object_=json_data)  # type: ignore
                 await self._emit_async(EventType.MESSAGE, parsed)
         except Exception as exc:
             await self._emit_async(EventType.ERROR, exc)
@@ -96,7 +96,7 @@ async def recv(self) -> V1SocketClientResponse:
         if isinstance(data, bytes):
             return data  # type: ignore
         json_data = json.loads(data)
-        return parse_obj_as(V1SocketClientResponse, json_data)  # type: ignore
+        return construct_type(type_=V1SocketClientResponse, object_=json_data)  # type: ignore
 
     async def _send(self, data: typing.Any) -> None:
         """
@@ -123,7 +123,7 @@ def __iter__(self):
             if isinstance(message, bytes):
                 yield message
             else:
-                yield parse_obj_as(V1SocketClientResponse, json.loads(message))  # type: ignore
+                yield construct_type(type_=V1SocketClientResponse, object_=json.loads(message))  # type: ignore
 
     def start_listening(self):
         """
@@ -142,7 +142,7 @@ def start_listening(self):
                     parsed = raw_message
                 else:
                     json_data = json.loads(raw_message)
-                    parsed = parse_obj_as(V1SocketClientResponse, json_data)  # type: ignore
+                    parsed = construct_type(type_=V1SocketClientResponse, object_=json_data)  # type: ignore
                 self._emit(EventType.MESSAGE, parsed)
         except Exception as exc:
             self._emit(EventType.ERROR, exc)
@@ -185,7 +185,7 @@ def recv(self) -> V1SocketClientResponse:
         if isinstance(data, bytes):
             return data  # type: ignore
         json_data = json.loads(data)
-        return parse_obj_as(V1SocketClientResponse, json_data)  # type: ignore
+        return construct_type(type_=V1SocketClientResponse, object_=json_data)  # type: ignore
 
     def _send(self, data: typing.Any) -> None:
         """

src/deepgram/listen/v2/socket_client.py

@@ -7,7 +7,7 @@
 import websockets
 import websockets.sync.connection as websockets_sync_connection
 from ...core.events import EventEmitterMixin, EventType
-from ...core.pydantic_utilities import parse_obj_as
+from ...core.unchecked_base_model import construct_type
 from .types.listen_v2close_stream import ListenV2CloseStream
 from .types.listen_v2connected import ListenV2Connected
 from .types.listen_v2fatal_error import ListenV2FatalError
@@ -31,7 +31,7 @@ async def __aiter__(self):
             if isinstance(message, bytes):
                 yield message
             else:
-                yield parse_obj_as(V2SocketClientResponse, json.loads(message))  # type: ignore
+                yield construct_type(type_=V2SocketClientResponse, object_=json.loads(message))  # type: ignore
 
     async def start_listening(self):
         """
@@ -50,7 +50,7 @@ async def start_listening(self):
                     parsed = raw_message
                 else:
                     json_data = json.loads(raw_message)
-                    parsed = parse_obj_as(V2SocketClientResponse, json_data)  # type: ignore
+                    parsed = construct_type(type_=V2SocketClientResponse, object_=json_data)  # type: ignore
                 await self._emit_async(EventType.MESSAGE, parsed)
         except Exception as exc:
             await self._emit_async(EventType.ERROR, exc)
@@ -79,7 +79,7 @@ async def recv(self) -> V2SocketClientResponse:
         if isinstance(data, bytes):
             return data  # type: ignore
         json_data = json.loads(data)
-        return parse_obj_as(V2SocketClientResponse, json_data)  # type: ignore
+        return construct_type(type_=V2SocketClientResponse, object_=json_data)  # type: ignore
 
     async def _send(self, data: typing.Any) -> None:
         """
@@ -106,7 +106,7 @@ def __iter__(self):
             if isinstance(message, bytes):
                 yield message
             else:
-                yield parse_obj_as(V2SocketClientResponse, json.loads(message))  # type: ignore
+                yield construct_type(type_=V2SocketClientResponse, object_=json.loads(message))  # type: ignore
 
     def start_listening(self):
         """
@@ -125,7 +125,7 @@ def start_listening(self):
                     parsed = raw_message
                 else:
                     json_data = json.loads(raw_message)
-                    parsed = parse_obj_as(V2SocketClientResponse, json_data)  # type: ignore
+                    parsed = construct_type(type_=V2SocketClientResponse, object_=json_data)  # type: ignore
                 self._emit(EventType.MESSAGE, parsed)
         except Exception as exc:
             self._emit(EventType.ERROR, exc)
@@ -154,7 +154,7 @@ def recv(self) -> V2SocketClientResponse:
         if isinstance(data, bytes):
             return data  # type: ignore
         json_data = json.loads(data)
-        return parse_obj_as(V2SocketClientResponse, json_data)  # type: ignore
+        return construct_type(type_=V2SocketClientResponse, object_=json_data)  # type: ignore
 
     def _send(self, data: typing.Any) -> None:
         """

src/deepgram/speak/v1/socket_client.py

@@ -7,7 +7,7 @@
 import websockets
 import websockets.sync.connection as websockets_sync_connection
 from ...core.events import EventEmitterMixin, EventType
-from ...core.pydantic_utilities import parse_obj_as
+from ...core.unchecked_base_model import construct_type
 from .types.speak_v1clear import SpeakV1Clear
 from .types.speak_v1cleared import SpeakV1Cleared
 from .types.speak_v1close import SpeakV1Close
@@ -35,7 +35,7 @@ async def __aiter__(self):
             if isinstance(message, bytes):
                 yield message
             else:
-                yield parse_obj_as(V1SocketClientResponse, json.loads(message))  # type: ignore
+                yield construct_type(type_=V1SocketClientResponse, object_=json.loads(message))  # type: ignore
 
     async def start_listening(self):
         """
@@ -54,7 +54,7 @@ async def start_listening(self):
                     parsed = raw_message
                 else:
                     json_data = json.loads(raw_message)
-                    parsed = parse_obj_as(V1SocketClientResponse, json_data)  # type: ignore
+                    parsed = construct_type(type_=V1SocketClientResponse, object_=json_data)  # type: ignore
                 await self._emit_async(EventType.MESSAGE, parsed)
         except Exception as exc:
             await self._emit_async(EventType.ERROR, exc)
@@ -97,7 +97,7 @@ async def recv(self) -> V1SocketClientResponse:
         if isinstance(data, bytes):
             return data  # type: ignore
         json_data = json.loads(data)
-        return parse_obj_as(V1SocketClientResponse, json_data)  # type: ignore
+        return construct_type(type_=V1SocketClientResponse, object_=json_data)  # type: ignore
 
     async def _send(self, data: typing.Any) -> None:
         """
@@ -124,7 +124,7 @@ def __iter__(self):
             if isinstance(message, bytes):
                 yield message
             else:
-                yield parse_obj_as(V1SocketClientResponse, json.loads(message))  # type: ignore
+                yield construct_type(type_=V1SocketClientResponse, object_=json.loads(message))  # type: ignore
 
     def start_listening(self):
         """
@@ -143,7 +143,7 @@ def start_listening(self):
                     parsed = raw_message
                 else:
                     json_data = json.loads(raw_message)
-                    parsed = parse_obj_as(V1SocketClientResponse, json_data)  # type: ignore
+                    parsed = construct_type(type_=V1SocketClientResponse, object_=json_data)  # type: ignore
                 self._emit(EventType.MESSAGE, parsed)
         except Exception as exc:
             self._emit(EventType.ERROR, exc)
@@ -186,7 +186,7 @@ def recv(self) -> V1SocketClientResponse:
         if isinstance(data, bytes):
             return data  # type: ignore
         json_data = json.loads(data)
-        return parse_obj_as(V1SocketClientResponse, json_data)  # type: ignore
+        return construct_type(type_=V1SocketClientResponse, object_=json_data)  # type: ignore
 
     def _send(self, data: typing.Any) -> None:
         """