From abb89eb23c6399b2396cf0457b36698f8dc12ca7 Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Tue, 27 Oct 2026 00:17:01 +0000 Subject: [PATCH 1/5] fix(examples): detect incomplete WebSocket responses --- README.md | 8 +++++++- examples/live/audio_transcript.py | 1 + examples/responses/multi_agent_websocket.py | 6 +++++- 3 files changed, 13 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 1fabef6a81..5f2c066cdb 100644 --- a/README.md +++ b/README.md @@ -396,7 +396,7 @@ Basic text based example: ```py import asyncio -from openai import AsyncOpenAI +from openai import AsyncOpenAI, OpenAIError async def main(): @@ -422,7 +422,13 @@ async def main(): print() elif event.type == "response.done": + if event.response.status != "completed": + raise OpenAIError("The Realtime response did not complete") break + elif event.type == "error": + raise OpenAIError("Realtime returned an error; inspect the event in your application") + else: + raise OpenAIError("Realtime disconnected before response.done") asyncio.run(main()) diff --git a/examples/live/audio_transcript.py b/examples/live/audio_transcript.py index ce952ee343..98045a0830 100644 --- a/examples/live/audio_transcript.py +++ b/examples/live/audio_transcript.py @@ -59,6 +59,7 @@ async def receive() -> None: await transcript.push(event) if event.type == "session.closed": return + raise OpenAIError("Live disconnected before session.closed") receiver = asyncio.create_task(receive()) try: diff --git a/examples/responses/multi_agent_websocket.py b/examples/responses/multi_agent_websocket.py index a6390419de..1817be6cf9 100644 --- a/examples/responses/multi_agent_websocket.py +++ b/examples/responses/multi_agent_websocket.py @@ -1,6 +1,6 @@ from __future__ import annotations -from openai import OpenAI +from openai import OpenAI, OpenAIError from openai.types.beta import BetaResponseOutputItem PROMPT = """Compare these fictional proposals on speed, cost, reliability, and security. @@ -42,5 +42,9 @@ def agent_name(item: BetaResponseOutputItem) -> str: print(event.delta, end="", flush=True) if event.type == "response.completed": break + if event.type in {"response.failed", "response.incomplete", "error"}: + raise OpenAIError(f"Multi-agent request ended with {event.type}") + else: + raise OpenAIError("Connection closed before the multi-agent response completed") print() From 24c4e56455b8337d99af943dbd1edc7ffd9ca810 Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Tue, 27 Oct 2026 03:20:39 +0000 Subject: [PATCH 2/5] fix(examples): surface Azure Realtime response failures --- examples/realtime/azure_realtime.py | 106 +++++++++++++++------------- 1 file changed, 55 insertions(+), 51 deletions(-) diff --git a/examples/realtime/azure_realtime.py b/examples/realtime/azure_realtime.py index a8bcf4c536..b0b044559c 100644 --- a/examples/realtime/azure_realtime.py +++ b/examples/realtime/azure_realtime.py @@ -20,60 +20,64 @@ async def main() -> None: Enter "q" to quit the conversation. """ - credential = DefaultAzureCredential() - token_provider = get_bearer_token_provider(credential, "https://cognitiveservices.azure.com/.default") - token = await token_provider() - - # The endpoint of your Azure OpenAI resource is required. You can set it in the AZURE_OPENAI_ENDPOINT - # environment variable. - # You can find it in the Microsoft Foundry portal in the Overview page of your Azure OpenAI resource. - # Example: https://{your-resource}.openai.azure.com - endpoint = os.environ["AZURE_OPENAI_ENDPOINT"] - - # The deployment name of the model you want to use is required. You can set it in the AZURE_OPENAI_DEPLOYMENT_NAME - # environment variable. - # You can find it in the Foundry portal in the "Models + endpoints" page of your Azure OpenAI resource. - # Example: gpt-realtime - deployment_name = os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"] - - base_url = endpoint.replace("https://", "wss://").rstrip("/") + "/openai/v1" - - # The APIs are compatible with the OpenAI client library. - # You can use the OpenAI client library to access the Azure OpenAI APIs. - # Make sure to set the baseURL and apiKey to use the Azure OpenAI endpoint and token. - client = AsyncOpenAI(websocket_base_url=base_url, api_key=token) - async with client.realtime.connect( - model=deployment_name, - ) as connection: - await connection.session.update( - session={ - "output_modalities": ["text"], - "model": deployment_name, - "type": "realtime", - } - ) - while True: - user_input = input("Enter a message: ") - if user_input == "q": - break - - await connection.conversation.item.create( - item={ - "type": "message", - "role": "user", - "content": [{"type": "input_text", "text": user_input}], - } + async with DefaultAzureCredential() as credential: + token_provider = get_bearer_token_provider(credential, "https://cognitiveservices.azure.com/.default") + token = await token_provider() + + # The endpoint of your Azure OpenAI resource is required. You can set it in the AZURE_OPENAI_ENDPOINT + # environment variable. + # You can find it in the Microsoft Foundry portal in the Overview page of your Azure OpenAI resource. + # Example: https://{your-resource}.openai.azure.com + endpoint = os.environ["AZURE_OPENAI_ENDPOINT"] + + # The deployment name of the model you want to use is required. You can set it in the AZURE_OPENAI_DEPLOYMENT_NAME + # environment variable. + # You can find it in the Foundry portal in the "Models + endpoints" page of your Azure OpenAI resource. + # Example: gpt-realtime + deployment_name = os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"] + + base_url = endpoint.replace("https://", "wss://").rstrip("/") + "/openai/v1" + + # The APIs are compatible with the OpenAI client library. + # You can use the OpenAI client library to access the Azure OpenAI APIs. + # Make sure to set the baseURL and apiKey to use the Azure OpenAI endpoint and token. + async with ( + AsyncOpenAI(websocket_base_url=base_url, api_key=token) as client, + client.realtime.connect(model=deployment_name) as connection, + ): + await connection.session.update( + session={ + "output_modalities": ["text"], + "model": deployment_name, + "type": "realtime", + }, ) - await connection.response.create() - async for event in connection: - if event.type == "response.output_text.delta": - print(event.delta, flush=True, end="") - elif event.type == "response.output_text.done": - print() - elif event.type == "response.done": + while True: + user_input = input("Enter a message: ") + if user_input == "q": break - await credential.close() + await connection.conversation.item.create( + item={ + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": user_input}], + } + ) + await connection.response.create() + async for event in connection: + if event.type == "response.output_text.delta": + print(event.delta, flush=True, end="") + elif event.type == "response.output_text.done": + print() + elif event.type == "error": + raise RuntimeError(f"Realtime error: {event.error.type} ({event.error.code})") + elif event.type == "response.done": + if event.response.status != "completed": + raise RuntimeError(f"Response did not complete successfully: {event.response.status}") + break + else: + raise RuntimeError("WebSocket closed before the response completed.") asyncio.run(main()) From e9fe7c31815f2d83c99f8c37d27aeacd2ab0da02 Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 02:54:28 +0000 Subject: [PATCH 3/5] Handle child response results and Live disconnects during transcript flush --- examples/live/audio_transcript.py | 3 ++- examples/responses/multi_agent_websocket.py | 4 ++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/examples/live/audio_transcript.py b/examples/live/audio_transcript.py index 98045a0830..f79ccb5fd8 100644 --- a/examples/live/audio_transcript.py +++ b/examples/live/audio_transcript.py @@ -73,9 +73,10 @@ async def receive() -> None: chunk = samples[offset : offset + chunk_bytes] await connection.session.input_audio.append(audio=base64.b64encode(chunk).decode("ascii")) await asyncio.sleep(len(chunk) / (24000 * 2)) + await transcript.close() + # Flushing the transcript may yield while the peer closes. if receiver.done(): await receiver - await transcript.close() finally: receiver.cancel() await asyncio.gather(receiver, return_exceptions=True) diff --git a/examples/responses/multi_agent_websocket.py b/examples/responses/multi_agent_websocket.py index 1817be6cf9..5e5002fca2 100644 --- a/examples/responses/multi_agent_websocket.py +++ b/examples/responses/multi_agent_websocket.py @@ -40,9 +40,9 @@ def agent_name(item: BetaResponseOutputItem) -> str: print(f"{separator}━━━ {role}: {name} ━━━\n") labeled_items.add(event.output_index) print(event.delta, end="", flush=True) - if event.type == "response.completed": + if event.type == "response.completed" and event.agent is None: break - if event.type in {"response.failed", "response.incomplete", "error"}: + if event.type in {"response.failed", "response.incomplete", "error"} and event.agent is None: raise OpenAIError(f"Multi-agent request ended with {event.type}") else: raise OpenAIError("Connection closed before the multi-agent response completed") From f311981f9fb894440d9b8266dc8d499c37bc4a21 Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 02:59:12 +0000 Subject: [PATCH 4/5] Narrow multi-agent terminal event types before accessing agent --- examples/responses/multi_agent_websocket.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/examples/responses/multi_agent_websocket.py b/examples/responses/multi_agent_websocket.py index 5e5002fca2..cb7d990fbd 100644 --- a/examples/responses/multi_agent_websocket.py +++ b/examples/responses/multi_agent_websocket.py @@ -42,7 +42,9 @@ def agent_name(item: BetaResponseOutputItem) -> str: print(event.delta, end="", flush=True) if event.type == "response.completed" and event.agent is None: break - if event.type in {"response.failed", "response.incomplete", "error"} and event.agent is None: + if ( + event.type == "response.failed" or event.type == "response.incomplete" or event.type == "error" + ) and event.agent is None: raise OpenAIError(f"Multi-agent request ended with {event.type}") else: raise OpenAIError("Connection closed before the multi-agent response completed") From 4474ba0fa84e1d82b46415d200362343c350bd4e Mon Sep 17 00:00:00 2001 From: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com> Date: Wed, 28 Oct 2026 04:36:29 +0000 Subject: [PATCH 5/5] fix: confirm Live audio transcript session finalization --- examples/live/audio_transcript.py | 15 ++- .../lib/live/test_audio_transcript_example.py | 104 ++++++++++++++++++ 2 files changed, 116 insertions(+), 3 deletions(-) create mode 100644 tests/lib/live/test_audio_transcript_example.py diff --git a/examples/live/audio_transcript.py b/examples/live/audio_transcript.py index f79ccb5fd8..d3d9ad054d 100644 --- a/examples/live/audio_transcript.py +++ b/examples/live/audio_transcript.py @@ -4,7 +4,9 @@ python examples/live/audio_transcript.py speech.wav Use a 16-bit mono WAV recorded at 24 kHz. Audio is paced in real time. This -example sends silence after the recording so the assistant can finish replying. +example sends silence for the listen window after the recording, then requests +session closure and waits for confirmation. The window does not guarantee that +the assistant has finished replying. """ from __future__ import annotations @@ -49,15 +51,21 @@ def render(segment: TranscriptSegment) -> None: event = await asyncio.wait_for(connection.recv(), timeout=30) if event.type == "error": raise OpenAIError("Live rejected session startup; inspect your model access and configuration") + if event.type == "session.closed": + raise OpenAIError("Live session ended before startup") if event.type == "session.started": break + closing = False + async def receive() -> None: async for event in connection: if event.type == "error": raise OpenAIError("Live returned an error while processing the audio") await transcript.push(event) if event.type == "session.closed": + if not closing or event.reason != "close_requested": + raise OpenAIError("Live session ended without confirming the requested close") return raise OpenAIError("Live disconnected before session.closed") @@ -73,10 +81,11 @@ async def receive() -> None: chunk = samples[offset : offset + chunk_bytes] await connection.session.input_audio.append(audio=base64.b64encode(chunk).decode("ascii")) await asyncio.sleep(len(chunk) / (24000 * 2)) - await transcript.close() - # Flushing the transcript may yield while the peer closes. if receiver.done(): await receiver + closing = True + await connection.session.close() + await asyncio.wait_for(receiver, timeout=30) finally: receiver.cancel() await asyncio.gather(receiver, return_exceptions=True) diff --git a/tests/lib/live/test_audio_transcript_example.py b/tests/lib/live/test_audio_transcript_example.py new file mode 100644 index 0000000000..e5846a93a4 --- /dev/null +++ b/tests/lib/live/test_audio_transcript_example.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import json +import wave +import asyncio +from pathlib import Path + +import pytest +from websockets.exceptions import ConnectionClosedOK +from websockets.sync.server import ServerConnection + +from openai import OpenAIError +from examples.live.audio_transcript import run + +from ..responses.test_websocket_session import script_server + + +@pytest.mark.parametrize("stage", ["startup", "audio", "requested"]) +@pytest.mark.parametrize( + "reason", ["close_requested", "expired", "content", "remote_hangup", "connection_lost", "error", "eof"] +) +async def test_audio_example_requires_its_requested_finalization( + stage: str, reason: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + wav_path = tmp_path / "fixture.wav" + with wave.open(str(wav_path), "wb") as wav: + wav.setparams((1, 2, 24000, 0, "NONE", "not compressed")) + wav.writeframes(bytes(24000 * 2 // 10)) + + sent_closed = False + received_close = False + audio_messages = 0 + + def script(socket: ServerConnection) -> None: + nonlocal sent_closed, received_close, audio_messages + + def send_closed() -> None: + nonlocal sent_closed + sent_closed = True + if reason == "eof": + socket.close() + elif reason == "error": + socket.send('{"type": "error", "event_id": "failure", "message": "Synthetic test error"}') + else: + socket.send(json.dumps({"type": "session.closed", "event_id": "done", "reason": reason})) + + assert json.loads(socket.recv(timeout=3))["type"] == "session.start" + if stage == "startup": + send_closed() + else: + socket.send('{"type": "session.started", "event_id": "started", "session": {"id": "fixture"}}') + # Leave the connection open after the terminal event. The client must + # reject that event immediately, rather than waiting for transport EOF. + try: + while True: + event = json.loads(socket.recv(timeout=3)) + if event["type"] == "session.input_audio.append": + audio_messages += 1 + if stage == "audio" and not sent_closed: + send_closed() + elif event["type"] == "session.close": + received_close = True + if stage == "requested": + # Final transcript data after close was requested must + # reach the caller before the confirmed result returns. + socket.send( + json.dumps( + { + "type": "session.output_transcript.delta", + "event_id": "text", + "delta": "Final fixture transcript.", + "start_ms": 0, + "end_ms": 100, + } + ) + ) + send_closed() + except ConnectionClosedOK: + return + + monkeypatch.setenv("OPENAI_API_KEY", "fake-transcript-fixture-key") + monkeypatch.setenv("NO_PROXY", "127.0.0.1") + monkeypatch.setenv("no_proxy", "127.0.0.1") + # The sample constructs its own client; restrict this fixture to loopback. + for name in ("ALL_PROXY", "all_proxy", "HTTP_PROXY", "http_proxy", "HTTPS_PROXY", "https_proxy"): + monkeypatch.delenv(name, raising=False) + with script_server(script) as url: + monkeypatch.setenv("OPENAI_BASE_URL", url) + if stage == "requested" and reason == "close_requested": + result = await asyncio.wait_for(run(wav_path, model="fixture-model", listen_seconds=0), timeout=1) + assert any(segment.text == "Final fixture transcript." for segment in result) + else: + # recv() can report transport EOF directly while awaiting startup; + # all terminal paths must fail instead of returning partial text. + with pytest.raises((OpenAIError, ConnectionClosedOK)): + await asyncio.wait_for(run(wav_path, model="fixture-model", listen_seconds=0), timeout=1) + assert sent_closed + assert received_close == (stage == "requested") + if stage == "startup": + assert audio_messages == 0 + elif stage == "requested": + assert audio_messages == 5 + else: + assert 1 <= audio_messages <= 5