fix(voice): honor client config for streamed STT

src/agents/models/_openai_websocket.py

@@ -0,0 +1,115 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+import httpx2
+from openai import AsyncOpenAI, NotGiven, Omit
+
+from .._httpx_compat import is_legacy_httpx_instance
+from ..exceptions import UserError
+
+
+def _is_openai_omitted_value(value: Any) -> bool:
+    return isinstance(value, Omit | NotGiven)
+
+
+async def refresh_openai_client_api_key_if_supported(client: Any) -> None:
+    """Refresh dynamic OpenAI client credentials before materializing handshake headers."""
+    refresh_api_key = getattr(client, "_refresh_api_key", None)
+    if callable(refresh_api_key):
+        await refresh_api_key()
+
+
+def _remove_header(headers: dict[str, str], key: object) -> None:
+    header_key = str(key)
+    for existing_key in list(headers):
+        if existing_key.lower() == header_key.lower():
+            del headers[existing_key]
+
+
+def _set_header(headers: dict[str, str], key: object, value: object) -> None:
+    header_key = str(key)
+    _remove_header(headers, header_key)
+    headers[header_key] = str(value)
+
+
+def merge_openai_client_websocket_headers(
+    client: AsyncOpenAI,
+    *,
+    extra_headers: Mapping[str, Any] | None = None,
+) -> dict[str, str]:
+    """Materialize OpenAI client auth/default headers for a WebSocket handshake."""
+    headers: dict[str, str] = {}
+    for source in (
+        getattr(client, "auth_headers", {}),
+        getattr(client, "default_headers", {}),
+    ):
+        for key, value in source.items():
+            if isinstance(value, NotGiven):
+                continue
+            if isinstance(value, Omit):
+                _remove_header(headers, key)
+                continue
+            _set_header(headers, key, value)
+
+    for key, value in (extra_headers or {}).items():
+        if isinstance(value, NotGiven):
+            continue
+        _remove_header(headers, key)
+        if isinstance(value, Omit):
+            continue
+        headers[str(key)] = str(value)
+
+    return headers
+
+
+def _merge_query_values(params: dict[str, Any], values: Mapping[str, Any]) -> None:
+    for key, value in values.items():
+        query_key = str(key)
+        if isinstance(value, Omit):
+            params.pop(query_key, None)
+            continue
+        if isinstance(value, NotGiven):
+            continue
+        params[query_key] = value
+
+
+def prepare_openai_client_websocket_base_url(
+    client: AsyncOpenAI,
+    *,
+    extra_query: Any = None,
+    context: str,
+) -> httpx2.URL:
+    """Build the client-derived WebSocket base URL and normalized query parameters.
+
+    Endpoint suffixes and transport-specific fixed query parameters are intentionally left to
+    each caller.
+    """
+    websocket_base_url = getattr(client, "websocket_base_url", None)
+    if websocket_base_url is not None:
+        if is_legacy_httpx_instance(websocket_base_url, "URL"):
+            websocket_base_url = str(websocket_base_url)
+        base_url = httpx2.URL(websocket_base_url)
+    else:
+        client_base_url = client.base_url
+        if is_legacy_httpx_instance(client_base_url, "URL"):
+            client_base_url = str(client_base_url)
+        base_url = httpx2.URL(client_base_url)
+
+    ws_scheme = {"http": "ws", "https": "wss"}.get(base_url.scheme, base_url.scheme)
+    base_url = base_url.copy_with(scheme=ws_scheme)
+    params: dict[str, Any] = dict(base_url.params)
+
+    default_query = getattr(client, "default_query", None)
+    if default_query is not None and not _is_openai_omitted_value(default_query):
+        if not isinstance(default_query, Mapping):
+            raise UserError(f"{context} client default_query must be a mapping.")
+        _merge_query_values(params, default_query)
+
+    if extra_query is not None and not _is_openai_omitted_value(extra_query):
+        if not isinstance(extra_query, Mapping):
+            raise UserError(f"{context} extra_query must be a mapping.")
+        _merge_query_values(params, extra_query)
+
+    return base_url.copy_with(params=params)

src/agents/models/openai_responses.py

@@ -91,6 +91,11 @@
 from ..util._json import _to_dump_compatible
 from ..version import __version__
 from ._openai_retry import get_openai_retry_advice
