fix: send realtime tool failure outputs

src/agents/realtime/session.py

@@ -574,6 +574,19 @@ async def _send_tool_rejection(
             )
         )
 
+    async def _send_tool_exception_output(
+        self,
+        event: RealtimeModelToolCallEvent,
+        exception: Exception,
+    ) -> None:
+        await self._model.send_event(
+            RealtimeModelSendToolOutput(
+                tool_call=event,
+                output=str(exception) or type(exception).__name__,
+                start_response=True,
+            )
+        )
+
     async def _resolve_approval_rejection_message(self, *, tool: FunctionTool, call_id: str) -> str:
         """Resolve model-visible output text for approval rejections."""
         explicit_message = self._context_wrapper.get_rejection_message(
@@ -694,11 +707,15 @@ async def _handle_tool_call(
                 tool_arguments=event.arguments,
                 agent=agent,
             )
-            result = await invoke_function_tool(
-                function_tool=func_tool,
-                context=tool_context,
-                arguments=event.arguments,
-            )
+            try:
+                result = await invoke_function_tool(
+                    function_tool=func_tool,
+                    context=tool_context,
+                    arguments=event.arguments,
+                )
+            except Exception as exc:
+                await self._send_tool_exception_output(event, exc)
+                raise
 
             await self._model.send_event(
                 RealtimeModelSendToolOutput(
@@ -729,7 +746,11 @@ async def _handle_tool_call(
             )
 
             # Execute the handoff to get the new agent
-            result = await handoff.on_invoke_handoff(self._context_wrapper, event.arguments)
+            try:
+                result = await handoff.on_invoke_handoff(self._context_wrapper, event.arguments)
+            except Exception as exc:
+                await self._send_tool_exception_output(event, exc)
+                raise
             if not isinstance(result, RealtimeAgent):
                 raise UserError(
                     f"Handoff {handoff.tool_name} returned invalid result: {type(result)}"

tests/realtime/test_session.py

@@ -1109,7 +1109,11 @@ async def invoke_slow_tool(_ctx: ToolContext[Any], _arguments: str) -> str:
         with pytest.raises(ToolTimeoutError, match="timed out"):
             await session._handle_tool_call(tool_call_event)
 
-        assert len(mock_model.sent_tool_outputs) == 0
+        assert len(mock_model.sent_tool_outputs) == 1
+        sent_call, sent_output, start_response = mock_model.sent_tool_outputs[0]
+        assert sent_call == tool_call_event
+        assert "timed out" in sent_output
+        assert start_response is True
         assert session._event_queue.qsize() == 1
 
         tool_start_event = await session._event_queue.get()
@@ -1196,7 +1200,11 @@ async def invoke_slow_tool(_ctx: ToolContext[Any], _arguments: str) -> str:
 
         assert isinstance(session._stored_exception, ToolTimeoutError)
         assert session._stored_exception.tool_name == "slow_tool"
-        assert len(mock_model.sent_tool_outputs) == 0
+        assert len(mock_model.sent_tool_outputs) == 1
+        sent_call, sent_output, start_response = mock_model.sent_tool_outputs[0]
+        assert sent_call == tool_call_event
+        assert "timed out" in sent_output
+        assert start_response is True
 
         events = []
         while True:
@@ -1286,6 +1294,34 @@ async def test_handoff_tool_handling(self, mock_model):
         # Verify agent was updated
         assert session._current_agent == second_agent
 
+    @pytest.mark.asyncio
+    async def test_handoff_tool_exception_sends_model_visible_output(self, mock_model):
+        target = RealtimeAgent(name="target_agent")
+        handoff = Handoff(
+            tool_name="switch_agent",
+            tool_description="switch",
+            input_json_schema={},
+            on_invoke_handoff=AsyncMock(side_effect=ValueError("handoff failed")),
+            input_filter=None,
+            agent_name=target.name,
+            is_enabled=True,
+        )
+        first_agent = RealtimeAgent(name="first_agent", handoffs=[handoff])
+        session = RealtimeSession(mock_model, first_agent, None)
+
+        tool_call_event = RealtimeModelToolCallEvent(
+            name="switch_agent", call_id="call_handoff_fail", arguments="{}"
+        )
+
+        with pytest.raises(ValueError, match="handoff failed"):
+            await session._handle_tool_call(tool_call_event)
+
+        assert len(mock_model.sent_tool_outputs) == 1
+        sent_call, sent_output, start_response = mock_model.sent_tool_outputs[0]
+        assert sent_call == tool_call_event
+        assert sent_output == "handoff failed"
+        assert start_response is True
+
     @pytest.mark.asyncio
     async def test_unknown_tool_handling(self, mock_model, mock_agent, mock_function_tool):
         """Test that unknown tools complete the model call without starting a response."""
@@ -1605,7 +1641,7 @@ async def invoke_tool(_ctx: ToolContext[Any], _arguments: str) -> str:
     async def test_function_tool_exception_handling(
         self, mock_model, mock_agent, mock_function_tool
     ):
-        """Test that exceptions in function tools are handled (currently they propagate)"""
+        """Test that function tool exceptions reach both the model and caller."""
         # Set up tool to raise exception
         mock_function_tool.on_invoke_tool.side_effect = ValueError("Tool error")
         mock_agent.get_all_tools.return_value = [mock_function_tool]
@@ -1616,7 +1652,6 @@ async def test_function_tool_exception_handling(
             name="test_function", call_id="call_error", arguments="{}"
         )
 
-        # Currently exceptions propagate (no error handling implemented)
         with pytest.raises(ValueError, match="Tool error"):
             await session._handle_tool_call(tool_call_event)
 
@@ -1626,8 +1661,11 @@ async def test_function_tool_exception_handling(
         assert isinstance(tool_start_event, RealtimeToolStart)
         assert tool_start_event.arguments == "{}"
 
-        # But no tool output should have been sent and no end event queued
-        assert len(mock_model.sent_tool_outputs) == 0
+        assert len(mock_model.sent_tool_outputs) == 1
+        sent_call, sent_output, start_response = mock_model.sent_tool_outputs[0]
+        assert sent_call == tool_call_event
+        assert sent_output == "Tool error"
+        assert start_response is True
 
     @pytest.mark.asyncio
     async def test_tool_call_with_complex_arguments(