Fix async tool handling for compatibility with all LLMs.

Author: filipi87

src/pipecat/frames/frames.py

@@ -1918,12 +1918,16 @@ class FunctionCallInProgressFrame(ControlFrame, UninterruptibleFrame):
         tool_call_id: Unique identifier for this function call.
         arguments: Arguments passed to the function.
         cancel_on_interruption: Whether to cancel this call if interrupted.
+        is_async: Whether this function call runs asynchronously. When True,
+            the LLM continues the conversation immediately without waiting for
+            the result. The result is injected later via a developer message.
     """
 
     function_name: str
     tool_call_id: str
     arguments: Any
     cancel_on_interruption: bool = False
+    is_async: bool = False
 
 
 @dataclass

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -1067,16 +1067,25 @@ async def _handle_function_call_result(self, frame: FunctionCallResultFrame):
             )
             return
 
+        in_progress_frame = self._function_calls_in_progress[frame.tool_call_id]
+        is_async = in_progress_frame.is_async if in_progress_frame else False
         del self._function_calls_in_progress[frame.tool_call_id]
 
         properties = frame.properties
 
-        # Update context with the function call result
-        if frame.result:
-            result = json.dumps(frame.result, ensure_ascii=False)
-            self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
+        result = json.dumps(frame.result, ensure_ascii=False) if frame.result else "COMPLETED"
+
+        if is_async:
+            # For async function calls instead of updating the existing IN_PROGRESS tool message we inject
+            # a new developer message so the LLM is notified of the completed result.
+            self._context.add_message(
+                {
+                    "role": "developer",
+                    "content": f"Async function with id '{frame.tool_call_id}' completed with result: {result}",
+                }
+            )
         else:
-            self._update_function_call_result(frame.function_name, frame.tool_call_id, "COMPLETED")
+            self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
 
         run_llm = False
 
@@ -1090,13 +1099,17 @@ async def _handle_function_call_result(self, frame: FunctionCallResultFrame):
             run_llm = await self._maybe_append_image_to_context(image_frame)
 
         # Run inference if the function call result requires it.
-        if frame.result:
+        if frame.result or is_async:
             if properties and properties.run_llm is not None:
                 # If the tool call result has a run_llm property, use it.
                 run_llm = properties.run_llm
             elif frame.run_llm is not None:
                 # If the frame is indicating we should run the LLM, do it.
                 run_llm = frame.run_llm
+            elif is_async:
+                # For async function calls, always run the LLM so it can
+                # respond to the completed result injected as a developer message.
+                run_llm = True
             else:
                 # If this is the last function call in progress, run the LLM.
                 run_llm = not bool(self._function_calls_in_progress)

src/pipecat/services/llm_service.py

@@ -124,13 +124,17 @@ class FunctionCallRegistryItem:
         function_name: The name of the function (None for catch-all handler).
         handler: The handler for processing function call parameters.
         cancel_on_interruption: Whether to cancel the call on interruption.
+        is_async: Whether this function call runs asynchronously. When True,
+            the LLM continues the conversation immediately without waiting for
+            the result. The result is injected later via a developer message.
         timeout_secs: Optional per-tool timeout in seconds. Overrides the global
             ``function_call_timeout_secs`` for this specific function.
     """
 
     function_name: Optional[str]
     handler: FunctionCallHandler | "DirectFunctionWrapper"
     cancel_on_interruption: bool
+    is_async: bool = False
     timeout_secs: Optional[float] = None
 
 
@@ -578,6 +582,7 @@ def register_function(
         handler: Any,
         *,
         cancel_on_interruption: bool = True,
+        is_async: bool = False,
         timeout_secs: Optional[float] = None,
     ):
         """Register a function handler for LLM function calls.
@@ -589,6 +594,10 @@ def register_function(
                 parameter.
             cancel_on_interruption: Whether to cancel this function call when an
                 interruption occurs. Defaults to True.
+            is_async: Whether this function call runs asynchronously. When True,
+                the LLM continues the conversation immediately without waiting for
+                the result. The result is injected later via a developer message.
+                Defaults to False.
             timeout_secs: Optional per-tool timeout in seconds. Overrides the global
                 ``function_call_timeout_secs`` for this specific function. Defaults to
                 None, which uses the global timeout.
@@ -599,6 +608,7 @@ def register_function(
             function_name=function_name,
             handler=handler,
             cancel_on_interruption=cancel_on_interruption,
+            is_async=is_async,
             timeout_secs=timeout_secs,
         )
 
@@ -607,6 +617,7 @@ def register_direct_function(
         handler: DirectFunction,
         *,
         cancel_on_interruption: bool = True,
+        is_async: bool = False,
         timeout_secs: Optional[float] = None,
     ):
         """Register a direct function handler for LLM function calls.
@@ -619,6 +630,10 @@ def register_direct_function(
             handler: The direct function to register. Must follow DirectFunction protocol.
             cancel_on_interruption: Whether to cancel this function call when an
                 interruption occurs. Defaults to True.
+            is_async: Whether this function call runs asynchronously. When True,
+                the LLM continues the conversation immediately without waiting for
+                the result. The result is injected later via a developer message.
+                Defaults to False.
             timeout_secs: Optional per-tool timeout in seconds. Overrides the global
                 ``function_call_timeout_secs`` for this specific function. Defaults to
                 None, which uses the global timeout.
@@ -628,6 +643,7 @@ def register_direct_function(
             function_name=wrapper.name,
             handler=wrapper,
             cancel_on_interruption=cancel_on_interruption,
+            is_async=is_async,
             timeout_secs=timeout_secs,
         )
 
@@ -766,6 +782,7 @@ async def _run_function_call(self, runner_item: FunctionCallRunnerItem):
             tool_call_id=runner_item.tool_call_id,
             arguments=runner_item.arguments,
             cancel_on_interruption=item.cancel_on_interruption,
+            is_async=item.is_async,
         )
 
         timeout_task: Optional[asyncio.Task] = None

src/pipecat/utils/context/llm_context_summarization.py

@@ -389,9 +389,13 @@ def _get_earliest_function_call_not_resolved_in_range(
 
         Scans messages from ``start_idx`` up to (but not including)
         ``summary_end`` to identify tool calls whose responses either don't
-        exist yet or fall in the kept portion of the context (>= summary_end).
+        exist yet, fall in the kept portion of the context (>= summary_end),
+        or are still marked as ``IN_PROGRESS`` (async calls whose results have
+        not yet arrived).
+
         This prevents summarizing tool call requests when their responses would
-        remain in the kept context as orphans, which the OpenAI API rejects.
+        remain in the kept context as orphans, which the OpenAI API rejects,
+        and avoids summarizing async function calls before their results arrive.
 
         Args:
             messages: List of messages to check.
@@ -428,11 +432,13 @@ def _get_earliest_function_call_not_resolved_in_range(
                             if tool_call_id:
                                 pending_tool_calls[tool_call_id] = i
 
-            # Check for tool results
+            # Check for tool results — treat IN_PROGRESS as still pending
+            # (async function calls whose results have not yet arrived).
             if role == "tool":
                 tool_call_id = msg.get("tool_call_id")
                 if tool_call_id and tool_call_id in pending_tool_calls:
-                    pending_tool_calls.pop(tool_call_id)
+                    if msg.get("content") != "IN_PROGRESS":
+                        pending_tool_calls.pop(tool_call_id)
 
         # If we have pending tool calls, return the earliest index
         if pending_tool_calls: