Realtime: send session.update event at connection

rm-openai · rm-openai · commit a59987649def · 2025-07-14T16:54:28.000-04:00
diff --git a/examples/realtime/demo.py b/examples/realtime/demo.py
@@ -58,55 +58,45 @@ async def run(self) -> None:
             self.session = session
             self.ui.set_is_connected(True)
             async for event in session:
-                await self.on_event(event)
+                await self._on_event(event)
+            print("done")
 
         # 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}")
-
-    async def on_event(self, event: RealtimeSessionEvent) -> None:
-        # Display event in the UI
-        try:
-            if event.type == "agent_start":
-                self.ui.add_transcript(f"Agent started: {event.agent.name}")
-            elif event.type == "agent_end":
-                self.ui.add_transcript(f"Agent ended: {event.agent.name}")
-            elif event.type == "handoff":
-                self.ui.add_transcript(
-                    f"Handoff from {event.from_agent.name} to {event.to_agent.name}"
-                )
-            elif event.type == "tool_start":
-                self.ui.add_transcript(f"Tool started: {event.tool.name}")
-            elif event.type == "tool_end":
-                self.ui.add_transcript(f"Tool ended: {event.tool.name}; output: {event.output}")
-            elif event.type == "audio_end":
-                self.ui.add_transcript("Audio ended")
-            elif event.type == "audio":
-                np_audio = np.frombuffer(event.audio.data, dtype=np.int16)
-                self.ui.play_audio(np_audio)
-            elif event.type == "audio_interrupted":
-                self.ui.add_transcript("Audio interrupted")
-            elif event.type == "error":
-                self.ui.add_transcript(f"Error: {event.error}")
-            elif event.type == "history_updated":
-                pass
-            elif event.type == "history_added":
-                pass
-            elif event.type == "raw_model_event":
-                self.ui.log_message(f"Raw model event: {event.data}")
-            else:
-                self.ui.log_message(f"Unknown event type: {event.type}")
-        except Exception as e:
-            # This can happen if the UI has already exited
-            self.ui.log_message(f"Event handling error: {str(e)}")
+        # Send the audio to the session
+        assert self.session is not None
+        await self.session.send_audio(audio_bytes)
+
+    async def _on_event(self, event: RealtimeSessionEvent) -> None:
+        if event.type == "agent_start":
+            self.ui.add_transcript(f"Agent started: {event.agent.name}")
+        elif event.type == "agent_end":
+            self.ui.add_transcript(f"Agent ended: {event.agent.name}")
+        elif event.type == "handoff":
+            self.ui.add_transcript(f"Handoff from {event.from_agent.name} to {event.to_agent.name}")
+        elif event.type == "tool_start":
+            self.ui.add_transcript(f"Tool started: {event.tool.name}")
+        elif event.type == "tool_end":
+            self.ui.add_transcript(f"Tool ended: {event.tool.name}; output: {event.output}")
+        elif event.type == "audio_end":
+            self.ui.add_transcript("Audio ended")
+        elif event.type == "audio":
+            np_audio = np.frombuffer(event.audio.data, dtype=np.int16)
+            self.ui.play_audio(np_audio)
+        elif event.type == "audio_interrupted":
+            self.ui.add_transcript("Audio interrupted")
+        elif event.type == "error":
+            self.ui.add_transcript(f"Error: {event.error}")
+        elif event.type == "history_updated":
+            pass
+        elif event.type == "history_added":
+            pass
+        elif event.type == "raw_model_event":
+            self.ui.log_message(f"Raw model event: {event.data}")
+        else:
+            self.ui.log_message(f"Unknown event type: {event.type}")
 
 
 if __name__ == "__main__":
diff --git a/examples/realtime/ui.py b/examples/realtime/ui.py
@@ -239,10 +239,7 @@ async def capture_audio(self) -> None:
 
                 # Call audio callback if set
                 if self.audio_callback:
-                    try:
-                        await self.audio_callback(audio_bytes)
-                    except Exception as e:
-                        self.log_message(f"Audio callback error: {e}")
+                    await self.audio_callback(audio_bytes)
 
                 # Yield control back to event loop
                 await asyncio.sleep(0)
diff --git a/src/agents/realtime/openai_realtime.py b/src/agents/realtime/openai_realtime.py
@@ -8,16 +8,22 @@
 from datetime import datetime
 from typing import Any, Callable, Literal
 
+import pydantic
 import websockets
 from openai.types.beta.realtime.conversation_item import ConversationItem
 from openai.types.beta.realtime.realtime_server_event import (
     RealtimeServerEvent as OpenAIRealtimeServerEvent,
 )
 from openai.types.beta.realtime.response_audio_delta_event import ResponseAudioDeltaEvent
