fix: #1942 Enable async tool calling in Realtime sessions

Author: seratch

examples/realtime/app/agent.py

@@ -1,3 +1,5 @@
+import asyncio
+
 from agents import function_tool
 from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX
 from agents.realtime import RealtimeAgent, realtime_handoff
@@ -13,20 +15,26 @@
     name_override="faq_lookup_tool", description_override="Lookup frequently asked questions."
 )
 async def faq_lookup_tool(question: str) -> str:
-    if "bag" in question or "baggage" in question:
+    print("faq_lookup_tool called with question:", question)
+
+    # Simulate a slow API call
+    await asyncio.sleep(3)
+
+    q = question.lower()
+    if "wifi" in q or "wi-fi" in q:
+        return "We have free wifi on the plane, join Airline-Wifi"
+    elif "bag" in q or "baggage" in q:
         return (
             "You are allowed to bring one bag on the plane. "
             "It must be under 50 pounds and 22 inches x 14 inches x 9 inches."
         )
-    elif "seats" in question or "plane" in question:
+    elif "seats" in q or "plane" in q:
         return (
             "There are 120 seats on the plane. "
             "There are 22 business class seats and 98 economy seats. "
             "Exit rows are rows 4 and 16. "
             "Rows 5-8 are Economy Plus, with extra legroom. "
         )
-    elif "wifi" in question:
-        return "We have free wifi on the plane, join Airline-Wifi"
     return "I'm sorry, I don't know the answer to that question."
 
 

examples/realtime/app/server.py

@@ -12,7 +12,7 @@
 from typing_extensions import assert_never
 
 from agents.realtime import RealtimeRunner, RealtimeSession, RealtimeSessionEvent
-from agents.realtime.config import RealtimeUserInputMessage
+from agents.realtime.config import RealtimeRunConfig, RealtimeUserInputMessage
 from agents.realtime.items import RealtimeItem
 from agents.realtime.model import RealtimeModelConfig
 from agents.realtime.model_inputs import RealtimeModelSendRawMessage
@@ -47,6 +47,9 @@ async def connect(self, websocket: WebSocket, session_id: str):
 
         agent = get_starting_agent()
         runner = RealtimeRunner(agent)
+        # If you want to customize the runner behavior, you can pass options:
+        # runner_config = RealtimeRunConfig(async_tool_calls=False)
+        # runner = RealtimeRunner(agent, config=runner_config)
         model_config: RealtimeModelConfig = {
             "initial_model_settings": {
                 "turn_detection": {

src/agents/realtime/config.py

@@ -184,6 +184,9 @@ class RealtimeRunConfig(TypedDict):
     tracing_disabled: NotRequired[bool]
     """Whether tracing is disabled for this run."""
 
+    async_tool_calls: NotRequired[bool]
+    """Whether function tool calls should run asynchronously. Defaults to True."""
+
     # TODO (rm) Add history audio storage config
 
 

src/agents/realtime/session.py

@@ -123,6 +123,8 @@ def __init__(
         )
 
         self._guardrail_tasks: set[asyncio.Task[Any]] = set()
+        self._tool_call_tasks: set[asyncio.Task[Any]] = set()
+        self._async_tool_calls: bool = bool(self._run_config.get("async_tool_calls", True))
 
     @property
     def model(self) -> RealtimeModel:
@@ -216,7 +218,10 @@ async def on_event(self, event: RealtimeModelEvent) -> None:
         if event.type == "error":
             await self._put_event(RealtimeError(info=self._event_info, error=event.error))
         elif event.type == "function_call":
-            await self._handle_tool_call(event)
+            if self._async_tool_calls:
+                self._enqueue_tool_call_task(event)
+            else:
+                await self._handle_tool_call(event)
         elif event.type == "audio":
             await self._put_event(
                 RealtimeAudio(
@@ -752,10 +757,47 @@ def _cleanup_guardrail_tasks(self) -> None:
                 task.cancel()
         self._guardrail_tasks.clear()
 
+    def _enqueue_tool_call_task(self, event: RealtimeModelToolCallEvent) -> None:
+        """Run tool calls in the background to avoid blocking realtime transport."""
+        task = asyncio.create_task(self._handle_tool_call(event))
+        self._tool_call_tasks.add(task)
+        task.add_done_callback(self._on_tool_call_task_done)
+
+    def _on_tool_call_task_done(self, task: asyncio.Task[Any]) -> None:
+        self._tool_call_tasks.discard(task)
+
+        if task.cancelled():
+            return
+
+        exception = task.exception()
+        if exception is None:
+            return
+
+        logger.exception("Realtime tool call task failed", exc_info=exception)
+
+        if self._stored_exception is None:
+            self._stored_exception = exception
+
+        asyncio.create_task(
+            self._put_event(
+                RealtimeError(
+                    info=self._event_info,
+                    error={"message": f"Tool call task failed: {exception}"},
+                )
+            )
+        )
+
+    def _cleanup_tool_call_tasks(self) -> None:
+        for task in self._tool_call_tasks:
+            if not task.done():
+                task.cancel()
+        self._tool_call_tasks.clear()
+
     async def _cleanup(self) -> None:
         """Clean up all resources and mark session as closed."""
         # Cancel and cleanup guardrail tasks
         self._cleanup_guardrail_tasks()
+        self._cleanup_tool_call_tasks()
 
         # Remove ourselves as a listener
         self._model.remove_listener(self)

tests/realtime/test_session.py

@@ -561,8 +561,13 @@ async def test_ignored_events_only_generate_raw_events(self, mock_model, mock_ag
 
     @pytest.mark.asyncio
     async def test_function_call_event_triggers_tool_handling(self, mock_model, mock_agent):
-        """Test that function_call events trigger tool call handling"""
-        session = RealtimeSession(mock_model, mock_agent, None)
+        """Test that function_call events trigger tool call handling synchronously when disabled"""
+        session = RealtimeSession(
+            mock_model,
+            mock_agent,
+            None,
+            run_config={"async_tool_calls": False},
+        )
 
         # Create function call event
         function_call_event = RealtimeModelToolCallEvent(
@@ -586,6 +591,34 @@ async def test_function_call_event_triggers_tool_handling(self, mock_model, mock
             assert isinstance(raw_event, RealtimeRawModelEvent)
             assert raw_event.data == function_call_event
 
+    @pytest.mark.asyncio
+    async def test_function_call_event_runs_async_by_default(self, mock_model, mock_agent):
+        """Function call handling should be scheduled asynchronously by default"""
+        session = RealtimeSession(mock_model, mock_agent, None)
+
+        function_call_event = RealtimeModelToolCallEvent(
+            name="test_function",
+            call_id="call_async",
+            arguments='{"param": "value"}',
+        )
+
+        with pytest.MonkeyPatch().context() as m:
+            handle_tool_call_mock = AsyncMock()
+            m.setattr(session, "_handle_tool_call", handle_tool_call_mock)
+
+            await session.on_event(function_call_event)
+
+            # Let the background task run
+            await asyncio.sleep(0)
+
+            handle_tool_call_mock.assert_awaited_once_with(function_call_event)
+
+        # Raw event still enqueued
+        assert session._event_queue.qsize() == 1
+        raw_event = await session._event_queue.get()
+        assert isinstance(raw_event, RealtimeRawModelEvent)
+        assert raw_event.data == function_call_event
+
 
 class TestHistoryManagement:
     """Test suite for history management and audio transcription in