[8/n] Make realtime more like the rest of agents sdk

Key changes:
1. Transport -> model.
2. Extract any model settings into `RealtimeSessionModelSettings`.
3. RealtimeRunConfig, similar to the RunConfig in `run.py`.

Next PR I'll update session to be better.

Author: rm-openai

examples/realtime/demo.py

@@ -5,11 +5,13 @@
 
 import numpy as np
 
+from agents.realtime import RealtimeSession
+
 # Add the current directory to path so we can import ui
 sys.path.append(os.path.dirname(os.path.abspath(__file__)))
 
 from agents import function_tool
-from agents.realtime import RealtimeAgent, RealtimeSession, RealtimeSessionEvent
+from agents.realtime import RealtimeAgent, RealtimeRunner, RealtimeSessionEvent
 
 if TYPE_CHECKING:
     from .ui import AppUI
@@ -38,23 +40,34 @@ def get_weather(city: str) -> str:
 
 class Example:
     def __init__(self) -> None:
-        self.session = RealtimeSession(agent)
         self.ui = AppUI()
         self.ui.connected = asyncio.Event()
         self.ui.last_audio_item_id = None
         # Set the audio callback
         self.ui.set_audio_callback(self.on_audio_recorded)
 
+        self.session: RealtimeSession | None = None
+
     async def run(self) -> None:
-        self.session.add_listener(self.on_event)
-        await self.session.connect()
-        self.ui.set_is_connected(True)
-        await self.ui.run_async()
+        # Start UI in a separate task instead of waiting for it to complete
+        ui_task = asyncio.create_task(self.ui.run_async())
+
+        # Set up session immediately without waiting for UI to finish
+        runner = RealtimeRunner(agent)
+        async with await runner.run() as session:
+            self.session = session
+            self.ui.set_is_connected(True)
+            async for event in session:
+                await self.on_event(event)
+
+        # Wait for UI task to complete when session ends
+        await ui_task
 
     async def on_audio_recorded(self, audio_bytes: bytes) -> None:
         """Called when audio is recorded by the UI."""
         try:
             # Send the audio to the session
+            assert self.session is not None
             await self.session.send_audio(audio_bytes)
         except Exception as e:
             self.ui.log_message(f"Error sending audio: {e}")

src/agents/realtime/__init__.py

@@ -1,5 +1,17 @@
 from .agent import RealtimeAgent, RealtimeAgentHooks, RealtimeRunHooks
-from .config import APIKeyOrKeyFunc
+from .config import (
+    RealtimeAudioFormat,
+    RealtimeClientMessage,
+    RealtimeInputAudioTranscriptionConfig,
+    RealtimeModelName,
+    RealtimeRunConfig,
+    RealtimeSessionModelSettings,
+    RealtimeTurnDetectionConfig,
+    RealtimeUserInput,
+    RealtimeUserInputMessage,
+    RealtimeUserInputText,
+)
+from .connection import RealtimeSession
 from .events import (
     RealtimeAgentEndEvent,
     RealtimeAgentStartEvent,
@@ -15,26 +27,33 @@
     RealtimeToolEnd,
     RealtimeToolStart,
 )
