deepgram · GregHolmes · Jun 1, 2026 · May 26, 2026
@@ -1,6 +1,11 @@
 # Custom client implementation extending BaseClient with additional features:
 # - access_token parameter support (Bearer token authentication)
 # - Automatic session ID generation and header injection (x-deepgram-session-id)
+# - transport_factory parameter for plugging in a custom WebSocket transport
+# - reconnect flag (default True) — declarative flag indicating whether the SDK
+#   is expected to manage WebSocket reconnects. Auto-disabled when a custom
+#   transport_factory is provided so transports that own their retry lifecycle
+#   (e.g. SageMaker) aren't double-stacked with future SDK-side reconnect logic.
 # This file is manually maintained and should not be regenerated
 src/deepgram/client.py
 

@@ -22,7 +22,7 @@ How to identify:
 - The file lives **outside `src/deepgram/`** in a hand-maintained location (e.g., `.claude/`, `docs/`)
 
 Current permanently frozen files:
-- `src/deepgram/client.py` — entirely custom (Bearer auth, session ID); no Fern equivalent
+- `src/deepgram/client.py` — entirely custom (Bearer auth, session ID, `transport_factory`, `reconnect` parity flag); no Fern equivalent
 - `src/deepgram/helpers/` — hand-written TextBuilder helpers
 - `src/deepgram/agent/v1/types/agent_v1history_content.py`, `src/deepgram/agent/v1/types/agent_v1history_function_calls.py`, `src/deepgram/agent/v1/types/agent_v1settings_agent_context_messages_item.py`, `src/deepgram/agent/v1/types/agent_v1settings_agent_context_messages_item_content.py`, `src/deepgram/agent/v1/types/agent_v1settings_agent_context_messages_item_content_role.py`, `src/deepgram/agent/v1/types/agent_v1settings_agent_context_messages_item_function_calls.py`, `src/deepgram/agent/v1/types/agent_v1settings_agent_context_messages_item_function_calls_function_calls_item.py` — hand-written compatibility aliases preserving old public Agent History type imports after regen renames
 - `src/deepgram/agent/v1/requests/agent_v1history_content.py`, `src/deepgram/agent/v1/requests/agent_v1history_function_calls.py`, `src/deepgram/agent/v1/requests/agent_v1settings_agent_context_messages_item.py`, `src/deepgram/agent/v1/requests/agent_v1settings_agent_context_messages_item_content.py`, `src/deepgram/agent/v1/requests/agent_v1settings_agent_context_messages_item_function_calls.py`, `src/deepgram/agent/v1/requests/agent_v1settings_agent_context_messages_item_function_calls_function_calls_item.py` — hand-written compatibility aliases preserving old public Agent History request-param imports after regen renames

@@ -12,6 +12,14 @@
 - `transport_factory` to replace the default `websockets` transport with a custom one:
   - A callable: ``factory(url, headers) -> transport`` returning an object with
     ``send()``, ``recv()``, iteration, and ``close()`` support.
+- `reconnect` flag (default `True`) declaring whether the SDK is expected to
+  manage WebSocket reconnects for this client. When a custom
+  ``transport_factory`` is set, ``reconnect`` is auto-disabled because the
+  custom transport owns its own retry/reconnect lifecycle; double-stacking
+  retries on top would cause storm-on-storm under burst load. The Python
+  SDK has no wrapper reconnect layer today, so this flag is declarative
+  only -- it documents intent and is reserved for any future SDK-side
+  reconnect logic.
 """
 
 import types
@@ -57,6 +65,12 @@ class DeepgramClient(BaseClient):
     - `transport_factory`: Custom sync WebSocket transport factory. A callable
                            ``factory(url, headers) -> transport`` whose return value must support
                            ``send()``, ``recv()``, iteration, and ``close()``.
+    - `reconnect`: Declarative flag (default ``True``) signalling whether the SDK is
+                    expected to manage WebSocket reconnects for this client. Auto-disabled
+                    when ``transport_factory`` is set, since the custom transport owns its
+                    retry/reconnect lifecycle. The Python SDK has no wrapper reconnect layer
+                    today, so this flag is declarative only -- it documents intent and is
+                    reserved for any future SDK-side reconnect logic.
     - `telemetry_opt_out`: Telemetry opt-out flag (maintained for backwards compatibility, no-op).
     - `telemetry_handler`: Telemetry handler (maintained for backwards compatibility, no-op).
     """
@@ -65,6 +79,7 @@ def __init__(self, *args, **kwargs) -> None:
         access_token: Optional[str] = kwargs.pop("access_token", None)
         session_id: Optional[str] = kwargs.pop("session_id", None)
         transport_factory: Optional[Callable] = kwargs.pop("transport_factory", None)
+        reconnect: bool = bool(kwargs.pop("reconnect", True))
         telemetry_opt_out: bool = bool(kwargs.pop("telemetry_opt_out", True))
         telemetry_handler: Optional[Any] = kwargs.pop("telemetry_handler", None)
 
@@ -95,9 +110,13 @@ def __init__(self, *args, **kwargs) -> None:
         if access_token is not None:
             _apply_bearer_authorization_override(self._client_wrapper, access_token)
 
