fix(examples): detect incomplete WebSocket responses

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())

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,16 +51,23 @@ 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")
 
             receiver = asyncio.create_task(receive())
             try:
@@ -74,7 +83,9 @@ async def receive() -> None:
                     await asyncio.sleep(len(chunk) / (24000 * 2))
                 if receiver.done():
                     await receiver
-                await transcript.close()
+                closing = True
+                await connection.session.close()
+                await asyncio.wait_for(receiver, timeout=30)
             finally:
                 receiver.cancel()
                 await asyncio.gather(receiver, return_exceptions=True)

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())

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.
@@ -40,7 +40,13 @@ 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 == "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")
 
 print()

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