-from .session import RealtimeSession
-from .transport import (
-    RealtimeModelName,
-    RealtimeSessionTransport,
-    RealtimeTransportConnectionOptions,
-    RealtimeTransportListener,
+from .model import (
+    RealtimeModel,
+    RealtimeModelConfig,
+    RealtimeModelListener,
 )
+from .runner import RealtimeRunner
 
 __all__ = [
     "RealtimeAgent",
     "RealtimeAgentHooks",
     "RealtimeRunHooks",
-    "RealtimeSession",
+    "RealtimeRunner",
     "RealtimeSessionListener",
     "RealtimeSessionListenerFunc",
-    "APIKeyOrKeyFunc",
+    "RealtimeRunConfig",
+    "RealtimeSessionModelSettings",
+    "RealtimeInputAudioTranscriptionConfig",
+    "RealtimeTurnDetectionConfig",
+    "RealtimeAudioFormat",
+    "RealtimeClientMessage",
+    "RealtimeUserInput",
+    "RealtimeUserInputMessage",
+    "RealtimeUserInputText",
     "RealtimeModelName",
-    "RealtimeSessionTransport",
-    "RealtimeTransportListener",
-    "RealtimeTransportConnectionOptions",
+    "RealtimeModel",
+    "RealtimeModelListener",
+    "RealtimeModelConfig",
     "RealtimeSessionEvent",
     "RealtimeAgentStartEvent",
     "RealtimeAgentEndEvent",
@@ -48,4 +67,5 @@
     "RealtimeError",
     "RealtimeHistoryUpdated",
     "RealtimeHistoryAdded",
+    "RealtimeSession",
 ]

src/agents/realtime/config.py

@@ -1,34 +1,44 @@
 from __future__ import annotations
 
-import inspect
 from typing import (
     Any,
-    Callable,
     Literal,
     Union,
 )
 
 from typing_extensions import NotRequired, TypeAlias, TypedDict
 
 from ..model_settings import ToolChoice
-from ..tool import FunctionTool
-from ..util._types import MaybeAwaitable
+from ..tool import Tool
+
+RealtimeModelName: TypeAlias = Union[
+    Literal[
+        "gpt-4o-realtime-preview",
+        "gpt-4o-mini-realtime-preview",
+        "gpt-4o-realtime-preview-2025-06-03",
+        "gpt-4o-realtime-preview-2024-12-17",
+        "gpt-4o-realtime-preview-2024-10-01",
+        "gpt-4o-mini-realtime-preview-2024-12-17",
+    ],
+    str,
+]
+"""The name of a realtime model."""
 
 
 class RealtimeClientMessage(TypedDict):
     type: str  # explicitly required
     other_data: NotRequired[dict[str, Any]]
 
 
-class UserInputText(TypedDict):
+class RealtimeUserInputText(TypedDict):
     type: Literal["input_text"]
     text: str
 
 
 class RealtimeUserInputMessage(TypedDict):
     type: Literal["message"]
     role: Literal["user"]
-    content: list[UserInputText]
+    content: list[RealtimeUserInputText]
 
 
 RealtimeUserInput: TypeAlias = Union[str, RealtimeUserInputMessage]
@@ -55,9 +65,11 @@ class RealtimeTurnDetectionConfig(TypedDict):
     threshold: NotRequired[float]
 
 
-class RealtimeSessionConfig(TypedDict):
-    api_key: NotRequired[APIKeyOrKeyFunc]
-    model: NotRequired[str]
+class RealtimeSessionModelSettings(TypedDict):
+    """Model settings for a realtime model session."""
+
+    model_name: NotRequired[RealtimeModelName]
+
     instructions: NotRequired[str]
     modalities: NotRequired[list[Literal["text", "audio"]]]
     voice: NotRequired[str]
@@ -68,24 +80,13 @@ class RealtimeSessionConfig(TypedDict):
     turn_detection: NotRequired[RealtimeTurnDetectionConfig]
 
     tool_choice: NotRequired[ToolChoice]
-    tools: NotRequired[list[FunctionTool]]
-
-
-APIKeyOrKeyFunc = str | Callable[[], MaybeAwaitable[str]]
-"""Either an API key or a function that returns an API key."""
-
+    tools: NotRequired[list[Tool]]
 
-async def get_api_key(key: APIKeyOrKeyFunc | None) -> str | None:
-    """Get the API key from the key or key function."""
-    if key is None:
-        return None
-    elif isinstance(key, str):
-        return key
 
-    result = key()
-    if inspect.isawaitable(result):
-        return await result
-    return result
+class RealtimeRunConfig(TypedDict):
+    model_settings: NotRequired[RealtimeSessionModelSettings]
 
     # TODO (rm) Add tracing support
     # tracing: NotRequired[RealtimeTracingConfig | None]
+    # TODO (rm) Add guardrail support
+    # TODO (rm) Add history audio storage config