feat(live): Support asynchronous non-blocking tools

src/google/adk/agents/invocation_context.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from typing import Any
 from typing import cast
 from typing import Optional
@@ -185,6 +186,9 @@ class InvocationContext(BaseModel):
   active_streaming_tools: Optional[dict[str, ActiveStreamingTool]] = None
   """The running streaming tools of this invocation."""
 
+  active_non_blocking_tool_tasks: Optional[dict[str, asyncio.Task[Any]]] = None
+  """The running non-blocking tool tasks of this invocation (Live only)."""
+
   transcription_cache: Optional[list[TranscriptionEntry]] = None
   """Caches necessary data, audio or contents, that are needed by transcription."""
 

src/google/adk/flows/llm_flows/base_llm_flow.py

@@ -466,6 +466,29 @@ async def _process_agent_tools(
           tool_context=tool_context, llm_request=llm_request
       )
 
+  if invocation_context.live_request_queue is not None:
+    _mark_live_async_tools_non_blocking(llm_request)
+
+
+def _mark_live_async_tools_non_blocking(llm_request: LlmRequest) -> None:
+  """Marks live streaming and response-scheduling tools as NON_BLOCKING.
+
+  These tools emit asynchronous FunctionResponses, which the Live API only
+  accepts for NON_BLOCKING declarations.
+  """
+  if not llm_request.config.tools:
+    return
+  for gemini_tool in llm_request.config.tools:
+    for declaration in gemini_tool.function_declarations or []:
+      tool = (llm_request.tools_dict or {}).get(declaration.name)
+      if not tool:
+        continue
+      is_streaming = hasattr(tool, 'func') and inspect.isasyncgenfunction(
+          tool.func
+      )
+      if is_streaming or tool.response_scheduling is not None:
+        declaration.behavior = types.Behavior.NON_BLOCKING
+
 
 class BaseLlmFlow(ABC):
   """A basic flow that calls the LLM in a loop until a final response is generated.
@@ -1141,23 +1164,26 @@ async def _postprocess_live(
 
     # Handles function calls.
     if model_response_event.get_function_calls():
-      function_response_event = await functions.handle_function_calls_live(
+      # handle_function_calls_live returns None when every call is deferred
+      # (e.g. all long-running or non-blocking), so guard before yielding to
+      # avoid emitting a None event into the live stream.
+      if function_response_event := await functions.handle_function_calls_live(
           invocation_context, model_response_event, llm_request.tools_dict
-      )
-      # Always yield the function response event first
-      yield function_response_event
-
-      # Check if this is a set_model_response function response
-      if json_response := _output_schema_processor.get_structured_model_response(
-          function_response_event
       ):
-        # Create and yield a final model response event
-        final_event = (
-            _output_schema_processor.create_final_model_response_event(
-                invocation_context, json_response
-            )
-        )
-        yield final_event
+        # Always yield the function response event first
+        yield function_response_event
+
+        # Check if this is a set_model_response function response
+        if json_response := _output_schema_processor.get_structured_model_response(
+            function_response_event
+        ):
+          # Create and yield a final model response event
+          final_event = (
+              _output_schema_processor.create_final_model_response_event(
+                  invocation_context, json_response
+              )
+          )
+          yield final_event
 
   async def _postprocess_run_processors_async(
       self, invocation_context: InvocationContext, llm_response: LlmResponse

src/google/adk/flows/llm_flows/functions.py

@@ -504,7 +504,7 @@ async def _run_on_tool_error_callbacks(
     else:
       raise tool_error
 
-  async def _run_with_trace():
+  async def _run_with_trace() -> Optional[Event]:
     nonlocal function_args
 
     # Step 1: Check if plugin before_tool_callback overrides the function
@@ -613,8 +613,8 @@ async def handle_function_calls_live(
   if not function_calls:
     return None
 
-  # Create async lock for active_streaming_tools modifications
-  streaming_lock = asyncio.Lock()
+  # Create async lock for active_streaming_tools and active_non_blocking_tool_tasks modifications
+  active_tools_lock = asyncio.Lock()
 
   # Create tasks for parallel execution
   tasks = [
@@ -624,7 +624,7 @@ async def handle_function_calls_live(
               function_call,
               tools_dict,
               agent,
-              streaming_lock,
+              active_tools_lock,
           )
       )
       for function_call in function_calls
@@ -671,7 +671,7 @@ async def _execute_single_function_call_live(
     function_call: types.FunctionCall,
     tools_dict: dict[str, BaseTool],
     agent: LlmAgent,
-    streaming_lock: asyncio.Lock,
+    active_tools_lock: asyncio.Lock,
 ) -> Optional[Event]:
   """Execute a single function call for live mode with thread safety."""
 
@@ -733,7 +733,7 @@ async def _run_on_tool_error_callbacks(
       )
     raise tool_error
 
-  async def _run_with_trace():
+  async def _run_with_trace() -> Optional[Event]:
     nonlocal function_args
 
     # Do not use "args" as the variable name, because it is a reserved keyword
@@ -770,7 +770,7 @@ async def _run_with_trace():
             function_call,
             function_args,
             invocation_context,
-            streaming_lock,
+            active_tools_lock,
         )
       except Exception as tool_error:
         error_response = await _run_on_tool_error_callbacks(
@@ -830,11 +830,62 @@ async def _run_with_trace():
     )
     return function_response_event
 
-  async with _instrumentation.record_tool_execution(
-      tool, agent, function_args
-  ) as tel_ctx:
-    tel_ctx.function_response_event = await _run_with_trace()
-    return tel_ctx.function_response_event
+  is_streaming = hasattr(tool, 'func') and inspect.isasyncgenfunction(tool.func)
+  is_non_blocking = not is_streaming and tool.response_scheduling is not None
+  if is_non_blocking:
+    task_key = f'{tool.name}_{function_call.id}'
+
+    async def _background_task() -> None:
+      try:
+        async with _instrumentation.record_tool_execution(
+            tool, agent, function_args
+        ) as tel_ctx:
+          function_response_event = await _run_with_trace()
+          tel_ctx.function_response_event = function_response_event
+
+          if function_response_event:
+            if (
+                invocation_context.session_service
+                and invocation_context.session
+            ):
+              await invocation_context.session_service.append_event(
+                  session=invocation_context.session,
+                  event=function_response_event,
+              )
+            if (
+                invocation_context.live_request_queue
+                and function_response_event.content
+            ):
+              invocation_context.live_request_queue.send_content(
+                  function_response_event.content
+              )
+      except Exception as e:
+        logger.error(
+            'Error executing non-blocking tool %s in background: %s',
+            tool.name,
+            e,
+            exc_info=True,
+        )
+      finally:
+        async with active_tools_lock:
+          if (
+              invocation_context.active_non_blocking_tool_tasks
+              and task_key in invocation_context.active_non_blocking_tool_tasks
+          ):
+            del invocation_context.active_non_blocking_tool_tasks[task_key]
+
+    task = asyncio.create_task(_background_task())
+    async with active_tools_lock:
+      if invocation_context.active_non_blocking_tool_tasks is None:
+        invocation_context.active_non_blocking_tool_tasks = {}
+      invocation_context.active_non_blocking_tool_tasks[task_key] = task
+    return None
+  else:
+    async with _instrumentation.record_tool_execution(
+        tool, agent, function_args
+    ) as tel_ctx:
+      tel_ctx.function_response_event = await _run_with_trace()
+      return tel_ctx.function_response_event
 
 
 async def _process_function_live_helper(
@@ -843,7 +894,7 @@ async def _process_function_live_helper(
     function_call,
     function_args,
     invocation_context,
-    streaming_lock: asyncio.Lock,
+    active_tools_lock: asyncio.Lock,
 ):
   function_response = None
   # Check if this is a stop_streaming function call
@@ -853,7 +904,7 @@ async def _process_function_live_helper(
   ):
     function_name = function_args['function_name']
     # Thread-safe access to active_streaming_tools
-    async with streaming_lock:
+    async with active_tools_lock:
       active_tasks = invocation_context.active_streaming_tools
       if (
           active_tasks
@@ -886,7 +937,7 @@ async def _process_function_live_helper(
           }
       if not function_response:
         # Clean up the reference under lock
-        async with streaming_lock:
+        async with active_tools_lock:
           if (
               invocation_context.active_streaming_tools
               and function_name in invocation_context.active_streaming_tools
@@ -917,13 +968,8 @@ async def run_tool_and_update_queue(tool, function_args, tool_context):
             )
         ) as agen:
           async for result in agen:
-            updated_content = types.Content(
-                role='user',
-                parts=[
-                    types.Part.from_text(
-                        text=f'Function {tool.name} returned: {result}'
-                    )
-                ],
+            updated_content = _build_function_response_content(
+                tool, result, tool_context.function_call_id
             )
             invocation_context.live_request_queue.send_content(updated_content)
       except asyncio.CancelledError:
@@ -933,7 +979,7 @@ async def run_tool_and_update_queue(tool, function_args, tool_context):
         run_tool_and_update_queue(tool, function_args, tool_context)
     )
 
-    async with streaming_lock:
+    async with active_tools_lock:
 
       if invocation_context.active_streaming_tools is None:
         invocation_context.active_streaming_tools = {}
@@ -1112,26 +1158,17 @@ def __build_response_event(
     tool_context: ToolContext,
     invocation_context: InvocationContext,
 ) -> Event:
-  # Specs requires the result to be a dict.
-  if not isinstance(function_result, dict):
-    function_result = {'result': function_result}
-
   function_response_parts = None
   if isinstance(tool, ComputerUseTool):
     function_response_parts = _try_decode_computer_use_image(
         tool, function_result
     )
 
-  part_function_response = types.Part.from_function_response(
-      name=tool.name,
-      response=function_result,
-      parts=function_response_parts,
-  )
-  part_function_response.function_response.id = tool_context.function_call_id
-
-  content = types.Content(
-      role='user',
-      parts=[part_function_response],
+  content = _build_function_response_content(
+      tool,
+      function_result,
+      tool_context.function_call_id,
+      function_response_parts,
   )
 
   function_response_event = Event(
@@ -1145,6 +1182,31 @@ def __build_response_event(
   return function_response_event
 
 
+def _build_function_response_content(
+    tool: BaseTool,
+    function_result: object,
+    function_call_id: Optional[str],
+    function_response_parts: Optional[list[types.FunctionResponsePart]] = None,
+) -> types.Content:
+  """Builds the content carrying a tool result as a FunctionResponse."""
+  # Specs requires the result to be a dict.
+  if not isinstance(function_result, dict):
+    function_result = {'result': function_result}
+
+  part_function_response = types.Part.from_function_response(
+      name=tool.name,
+      response=function_result,
+      parts=function_response_parts,
+  )
+  part_function_response.function_response.id = function_call_id
+  if tool.response_scheduling is not None:
+    part_function_response.function_response.scheduling = (
+        tool.response_scheduling
+    )
+
+  return types.Content(role='user', parts=[part_function_response])
+
+
 def deep_merge_dicts(d1: dict, d2: dict) -> dict:
   """Recursively merges d2 into d1."""
   for key, value in d2.items():

src/google/adk/tools/base_tool.py

@@ -65,18 +65,32 @@ class BaseTool(ABC):
   NOTE: the entire dict must be JSON serializable.
   """
 
+  response_scheduling: Optional[types.FunctionResponseScheduling] = None
+  """Controls when the model reacts to the tool's response (Live API only).
+
+  Applied to the emitted ``FunctionResponse`` for asynchronous function calling:
+    - ``SILENT``: feeds the response back without triggering a model turn.
+    - ``WHEN_IDLE``: defers the reaction until the model is idle.
+    - ``INTERRUPT``: reacts immediately.
+
+  Ignored by models that don't support asynchronous function calling. ``None``
+  preserves the default behavior.
+  """
+
   def __init__(
       self,
       *,
       name,
       description,
       is_long_running: bool = False,
       custom_metadata: Optional[dict[str, Any]] = None,
+      response_scheduling: Optional[types.FunctionResponseScheduling] = None,
   ):
     self.name = name
     self.description = description
     self.is_long_running = is_long_running
     self.custom_metadata = custom_metadata
+    self.response_scheduling = response_scheduling
 
   def _get_declaration(self) -> Optional[types.FunctionDeclaration]:
     """Gets the OpenAPI specification of this tool in the form of a FunctionDeclaration.