[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

.vscode/launch.json

@@ -0,0 +1,14 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Python Debugger: Python File",
+            "type": "debugpy",
+            "request": "launch",
+            "program": "${file}"
+        }
+    ]
+}

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,16 @@
 from .agent import RealtimeAgent, RealtimeAgentHooks, RealtimeRunHooks
-from .config import APIKeyOrKeyFunc
+from .config import (
+    RealtimeAudioFormat,
+    RealtimeClientMessage,
+    RealtimeInputAudioTranscriptionConfig,
+    RealtimeModelName,
+    RealtimeRunConfig,
+    RealtimeSessionModelSettings,
+    RealtimeTurnDetectionConfig,
+    RealtimeUserInput,
+    RealtimeUserInputMessage,
+    RealtimeUserInputText,
+)
 from .events import (
     RealtimeAgentEndEvent,
     RealtimeAgentStartEvent,
@@ -15,26 +26,34 @@
     RealtimeToolEnd,
     RealtimeToolStart,
 )
-from .session import RealtimeSession
-from .transport import (
-    RealtimeModelName,
-    RealtimeSessionTransport,
-    RealtimeTransportConnectionOptions,
-    RealtimeTransportListener,
+from .model import (
+    RealtimeModel,
+    RealtimeModelConfig,
+    RealtimeModelListener,
 )
+from .runner import RealtimeRunner
+from .session import RealtimeSession
 
 __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

src/agents/realtime/events.py

@@ -7,7 +7,7 @@
 from ..tool import Tool
 from .agent import RealtimeAgent
 from .items import RealtimeItem
-from .transport_events import RealtimeTransportAudioEvent, RealtimeTransportEvent
+from .model_events import RealtimeModelAudioEvent, RealtimeModelEvent
 
 
 @dataclass
@@ -96,7 +96,7 @@ class RealtimeToolEnd:
 class RealtimeRawTransportEvent:
     """Forwards raw events from the transport layer."""
 
-    data: RealtimeTransportEvent
+    data: RealtimeModelEvent
     """The raw data from the transport layer."""
 
     info: RealtimeEventInfo
@@ -119,7 +119,7 @@ class RealtimeAudioEnd:
 class RealtimeAudio:
     """Triggered when the agent generates new audio to be played."""
 
-    audio: RealtimeTransportAudioEvent
+    audio: RealtimeModelAudioEvent
     """The audio event from the transport layer."""
 
     info: RealtimeEventInfo

src/agents/realtime/model.py

@@ -0,0 +1,97 @@
+import abc
+from typing import Any, Callable
+
+from typing_extensions import NotRequired, TypedDict
+
+from ..util._types import MaybeAwaitable
+from .config import (
+    RealtimeClientMessage,
+    RealtimeSessionModelSettings,
+    RealtimeUserInput,
+)
+from .model_events import RealtimeModelEvent, RealtimeModelToolCallEvent
+
+
+class RealtimeModelListener(abc.ABC):
+    """A listener for realtime transport events."""
+
+    @abc.abstractmethod
+    async def on_event(self, event: RealtimeModelEvent) -> None:
+        """Called when an event is emitted by the realtime transport."""
+        pass
+
+
+class RealtimeModelConfig(TypedDict):
+    """Options for connecting to a realtime model."""
+
+    api_key: NotRequired[str | Callable[[], MaybeAwaitable[str]]]
+    """The API key (or function that returns a key) to use when connecting. If unset, the model will
+    try to use a sane default. For example, the OpenAI Realtime model will try to use the
+    `OPENAI_API_KEY`  environment variable.
+    """
+
+    url: NotRequired[str]
+    """The URL to use when connecting. If unset, the model will use a sane default. For example,
+    the OpenAI Realtime model will use the default OpenAI WebSocket URL.
+    """
+
+    initial_model_settings: NotRequired[RealtimeSessionModelSettings]
+
+
+class RealtimeModel(abc.ABC):
+    """Interface for connecting to a realtime model and sending/receiving events."""
+
+    @abc.abstractmethod
+    async def connect(self, options: RealtimeModelConfig) -> None:
+        """Establish a connection to the model and keep it alive."""
+        pass
+
+    @abc.abstractmethod
+    def add_listener(self, listener: RealtimeModelListener) -> None:
+        """Add a listener to the model."""
+        pass
+
+    @abc.abstractmethod
+    def remove_listener(self, listener: RealtimeModelListener) -> None:
+        """Remove a listener from the model."""
+        pass
+
+    @abc.abstractmethod
+    async def send_event(self, event: RealtimeClientMessage) -> None:
+        """Send an event to the model."""
+        pass
+
+    @abc.abstractmethod
+    async def send_message(
+        self, message: RealtimeUserInput, other_event_data: dict[str, Any] | None = None
+    ) -> None:
+        """Send a message to the model."""
+        pass
+
+    @abc.abstractmethod
+    async def send_audio(self, audio: bytes, *, commit: bool = False) -> None:
+        """Send a raw audio chunk to the model.
+
+        Args:
+            audio: The audio data to send.
+            commit: Whether to commit the audio buffer to the model.  If the model does not do turn
+            detection, this can be used to indicate the turn is completed.
+        """
+        pass
+
+    @abc.abstractmethod
+    async def send_tool_output(
+        self, tool_call: RealtimeModelToolCallEvent, output: str, start_response: bool
+    ) -> None:
+        """Send tool output to the model."""
+        pass
+
+    @abc.abstractmethod
+    async def interrupt(self) -> None:
+        """Interrupt the model. For example, could be triggered by a guardrail."""
+        pass
+
+    @abc.abstractmethod
+    async def close(self) -> None:
+        """Close the session."""
+        pass