-        # Install custom WebSocket transport if provided
+        # Install custom WebSocket transport if provided. Auto-disable
+        # `reconnect`: a custom transport owns its retry lifecycle, so flip
+        # the flag off even if the caller left it at the default.
         if transport_factory is not None:
             install_transport(sync_factory=transport_factory)
+            reconnect = False
+        self.reconnect = reconnect
 
         # Store telemetry handler for backwards compatibility (no-op, telemetry not implemented)
         self._telemetry_handler = None
@@ -114,6 +133,12 @@ class AsyncDeepgramClient(AsyncBaseClient):
     - `transport_factory`: Custom async WebSocket transport factory. A callable
                            ``factory(url, headers) -> transport`` whose return value must support
                            ``send()``, ``recv()``, async iteration, and ``close()``.
+    - `reconnect`: Declarative flag (default ``True``) signalling whether the SDK is
+                    expected to manage WebSocket reconnects for this client. Auto-disabled
+                    when ``transport_factory`` is set, since the custom transport owns its
+                    retry/reconnect lifecycle. The Python SDK has no wrapper reconnect layer
+                    today, so this flag is declarative only -- it documents intent and is
+                    reserved for any future SDK-side reconnect logic.
     - `telemetry_opt_out`: Telemetry opt-out flag (maintained for backwards compatibility, no-op).
     - `telemetry_handler`: Telemetry handler (maintained for backwards compatibility, no-op).
     """
@@ -122,6 +147,7 @@ def __init__(self, *args, **kwargs) -> None:
         access_token: Optional[str] = kwargs.pop("access_token", None)
         session_id: Optional[str] = kwargs.pop("session_id", None)
         transport_factory: Optional[Callable] = kwargs.pop("transport_factory", None)
+        reconnect: bool = bool(kwargs.pop("reconnect", True))
         telemetry_opt_out: bool = bool(kwargs.pop("telemetry_opt_out", True))
         telemetry_handler: Optional[Any] = kwargs.pop("telemetry_handler", None)
 
@@ -152,9 +178,13 @@ def __init__(self, *args, **kwargs) -> None:
         if access_token is not None:
             _apply_bearer_authorization_override(self._client_wrapper, access_token)
 
-        # Install custom WebSocket transport if provided
+        # Install custom WebSocket transport if provided. Auto-disable
+        # `reconnect`: a custom transport owns its retry lifecycle, so flip
+        # the flag off even if the caller left it at the default.
         if transport_factory is not None:
             install_transport(async_factory=transport_factory)
+            reconnect = False
+        self.reconnect = reconnect
 
         # Store telemetry handler for backwards compatibility (no-op, telemetry not implemented)
         self._telemetry_handler = None
@@ -479,3 +479,54 @@ def test_async_deepgram_client_installs_async_transport(self):
             mod = sys.modules[mod_path]
             if hasattr(mod, "websockets_client_connect"):
                 assert isinstance(mod.websockets_client_connect, _AsyncTransportShim)
+
+
+# ---------------------------------------------------------------------------
+# `reconnect` parity flag
+# ---------------------------------------------------------------------------
+
+class TestReconnectFlag:
+    """The `reconnect` flag is declarative in Python (no wrapper layer to disable
+    today), but is auto-disabled when a custom transport is in use so transports
+    that own their retry lifecycle aren't double-stacked with future SDK-side
+    reconnect logic."""
+
+    def test_default_reconnect_is_true(self):
+        from deepgram.client import DeepgramClient
+        client = DeepgramClient(api_key="test-key")
+        assert client.reconnect is True
+
+    def test_explicit_reconnect_false(self):
+        from deepgram.client import DeepgramClient
+        client = DeepgramClient(api_key="test-key", reconnect=False)
+        assert client.reconnect is False
+
+    def test_transport_factory_auto_disables_reconnect(self):
+        _ensure_modules_loaded()
+        factory = MagicMock()
+        from deepgram.client import DeepgramClient
+        client = DeepgramClient(api_key="test-key", transport_factory=factory)
+        assert client.reconnect is False
+
+    def test_transport_factory_overrides_explicit_true(self):
+        """Even when the caller passes reconnect=True, a custom transport_factory
+        wins -- the custom transport owns its retry lifecycle."""
+        _ensure_modules_loaded()
+        factory = MagicMock()
+        from deepgram.client import DeepgramClient
+        client = DeepgramClient(
+            api_key="test-key", transport_factory=factory, reconnect=True
+        )
+        assert client.reconnect is False
+
+    def test_async_default_reconnect_is_true(self):
+        from deepgram.client import AsyncDeepgramClient
+        client = AsyncDeepgramClient(api_key="test-key")
+        assert client.reconnect is True
+
+    def test_async_transport_factory_auto_disables_reconnect(self):
+        _ensure_modules_loaded()
+        factory = MagicMock()
+        from deepgram.client import AsyncDeepgramClient
+        client = AsyncDeepgramClient(api_key="test-key", transport_factory=factory)
+        assert client.reconnect is False