From ead03d3104e14d3b5939286e2869288581b40ab7 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Wed, 28 Oct 2026 22:53:35 +0000 Subject: [PATCH 1/2] Fix Realtime example keepalive while awaiting input --- examples/realtime/realtime.py | 2 +- tests/lib/test_realtime_example_keepalive.py | 56 ++++++++++++++++++++ 2 files changed, 57 insertions(+), 1 deletion(-) create mode 100644 tests/lib/test_realtime_example_keepalive.py diff --git a/examples/realtime/realtime.py b/examples/realtime/realtime.py index f70dd0b9c6..a1e21ed932 100755 --- a/examples/realtime/realtime.py +++ b/examples/realtime/realtime.py @@ -30,7 +30,7 @@ async def main() -> None: } ) while True: - user_input = input("Enter a message: ") + user_input = await asyncio.to_thread(input, "Enter a message: ") if user_input == "q": break diff --git a/tests/lib/test_realtime_example_keepalive.py b/tests/lib/test_realtime_example_keepalive.py new file mode 100644 index 0000000000..a9e137b78e --- /dev/null +++ b/tests/lib/test_realtime_example_keepalive.py @@ -0,0 +1,56 @@ +import os +import sys +import json +import asyncio +from pathlib import Path + +import pytest +from websockets.asyncio.server import ServerConnection, serve + + +@pytest.mark.asyncio +async def test_interactive_realtime_example_answers_ping_while_waiting_for_input() -> None: + root = Path(__file__).resolve().parents[2] + connected: asyncio.Future[ServerConnection] = asyncio.get_running_loop().create_future() + + async def accept(socket: ServerConnection) -> None: + connected.set_result(socket) + await socket.wait_closed() + + async with serve(accept, "127.0.0.1", 0, ping_interval=None) as server: + port = server.sockets[0].getsockname()[1] + env = { + **os.environ, + "OPENAI_API_KEY": "example-test-key", + "OPENAI_BASE_URL": f"http://127.0.0.1:{port}/v1", + "PYTHONPATH": str(root / "src"), + "PYTHONUNBUFFERED": "1", + } + # This subprocess only connects to the local test peer. + for name in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy"): + env.pop(name, None) + + process = await asyncio.create_subprocess_exec( + sys.executable, + str(root / "examples" / "realtime" / "realtime.py"), + env=env, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + assert process.stdin is not None + assert process.stdout is not None + try: + socket = await asyncio.wait_for(connected, timeout=10) + event = json.loads(await asyncio.wait_for(socket.recv(), timeout=10)) + assert event["type"] == "session.update" + prompt = b"Enter a message: " + assert await asyncio.wait_for(process.stdout.readexactly(len(prompt)), timeout=10) == prompt + pong = await socket.ping() + await asyncio.wait_for(pong, timeout=2) + finally: + if process.returncode is None: + process.stdin.write(b"q\n") + await process.stdin.drain() + _, stderr = await asyncio.wait_for(process.communicate(), timeout=10) + assert process.returncode == 0, stderr.decode() From 1a05a26209130ccd90e44d7f3106ba694b66740f Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Wed, 28 Oct 2026 23:15:05 +0000 Subject: [PATCH 2/2] Keep Realtime console input interruptible --- examples/realtime/realtime.py | 22 ++-- tests/lib/test_realtime_example_keepalive.py | 104 +++++++++++++++++-- 2 files changed, 108 insertions(+), 18 deletions(-) diff --git a/examples/realtime/realtime.py b/examples/realtime/realtime.py index a1e21ed932..3f4ad1b975 100755 --- a/examples/realtime/realtime.py +++ b/examples/realtime/realtime.py @@ -1,7 +1,5 @@ #!/usr/bin/env -S uv run python -import asyncio - -from openai import AsyncOpenAI +from openai import OpenAI # Azure OpenAI Realtime Docs @@ -10,7 +8,7 @@ # Entra ID auth: https://learn.microsoft.com/azure/ai-services/openai/how-to/managed-identity -async def main() -> None: +def main() -> None: """The following example demonstrates how to configure OpenAI to use the Realtime API. For an audio example, see push_to_talk_app.py and update the client and model parameter accordingly. @@ -18,11 +16,11 @@ async def main() -> None: Enter "q" to quit the conversation. """ - client = AsyncOpenAI() - async with client.realtime.connect( + client = OpenAI() + with client.realtime.connect( model="gpt-realtime", ) as connection: - await connection.session.update( + connection.session.update( session={ "output_modalities": ["text"], "model": "gpt-realtime", @@ -30,19 +28,19 @@ async def main() -> None: } ) while True: - user_input = await asyncio.to_thread(input, "Enter a message: ") + user_input = input("Enter a message: ") if user_input == "q": break - await connection.conversation.item.create( + 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: + connection.response.create() + for event in connection: if event.type == "response.output_text.delta": print(event.delta, flush=True, end="") elif event.type == "response.output_text.done": @@ -51,4 +49,4 @@ async def main() -> None: break -asyncio.run(main()) +main() diff --git a/tests/lib/test_realtime_example_keepalive.py b/tests/lib/test_realtime_example_keepalive.py index a9e137b78e..daca485716 100644 --- a/tests/lib/test_realtime_example_keepalive.py +++ b/tests/lib/test_realtime_example_keepalive.py @@ -1,6 +1,7 @@ import os import sys import json +import signal import asyncio from pathlib import Path @@ -9,7 +10,20 @@ @pytest.mark.asyncio -async def test_interactive_realtime_example_answers_ping_while_waiting_for_input() -> None: +@pytest.mark.parametrize( + "exit_action", + [ + "conversation_then_q", + "eof", + pytest.param( + "sigint", + marks=pytest.mark.skipif( + sys.platform == "win32", reason="Windows uses console events instead of POSIX SIGINT" + ), + ), + ], +) +async def test_interactive_realtime_example_remains_responsive_at_input(exit_action: str) -> None: root = Path(__file__).resolve().parents[2] connected: asyncio.Future[ServerConnection] = asyncio.get_running_loop().create_future() @@ -18,7 +32,7 @@ async def accept(socket: ServerConnection) -> None: await socket.wait_closed() async with serve(accept, "127.0.0.1", 0, ping_interval=None) as server: - port = server.sockets[0].getsockname()[1] + port = next(iter(server.sockets)).getsockname()[1] env = { **os.environ, "OPENAI_API_KEY": "example-test-key", @@ -48,9 +62,87 @@ async def accept(socket: ServerConnection) -> None: assert await asyncio.wait_for(process.stdout.readexactly(len(prompt)), timeout=10) == prompt pong = await socket.ping() await asyncio.wait_for(pong, timeout=2) - finally: - if process.returncode is None: + + if exit_action == "conversation_then_q": + process.stdin.write(b"Hello from the example test\n") + await process.stdin.drain() + item_event = json.loads(await asyncio.wait_for(socket.recv(), timeout=10)) + assert item_event["type"] == "conversation.item.create" + assert item_event["item"]["content"] == [{"type": "input_text", "text": "Hello from the example test"}] + response_event = json.loads(await asyncio.wait_for(socket.recv(), timeout=10)) + assert response_event["type"] == "response.create" + await socket.send( + json.dumps( + { + "type": "response.output_text.delta", + "event_id": "event_delta", + "response_id": "resp_example", + "item_id": "item_example", + "output_index": 0, + "content_index": 0, + "delta": "Hello back", + } + ) + ) + await socket.send( + json.dumps( + { + "type": "response.output_text.done", + "event_id": "event_text_done", + "response_id": "resp_example", + "item_id": "item_example", + "output_index": 0, + "content_index": 0, + "text": "Hello back", + } + ) + ) + await socket.send( + json.dumps( + { + "type": "response.done", + "event_id": "event_response_done", + "response": { + "id": "resp_example", + "object": "realtime.response", + "status": "completed", + "status_details": None, + "output": [ + { + "id": "item_example", + "object": "realtime.item", + "type": "message", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "Hello back"}], + } + ], + }, + } + ) + ) + output = await asyncio.wait_for(process.stdout.readuntil(prompt), timeout=10) + assert output == b"Hello back" + os.linesep.encode() + prompt + pong = await socket.ping() + await asyncio.wait_for(pong, timeout=2) process.stdin.write(b"q\n") await process.stdin.drain() - _, stderr = await asyncio.wait_for(process.communicate(), timeout=10) - assert process.returncode == 0, stderr.decode() + elif exit_action == "eof": + process.stdin.close() + else: + process.send_signal(signal.SIGINT) + + await asyncio.wait_for(process.wait(), timeout=2) + _, stderr = await process.communicate() + if exit_action == "conversation_then_q": + assert process.returncode == 0, stderr.decode() + elif exit_action == "eof": + assert process.returncode != 0 + assert b"EOFError" in stderr + else: + assert process.returncode != 0 + assert b"KeyboardInterrupt" in stderr + finally: + if process.returncode is None: + process.kill() + await process.communicate()