+from openai.types.beta.realtime.session_update_event import (
+    Session as OpenAISessionObject,
+    SessionTool as OpenAISessionTool,
+)
 from pydantic import TypeAdapter
 from typing_extensions import assert_never
 from websockets.asyncio.client import ClientConnection
 
+from agents.tool import FunctionTool, Tool
 from agents.util._types import MaybeAwaitable
 
 from ..exceptions import UserError
@@ -56,6 +62,17 @@
     RealtimeModelSendUserInput,
 )
 
+DEFAULT_MODEL_SETTINGS: RealtimeSessionModelSettings = {
+    "voice": "ash",
+    "modalities": ["text", "audio"],
+    "input_audio_format": "pcm16",
+    "output_audio_format": "pcm16",
+    "input_audio_transcription": {
+        "model": "gpt-4o-mini-transcribe",
+    },
+    "turn_detection": {"type": "semantic_vad"},
+}
+
 
 async def get_api_key(key: str | Callable[[], MaybeAwaitable[str]] | None) -> str | None:
     if isinstance(key, str):
@@ -110,6 +127,7 @@ async def connect(self, options: RealtimeModelConfig) -> None:
         }
         self._websocket = await websockets.connect(url, additional_headers=headers)
         self._websocket_task = asyncio.create_task(self._listen_for_messages())
