Skip to content

Commit 8559299

Browse files
markstuart-oaimarkstuart-oai
authored
fix(examples): detect incomplete WebSocket responses (#4187)
## Summary The Realtime README, multi-agent Responses sample and Live transcript sample report errors and early closure. The Azure Realtime CLI also rejects unsuccessful terminal responses. It closes its credential, client and connection on normal and error paths. Only coordinator terminal events end the multi-agent sample; delegated agents may fail or finish while the coordinator continues. The Live reader captures through a bounded listen window, requests session closure, and returns only after `session.closed` confirms `close_requested`. It receives final transcript data through that confirmation. Unexpected closure, errors and EOF before or during finalization fail instead of returning partial text. The listen window does not guarantee a complete model answer. ## Verification - Current Live verification: 223 focused tests pass in both checkouts, including 21 real WebSocket startup/audio/finalization cases for all five close reasons, protocol errors and EOF. The original 15 closure cases fail against the preceding implementation. Scoped Ruff checks and Python 3.10 compilation pass. Hosted Python 3.10, Python 3.14 and HTTPX2 suites passed, along with lint, build, policy, lock and security checks. - Earlier checks on the unchanged multi-agent sample: 20 child/coordinator terminal-event combinations passed; Ruff lint/format and the CI type-narrowing check passed. - 18 local WebSocket success and failure cases passed for the README, multi-agent and Live samples. - Six Azure cases passed against this SDK: completion, protocol error, early close, failed, cancelled and incomplete responses. - Azure tests use the real Identity provider with local token and TLS WebSocket fixtures. No Azure cloud service was contacted. - Pinned Ruff format/lint, Python 3.10 execution and unchanged custom-code budget guards passed. Generated code and generation metadata are unchanged. --------- Co-authored-by: Mark Stuart <323302876+markstuart-oai@users.noreply.github.com>
1 parent fa67521 commit 8559299

5 files changed

Lines changed: 187 additions & 56 deletions

File tree

‎README.md‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -396,7 +396,7 @@ Basic text based example:
396396

397397
```py
398398
import asyncio
399-
from openai import AsyncOpenAI
399+
from openai import AsyncOpenAI, OpenAIError
400400

401401

402402
async def main():
@@ -422,7 +422,13 @@ async def main():
422422
print()
423423

424424
elif event.type == "response.done":
425+
if event.response.status != "completed":
426+
raise OpenAIError("The Realtime response did not complete")
425427
break
428+
elif event.type == "error":
429+
raise OpenAIError("Realtime returned an error; inspect the event in your application")
430+
else:
431+
raise OpenAIError("Realtime disconnected before response.done")
426432

427433

428434
asyncio.run(main())

‎examples/live/audio_transcript.py‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,9 @@
44
python examples/live/audio_transcript.py speech.wav
55
66
Use a 16-bit mono WAV recorded at 24 kHz. Audio is paced in real time. This
7-
example sends silence after the recording so the assistant can finish replying.
7+
example sends silence for the listen window after the recording, then requests
8+
session closure and waits for confirmation. The window does not guarantee that
9+
the assistant has finished replying.
810
"""
911

1012
from __future__ import annotations
@@ -49,16 +51,23 @@ def render(segment: TranscriptSegment) -> None:
4951
event = await asyncio.wait_for(connection.recv(), timeout=30)
5052
if event.type == "error":
5153
raise OpenAIError("Live rejected session startup; inspect your model access and configuration")
54+
if event.type == "session.closed":
55+
raise OpenAIError("Live session ended before startup")
5256
if event.type == "session.started":
5357
break
5458

59+
closing = False
60+
5561
async def receive() -> None:
5662
async for event in connection:
5763
if event.type == "error":
5864
raise OpenAIError("Live returned an error while processing the audio")
5965
await transcript.push(event)
6066
if event.type == "session.closed":
67+
if not closing or event.reason != "close_requested":
68+
raise OpenAIError("Live session ended without confirming the requested close")
6169
return
70+
raise OpenAIError("Live disconnected before session.closed")
6271

6372
receiver = asyncio.create_task(receive())
6473
try:
@@ -74,7 +83,9 @@ async def receive() -> None:
7483
await asyncio.sleep(len(chunk) / (24000 * 2))
7584
if receiver.done():
7685
await receiver
77-
await transcript.close()
86+
closing = True
87+
await connection.session.close()
88+
await asyncio.wait_for(receiver, timeout=30)
7889
finally:
7990
receiver.cancel()
8091
await asyncio.gather(receiver, return_exceptions=True)

‎examples/realtime/azure_realtime.py‎

Lines changed: 55 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -20,60 +20,64 @@ async def main() -> None:
2020
Enter "q" to quit the conversation.
2121
"""
2222

23-
credential = DefaultAzureCredential()
24-
token_provider = get_bearer_token_provider(credential, "https://cognitiveservices.azure.com/.default")
25-
token = await token_provider()
26-
27-
# The endpoint of your Azure OpenAI resource is required. You can set it in the AZURE_OPENAI_ENDPOINT
28-
# environment variable.
29-
# You can find it in the Microsoft Foundry portal in the Overview page of your Azure OpenAI resource.
30-
# Example: https://{your-resource}.openai.azure.com
31-
endpoint = os.environ["AZURE_OPENAI_ENDPOINT"]
32-
33-
# The deployment name of the model you want to use is required. You can set it in the AZURE_OPENAI_DEPLOYMENT_NAME
34-
# environment variable.
35-
# You can find it in the Foundry portal in the "Models + endpoints" page of your Azure OpenAI resource.
36-
# Example: gpt-realtime
37-
deployment_name = os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"]
38-
39-
base_url = endpoint.replace("https://", "wss://").rstrip("/") + "/openai/v1"
40-
41-
# The APIs are compatible with the OpenAI client library.
42-
# You can use the OpenAI client library to access the Azure OpenAI APIs.
43-
# Make sure to set the baseURL and apiKey to use the Azure OpenAI endpoint and token.
44-
client = AsyncOpenAI(websocket_base_url=base_url, api_key=token)
45-
async with client.realtime.connect(
46-
model=deployment_name,
47-
) as connection:
48-
await connection.session.update(
49-
session={
50-
"output_modalities": ["text"],
51-
"model": deployment_name,
52-
"type": "realtime",
53-
}
54-
)
55-
while True:
56-
user_input = input("Enter a message: ")
57-
if user_input == "q":
58-
break
59-
60-
await connection.conversation.item.create(
61-
item={
62-
"type": "message",
63-
"role": "user",
64-
"content": [{"type": "input_text", "text": user_input}],
65-
}
23+
async with DefaultAzureCredential() as credential:
24+
token_provider = get_bearer_token_provider(credential, "https://cognitiveservices.azure.com/.default")
25+
token = await token_provider()
26+
27+
# The endpoint of your Azure OpenAI resource is required. You can set it in the AZURE_OPENAI_ENDPOINT
28+
# environment variable.
29+
# You can find it in the Microsoft Foundry portal in the Overview page of your Azure OpenAI resource.
30+
# Example: https://{your-resource}.openai.azure.com
31+
endpoint = os.environ["AZURE_OPENAI_ENDPOINT"]
32+
33+
# The deployment name of the model you want to use is required. You can set it in the AZURE_OPENAI_DEPLOYMENT_NAME
34+
# environment variable.
35+
# You can find it in the Foundry portal in the "Models + endpoints" page of your Azure OpenAI resource.
36+
# Example: gpt-realtime
37+
deployment_name = os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"]
38+
39+
base_url = endpoint.replace("https://", "wss://").rstrip("/") + "/openai/v1"
40+
41+
# The APIs are compatible with the OpenAI client library.
42+
# You can use the OpenAI client library to access the Azure OpenAI APIs.
43+
# Make sure to set the baseURL and apiKey to use the Azure OpenAI endpoint and token.
44+
async with (
45+
AsyncOpenAI(websocket_base_url=base_url, api_key=token) as client,
46+
client.realtime.connect(model=deployment_name) as connection,
47+
):
48+
await connection.session.update(
49+
session={
50+
"output_modalities": ["text"],
51+
"model": deployment_name,
52+
"type": "realtime",
53+
},
6654
)
67-
await connection.response.create()
68-
async for event in connection:
69-
if event.type == "response.output_text.delta":
70-
print(event.delta, flush=True, end="")
71-
elif event.type == "response.output_text.done":
72-
print()
73-
elif event.type == "response.done":
55+
while True:
56+
user_input = input("Enter a message: ")
57+
if user_input == "q":
7458
break
7559

76-
await credential.close()
60+
await connection.conversation.item.create(
61+
item={
62+
"type": "message",
63+
"role": "user",
64+
"content": [{"type": "input_text", "text": user_input}],
65+
}
66+
)
67+
await connection.response.create()
68+
async for event in connection:
69+
if event.type == "response.output_text.delta":
70+
print(event.delta, flush=True, end="")
71+
elif event.type == "response.output_text.done":
72+
print()
73+
elif event.type == "error":
74+
raise RuntimeError(f"Realtime error: {event.error.type} ({event.error.code})")
75+
elif event.type == "response.done":
76+
if event.response.status != "completed":
77+
raise RuntimeError(f"Response did not complete successfully: {event.response.status}")
78+
break
79+
else:
80+
raise RuntimeError("WebSocket closed before the response completed.")
7781

7882

7983
asyncio.run(main())

‎examples/responses/multi_agent_websocket.py‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
from __future__ import annotations
22

3-
from openai import OpenAI
3+
from openai import OpenAI, OpenAIError
44
from openai.types.beta import BetaResponseOutputItem
55

66
PROMPT = """Compare these fictional proposals on speed, cost, reliability, and security.
@@ -40,7 +40,13 @@ def agent_name(item: BetaResponseOutputItem) -> str:
4040
print(f"{separator}━━━ {role}: {name} ━━━\n")
4141
labeled_items.add(event.output_index)
4242
print(event.delta, end="", flush=True)
43-
if event.type == "response.completed":
43+
if event.type == "response.completed" and event.agent is None:
4444
break
45+
if (
46+
event.type == "response.failed" or event.type == "response.incomplete" or event.type == "error"
47+
) and event.agent is None:
48+
raise OpenAIError(f"Multi-agent request ended with {event.type}")
49+
else:
50+
raise OpenAIError("Connection closed before the multi-agent response completed")
4551

4652
print()

‎tests/lib/live/test_audio_transcript_example.py‎

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
from __future__ import annotations
2+
3+
import json
4+
import wave
5+
import asyncio
6+
from pathlib import Path
7+
8+
import pytest
9+
from websockets.exceptions import ConnectionClosedOK
10+
from websockets.sync.server import ServerConnection
11+
12+
from openai import OpenAIError
13+
from examples.live.audio_transcript import run
14+
15+
from ..responses.test_websocket_session import script_server
16+
17+
18+
@pytest.mark.parametrize("stage", ["startup", "audio", "requested"])
19+
@pytest.mark.parametrize(
20+
"reason", ["close_requested", "expired", "content", "remote_hangup", "connection_lost", "error", "eof"]
21+
)
22+
async def test_audio_example_requires_its_requested_finalization(
23+
stage: str, reason: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
24+
) -> None:
25+
wav_path = tmp_path / "fixture.wav"
26+
with wave.open(str(wav_path), "wb") as wav:
27+
wav.setparams((1, 2, 24000, 0, "NONE", "not compressed"))
28+
wav.writeframes(bytes(24000 * 2 // 10))
29+
30+
sent_closed = False
31+
received_close = False
32+
audio_messages = 0
33+
34+
def script(socket: ServerConnection) -> None:
35+
nonlocal sent_closed, received_close, audio_messages
36+
37+
def send_closed() -> None:
38+
nonlocal sent_closed
39+
sent_closed = True
40+
if reason == "eof":
41+
socket.close()
42+
elif reason == "error":
43+
socket.send('{"type": "error", "event_id": "failure", "message": "Synthetic test error"}')
44+
else:
45+
socket.send(json.dumps({"type": "session.closed", "event_id": "done", "reason": reason}))
46+
47+
assert json.loads(socket.recv(timeout=3))["type"] == "session.start"
48+
if stage == "startup":
49+
send_closed()
50+
else:
51+
socket.send('{"type": "session.started", "event_id": "started", "session": {"id": "fixture"}}')
52+
# Leave the connection open after the terminal event. The client must
53+
# reject that event immediately, rather than waiting for transport EOF.
54+
try:
55+
while True:
56+
event = json.loads(socket.recv(timeout=3))
57+
if event["type"] == "session.input_audio.append":
58+
audio_messages += 1
59+
if stage == "audio" and not sent_closed:
60+
send_closed()
61+
elif event["type"] == "session.close":
62+
received_close = True
63+
if stage == "requested":
64+
# Final transcript data after close was requested must
65+
# reach the caller before the confirmed result returns.
66+
socket.send(
67+
json.dumps(
68+
{
69+
"type": "session.output_transcript.delta",
70+
"event_id": "text",
71+
"delta": "Final fixture transcript.",
72+
"start_ms": 0,
73+
"end_ms": 100,
74+
}
75+
)
76+
)
77+
send_closed()
78+
except ConnectionClosedOK:
79+
return
80+
81+
monkeypatch.setenv("OPENAI_API_KEY", "fake-transcript-fixture-key")
82+
monkeypatch.setenv("NO_PROXY", "127.0.0.1")
83+
monkeypatch.setenv("no_proxy", "127.0.0.1")
84+
# The sample constructs its own client; restrict this fixture to loopback.
85+
for name in ("ALL_PROXY", "all_proxy", "HTTP_PROXY", "http_proxy", "HTTPS_PROXY", "https_proxy"):
86+
monkeypatch.delenv(name, raising=False)
87+
with script_server(script) as url:
88+
monkeypatch.setenv("OPENAI_BASE_URL", url)
89+
if stage == "requested" and reason == "close_requested":
90+
result = await asyncio.wait_for(run(wav_path, model="fixture-model", listen_seconds=0), timeout=1)
91+
assert any(segment.text == "Final fixture transcript." for segment in result)
92+
else:
93+
# recv() can report transport EOF directly while awaiting startup;
94+
# all terminal paths must fail instead of returning partial text.
95+
with pytest.raises((OpenAIError, ConnectionClosedOK)):
96+
await asyncio.wait_for(run(wav_path, model="fixture-model", listen_seconds=0), timeout=1)
97+
assert sent_closed
98+
assert received_close == (stage == "requested")
99+
if stage == "startup":
100+
assert audio_messages == 0
101+
elif stage == "requested":
102+
assert audio_messages == 5
103+
else:
104+
assert 1 <= audio_messages <= 5

0 commit comments

Comments
 (0)