+from ._openai_websocket import (
+    merge_openai_client_websocket_headers,
+    prepare_openai_client_websocket_base_url,
+    refresh_openai_client_api_key_if_supported,
+)
 from ._response_terminal import response_error_event_failure_error, response_terminal_failure_error
 from ._retry_runtime import (
     should_disable_provider_managed_retries,
@@ -177,10 +182,8 @@ def _materialize_responses_tool_params(
 
 
 async def _refresh_openai_client_api_key_if_supported(client: Any) -> None:
-    """Refresh client auth if the current OpenAI SDK exposes a refresh hook."""
-    refresh_api_key = getattr(client, "_refresh_api_key", None)
-    if callable(refresh_api_key):
-        await refresh_api_key()
+    """Backward-compatible wrapper around shared WebSocket client credential refresh."""
+    await refresh_openai_client_api_key_if_supported(client)
 
 
 def _construct_response_stream_event_from_payload(
@@ -1533,76 +1536,19 @@ async def _prepare_websocket_request(
         return frame, ws_url, handshake_headers
 
     def _merge_websocket_headers(self, extra_headers: Mapping[str, Any]) -> dict[str, str]:
-        headers: dict[str, str] = {}
-        for source in (
-            getattr(self._client, "auth_headers", {}),
-            self._client.default_headers,
-        ):
-            for key, value in source.items():
-                if _is_openai_omitted_value(value):
-                    continue
-                header_key = str(key)
-                for existing_key in list(headers):
-                    if existing_key.lower() == header_key.lower():
-                        del headers[existing_key]
-                headers[header_key] = str(value)
-
-        for key, value in extra_headers.items():
-            if isinstance(value, NotGiven):
-                continue
-            header_key = str(key)
-            for existing_key in list(headers):
-                if existing_key.lower() == header_key.lower():
-                    del headers[existing_key]
-            if isinstance(value, Omit):
-                continue
-            headers[header_key] = str(value)
-
-        return headers
+        return merge_openai_client_websocket_headers(
+            self._client,
+            extra_headers=extra_headers,
+        )
 
     def _prepare_websocket_url(self, extra_query: Any) -> str:
-        if self._client.websocket_base_url is not None:
-            websocket_base_url = self._client.websocket_base_url
-            if is_legacy_httpx_instance(websocket_base_url, "URL"):
-                websocket_base_url = str(websocket_base_url)
-            base_url = httpx2.URL(websocket_base_url)
-            ws_scheme = {"http": "ws", "https": "wss"}.get(base_url.scheme, base_url.scheme)
-            base_url = base_url.copy_with(scheme=ws_scheme)
-        else:
-            client_base_url = self._client.base_url
-            ws_scheme = {"http": "ws", "https": "wss"}.get(
-                client_base_url.scheme, client_base_url.scheme
-            )
-            base_url = client_base_url.copy_with(scheme=ws_scheme)
-
-        params: dict[str, Any] = dict(base_url.params)
-        default_query = getattr(self._client, "default_query", None)
-        if default_query is not None and not _is_openai_omitted_value(default_query):
-            if not isinstance(default_query, Mapping):
-                raise UserError("Responses websocket client default_query must be a mapping.")
-            for key, value in default_query.items():
-                query_key = str(key)
-                if isinstance(value, Omit):
-                    params.pop(query_key, None)
-                    continue
-                if isinstance(value, NotGiven):
-                    continue
-                params[query_key] = value
-
-        if extra_query is not None and not _is_openai_omitted_value(extra_query):
-            if not isinstance(extra_query, Mapping):
-                raise UserError("Responses websocket extra_query must be a mapping.")
-            for key, value in extra_query.items():
-                query_key = str(key)
-                if isinstance(value, Omit):
-                    params.pop(query_key, None)
-                    continue
-                if isinstance(value, NotGiven):
-                    continue
-                params[query_key] = value
-
+        base_url = prepare_openai_client_websocket_base_url(
+            self._client,
+            extra_query=extra_query,
+            context="Responses websocket",
+        )
         path = base_url.path.rstrip("/") + "/responses"
-        return str(base_url.copy_with(path=path, params=params))
+        return str(base_url.copy_with(path=path))
 
     async def _ensure_websocket_connection(
         self,

src/agents/voice/models/openai_stt.py

@@ -13,6 +13,11 @@
 from ... import _debug
 from ...exceptions import AgentsException, UserError
 from ...logger import logger
+from ...models._openai_websocket import (
+    merge_openai_client_websocket_headers,
+    prepare_openai_client_websocket_base_url,
+    refresh_openai_client_api_key_if_supported,
+)
 from ...tracing import Span, SpanError, TranscriptionSpanData, transcription_span
 from ...util._error_tracing import get_trace_error
 from ..exceptions import STTWebsocketConnectionError
@@ -58,6 +63,24 @@ def _audio_buffer_to_base64(buffer: npt.NDArray[np.int16 | np.float32]) -> str:
     return base64.b64encode(buffer.tobytes()).decode("utf-8")
 
 
+def _prepare_websocket_url(client: AsyncOpenAI) -> str:
+    base_url = prepare_openai_client_websocket_base_url(
+        client,
+        context="Streamed STT websocket",
+    )
+    params: dict[str, Any] = dict(base_url.params)
+    params["intent"] = "transcription"
+    path = base_url.path.rstrip("/") + "/realtime"
+    return str(base_url.copy_with(path=path, params=params))
+
+
+def _prepare_websocket_headers(client: AsyncOpenAI) -> dict[str, str]:
+    return merge_openai_client_websocket_headers(
+        client,
+        extra_headers={"OpenAI-Log-Session": "1"},
+    )
+
+
 async def _wait_for_event(
     event_queue: asyncio.Queue[dict[str, Any] | ErrorSentinel],
     expected_types: list[str],
@@ -303,12 +326,10 @@ async def _stream_audio(
 
     async def _process_websocket_connection(self) -> None:
         try:
+            await refresh_openai_client_api_key_if_supported(self._client)
             async with websockets.connect(
-                "wss://api.openai.com/v1/realtime?intent=transcription",
-                additional_headers={
-                    "Authorization": f"Bearer {self._client.api_key}",
-                    "OpenAI-Log-Session": "1",
-                },
+                _prepare_websocket_url(self._client),
+                additional_headers=_prepare_websocket_headers(self._client),
             ) as ws:
                 await self._setup_connection(ws)
                 self._process_events_task = asyncio.create_task(self._handle_events())