feat: adjust relatime examples for Realtime API Beta

examples/realtime/azure_realtime.py

@@ -1,5 +1,29 @@
-import os
+#!/usr/bin/env uv run
+#
+# /// script
+# requires-python = ">=3.9"
+# dependencies = [
+# "textual",
+# "numpy",
+# "pyaudio",
+# "pydub",
+# "sounddevice",
+# "openai[realtime]",
+# "azure-identity",
+# "aiohttp",
+# "python-dotenv",
+# ]
+#
+# [tool.uv.sources]
+# openai = { path = "../../", editable = true }
+# ///
+
+from dotenv import load_dotenv
+load_dotenv()
+
 import asyncio
+import base64
+import os
 
 from azure.identity.aio import DefaultAzureCredential, get_bearer_token_provider
 
@@ -22,20 +46,23 @@ async def main() -> None:
 
  credential = DefaultAzureCredential()
  client = AsyncAzureOpenAI(
+ azure_deployment="gpt-realtime",
  azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
  azure_ad_token_provider=get_bearer_token_provider(credential, "https://cognitiveservices.azure.com/.default"),
- api_version="2024-10-01-preview",
+ api_version="2025-04-01-preview",
  )
- async with client.realtime.connect(
+
+ async with client.beta.realtime.connect(
  model="gpt-realtime", # deployment name for your model
  ) as connection:
  await connection.session.update(
  session={
- "output_modalities": ["text"],
- "model": "gpt-realtime",
- "type": "realtime",
+ # "output_modalities": ["text"],
+ # "model": "gpt-realtime",
+ # "type": "realtime",
  }
  )
+
  while True:
  user_input = input("Enter a message: ")
  if user_input == "q":
@@ -48,14 +75,28 @@ async def main() -> None:
  "content": [{"type": "input_text", "text": user_input}],
  }
  )
+
  await connection.response.create()
  async for event in connection:
- if event.type == "response.output_text.delta":
+ print(f"Event: {event.type}")
+
+ if event.type == "error":
+ print(f"ERROR: {event}")
+
+ if event.type == "response.text.delta":
  print(event.delta, flush=True, end="")
- elif event.type == "response.output_text.done":
+ if event.type == "response.text.done":
  print()
- elif event.type == "response.done":
- break
+ if event.type == "response.done":
+ print(f"final response: {event.response.output[0].content[0].transcript}")
+ print(f"usage: {event.response.usage}")
+
+ if event.type == "response.audio.delta":
+ audio_data = base64.b64decode(event.delta)
+ print(f"Received {len(audio_data)} bytes of audio data.")
+
+ if event.type == "response.audio_transcript.delta":
+ print(f"Received text delta: {event.delta}")
 
  await credential.close()
 

examples/realtime/push_to_talk_app.py

@@ -18,28 +18,36 @@
 # "pydub",
 # "sounddevice",
 # "openai[realtime]",
+# "azure-identity",
+# "aiohttp",
+# "python-dotenv",
 # ]
 #
 # [tool.uv.sources]
 # openai = { path = "../../", editable = true }
 # ///
 from __future__ import annotations
 
+from dotenv import load_dotenv
+load_dotenv()
+
 import base64
 import asyncio
+import os
 from typing import Any, cast
 from typing_extensions import override
 
 from textual import events
 from audio_util import CHANNELS, SAMPLE_RATE, AudioPlayerAsync
 from textual.app import App, ComposeResult
-from textual.widgets import Button, Static, RichLog
+from textual.widgets import Static, RichLog
 from textual.reactive import reactive
 from textual.containers import Container
 
-from openai import AsyncOpenAI
-from openai.types.realtime.session import Session
+from openai import AsyncAzureOpenAI
+from openai.types.realtime.session_update_event import Session
 from openai.resources.realtime.realtime import AsyncRealtimeConnection
+from azure.identity.aio import DefaultAzureCredential, get_bearer_token_provider
 
 
 class SessionDisplay(Static):
@@ -60,7 +68,9 @@ class AudioStatusIndicator(Static):
  @override
  def render(self) -> str:
  status = (
- "🔴 Recording... (Press K to stop)" if self.is_recording else "⚪ Press K to start recording (Q to quit)"
+ "🔴 Recording... (Press K to stop)"
+ if self.is_recording
+ else "⚪ Press K to start recording (Q to quit)"
  )
  return status
 
@@ -123,7 +133,7 @@ class RealtimeApp(App[None]):
  }
  """
 
- client: AsyncOpenAI
+ client: AsyncAzureOpenAI
  should_send_audio: asyncio.Event
  audio_player: AudioPlayerAsync
  last_audio_item_id: str | None
@@ -135,7 +145,15 @@ def __init__(self) -> None:
  super().__init__()
  self.connection = None
  self.session = None
- self.client = AsyncOpenAI()
+ credential = DefaultAzureCredential()
+ self.client = AsyncAzureOpenAI(
+ azure_deployment="gpt-realtime",
+ azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
+ azure_ad_token_provider=get_bearer_token_provider(
+ credential, "https://cognitiveservices.azure.com/.default"
+ ),
+ api_version="2025-04-01-preview",
+ )
  self.audio_player = AudioPlayerAsync()
  self.last_audio_item_id = None
  self.should_send_audio = asyncio.Event()
@@ -247,7 +265,9 @@ async def send_mic_audio(self) -> None:
  asyncio.create_task(connection.send({"type": "response.cancel"}))
  sent_audio = True
 
- await connection.input_audio_buffer.append(audio=base64.b64encode(cast(Any, data)).decode("utf-8"))
+ await connection.input_audio_buffer.append(
+ audio=base64.b64encode(cast(Any, data)).decode("utf-8")
+ )
 
  await asyncio.sleep(0)
  except KeyboardInterrupt:
@@ -258,10 +278,6 @@ async def send_mic_audio(self) -> None:
 
  async def on_key(self, event: events.Key) -> None:
  """Handle key press events."""
- if event.key == "enter":
- self.query_one(Button).press()
- return
-
  if event.key == "q":
  self.exit()
  return