+        await self._update_session_config(model_settings)
 
     async def _send_tracing_config(
         self, tracing_config: RealtimeModelTracingConfig | Literal["auto"] | None
@@ -127,11 +145,13 @@ async def _send_tracing_config(
 
     def add_listener(self, listener: RealtimeModelListener) -> None:
         """Add a listener to the model."""
-        self._listeners.append(listener)
+        if listener not in self._listeners:
+            self._listeners.append(listener)
 
     def remove_listener(self, listener: RealtimeModelListener) -> None:
         """Remove a listener from the model."""
-        self._listeners.remove(listener)
+        if listener in self._listeners:
+            self._listeners.remove(listener)
 
     async def _emit_event(self, event: RealtimeModelEvent) -> None:
         """Emit an event to the listeners."""
@@ -195,78 +215,55 @@ async def _send_raw_message(self, event: RealtimeModelSendRawMessage) -> None:
         """Send a raw message to the model."""
         assert self._websocket is not None, "Not connected"
 
-        try:
-            converted_event = {
-                "type": event.message["type"],
-            }
+        converted_event = {
+            "type": event.message["type"],
+        }
 
-            converted_event.update(event.message.get("other_data", {}))
+        converted_event.update(event.message.get("other_data", {}))
 
-            await self._websocket.send(json.dumps(converted_event))
-        except Exception as e:
-            await self._emit_event(
-                RealtimeModelExceptionEvent(
-                    exception=e,
-                    context=f"Failed to send event: {event.message.get('type', 'unknown')}",
-                )
-            )
+        await self._websocket.send(json.dumps(converted_event))
 
     async def _send_user_input(self, event: RealtimeModelSendUserInput) -> None:
-        """Send a user input to the model."""
-        try:
-            message = (
-                event.user_input
-                if isinstance(event.user_input, dict)
-                else {
-                    "type": "message",
-                    "role": "user",
-                    "content": [{"type": "input_text", "text": event.user_input}],
-                }
-            )
-            other_data = {
-                "item": message,
+        message = (
+            event.user_input
+            if isinstance(event.user_input, dict)
+            else {
+                "type": "message",
+                "role": "user",
+                "content": [{"type": "input_text", "text": event.user_input}],
             }
+        )
+        other_data = {
+            "item": message,
+        }
 
-            await self._send_raw_message(
-                RealtimeModelSendRawMessage(
-                    message={"type": "conversation.item.create", "other_data": other_data}
-                )
-            )
-            await self._send_raw_message(
-                RealtimeModelSendRawMessage(message={"type": "response.create"})
-            )
-        except Exception as e:
-            await self._emit_event(
-                RealtimeModelExceptionEvent(exception=e, context="Failed to send message")
+        await self._send_raw_message(
+            RealtimeModelSendRawMessage(
+                message={"type": "conversation.item.create", "other_data": other_data}
             )
+        )
+        await self._send_raw_message(
+            RealtimeModelSendRawMessage(message={"type": "response.create"})
+        )
 
     async def _send_audio(self, event: RealtimeModelSendAudio) -> None:
-        """Send audio to the model."""
-        assert self._websocket is not None, "Not connected"
-
-        try:
-            base64_audio = base64.b64encode(event.audio).decode("utf-8")
-            await self._send_raw_message(
-                RealtimeModelSendRawMessage(
-                    message={
-                        "type": "input_audio_buffer.append",
-                        "other_data": {
-                            "audio": base64_audio,
-                        },
-                    }
-                )
+        base64_audio = base64.b64encode(event.audio).decode("utf-8")
+        await self._send_raw_message(
+            RealtimeModelSendRawMessage(
+                message={
+                    "type": "input_audio_buffer.append",
+                    "other_data": {
+                        "audio": base64_audio,
+                    },
+                }
             )
-            if event.commit:
-                await self._send_raw_message(
-                    RealtimeModelSendRawMessage(message={"type": "input_audio_buffer.commit"})
-                )
-        except Exception as e:
-            await self._emit_event(
-                RealtimeModelExceptionEvent(exception=e, context="Failed to send audio")
+        )
+        if event.commit:
+            await self._send_raw_message(
+                RealtimeModelSendRawMessage(message={"type": "input_audio_buffer.commit"})
             )
 
     async def _send_tool_output(self, event: RealtimeModelSendToolOutput) -> None:
-        """Send tool output to the model."""
         await self._send_raw_message(
             RealtimeModelSendRawMessage(
                 message={
@@ -299,7 +296,6 @@ async def _send_tool_output(self, event: RealtimeModelSendToolOutput) -> None:
             )
 
     async def _send_interrupt(self, event: RealtimeModelSendInterrupt) -> None:
-        """Send an interrupt to the model."""
         if not self._current_item_id or not self._audio_start_time:
             return
 
@@ -418,8 +414,17 @@ async def _handle_ws_event(self, event: dict[str, Any]):
             parsed: OpenAIRealtimeServerEvent = TypeAdapter(
                 OpenAIRealtimeServerEvent
             ).validate_python(event)
+        except pydantic.ValidationError as e:
+            logger.error(f"Failed to validate server event: {event}", exc_info=True)
+            await self._emit_event(
+                RealtimeModelErrorEvent(
+                    error=e,
+                )
+            )
+            return
         except Exception as e:
             event_type = event.get("type", "unknown") if isinstance(event, dict) else "unknown"
+            logger.error(f"Failed to validate server event: {event}", exc_info=True)
             await self._emit_event(
                 RealtimeModelExceptionEvent(
                     exception=e,
@@ -492,3 +497,66 @@ async def _handle_ws_event(self, event: dict[str, Any]):
             or parsed.type == "response.output_item.done"
         ):
             await self._handle_output_item(parsed.item)
+
+    async def _update_session_config(self, model_settings: RealtimeSessionModelSettings) -> None:
+        session_config = self._get_session_config(model_settings)
+        await self._send_raw_message(
+            RealtimeModelSendRawMessage(
+                message={
+                    "type": "session.update",
+                    "other_data": {
+                        "session": session_config.model_dump(exclude_unset=True, exclude_none=True)
+                    },
+                }
+            )
+        )
+
+    def _get_session_config(
+        self, model_settings: RealtimeSessionModelSettings
+    ) -> OpenAISessionObject:
+        """Get the session config."""
+        return OpenAISessionObject(
+            instructions=model_settings.get("instructions", None),
+            model=(
+                model_settings.get("model_name", self.model)  # type: ignore
+                or DEFAULT_MODEL_SETTINGS.get("model_name")
+            ),
+            voice=model_settings.get("voice", DEFAULT_MODEL_SETTINGS.get("voice")),
+            modalities=model_settings.get("modalities", DEFAULT_MODEL_SETTINGS.get("modalities")),
+            input_audio_format=model_settings.get(
+                "input_audio_format",
+                DEFAULT_MODEL_SETTINGS.get("input_audio_format"),  # type: ignore
+            ),
+            output_audio_format=model_settings.get(
+                "output_audio_format",
+                DEFAULT_MODEL_SETTINGS.get("output_audio_format"),  # type: ignore
+            ),
+            input_audio_transcription=model_settings.get(
+                "input_audio_transcription",
+                DEFAULT_MODEL_SETTINGS.get("input_audio_transcription"),  # type: ignore
+            ),
+            turn_detection=model_settings.get(
+                "turn_detection",
+                DEFAULT_MODEL_SETTINGS.get("turn_detection"),  # type: ignore
+            ),
+            tool_choice=model_settings.get(
+                "tool_choice",
+                DEFAULT_MODEL_SETTINGS.get("tool_choice"),  # type: ignore
+            ),
+            tools=self._tools_to_session_tools(model_settings.get("tools", [])),
+        )
+
+    def _tools_to_session_tools(self, tools: list[Tool]) -> list[OpenAISessionTool]:
+        converted_tools: list[OpenAISessionTool] = []
+        for tool in tools:
+            if not isinstance(tool, FunctionTool):
+                raise UserError(f"Tool {tool.name} is unsupported. Must be a function tool.")
+            converted_tools.append(
+                OpenAISessionTool(
+                    name=tool.name,
+                    description=tool.description,
+                    parameters=tool.params_json_schema,
+                    type="function",
+                )
+            )
+        return converted_tools
diff --git a/src/agents/realtime/session.py b/src/agents/realtime/session.py
