Add functionality to update agent during realtime session

Author: marysha-openai

src/agents/realtime/session.py

@@ -180,6 +180,29 @@ async def interrupt(self) -> None:
         """Interrupt the model."""
         await self._model.send_event(RealtimeModelSendInterrupt())
 
+    async def update_agent(self, agent: RealtimeAgent) -> None:
+        """Update the active agent for this session and apply its settings to the model."""
+
+        previous_agent = self._current_agent
+        self._current_agent = agent
+
+        updated_settings = await self._get_updated_model_settings_from_agent(
+            starting_settings=None,
+            agent=self._current_agent,
+        )
+
+        await self._put_event(
+            RealtimeHandoffEvent(
+                from_agent=previous_agent,
+                to_agent=self._current_agent,
+                info=self._event_info,
+            )
+        )
+
+        await self._model.send_event(
+            RealtimeModelSendSessionUpdate(session_settings=updated_settings)
+        )
+
     async def on_event(self, event: RealtimeModelEvent) -> None:
         await self._put_event(RealtimeRawModelEvent(data=event, info=self._event_info))
 

tests/realtime/test_session.py

@@ -15,6 +15,7 @@
     RealtimeAudioInterrupted,
     RealtimeError,
     RealtimeGuardrailTripped,
+    RealtimeHandoffEvent,
     RealtimeHistoryAdded,
     RealtimeHistoryUpdated,
     RealtimeRawModelEvent,
@@ -45,6 +46,7 @@
     RealtimeModelTurnEndedEvent,
     RealtimeModelTurnStartedEvent,
 )
+from agents.realtime.model_inputs import RealtimeModelSendSessionUpdate
 from agents.realtime.session import RealtimeSession
 from agents.tool import FunctionTool
 from agents.tool_context import ToolContext
@@ -1488,3 +1490,30 @@ async def mock_get_handoffs(cls, agent, context_wrapper):
             assert model_settings["voice"] == "model_config_only_voice"
             assert model_settings["tool_choice"] == "required"
             assert model_settings["output_audio_format"] == "g711_ulaw"
+
+
+class TestUpdateAgentFunctionality:
+    """Tests for update agent functionality in RealtimeSession"""
+
+    @pytest.mark.asyncio
+    async def test_update_agent_creates_handoff_and_session_update_event(self, mock_model):
+        first_agent = RealtimeAgent(name="first", instructions="first", tools=[], handoffs=[])
+        second_agent = RealtimeAgent(name="second", instructions="second", tools=[], handoffs=[])
+
+        session = RealtimeSession(mock_model, first_agent, None)
+
+        await session.update_agent(second_agent)
+
+        # Should have sent handoff event
+        handoff_event = await session._event_queue.get()
+        assert isinstance(handoff_event, RealtimeHandoffEvent)
+        assert handoff_event.from_agent == first_agent
+        assert handoff_event.to_agent == second_agent
+
+        # Should have sent session update
+        session_update_event = mock_model.sent_events[0]
+        assert isinstance(session_update_event, RealtimeModelSendSessionUpdate)
+        assert session_update_event.session_settings["instructions"] == "second"
+
+        # Check that the current agent and session settings are updated
+        assert session._current_agent == second_agent