feat(live): Run non-blocking tools in a background task

This change introduces logic to identify and execute tools that are non-blocking in a separate asyncio task. The function now returns immediately for these tools, and the tool's response is sent back through the LiveRequestQueue when the background task completes.

Co-authored-by: Liang Wu <wuliang@google.com>
PiperOrigin-RevId: 945327698

Author: wuliang229

contributing/samples/live/live_non_blocking_tool_agent/README.md

@@ -0,0 +1,27 @@
+# Live Non-Blocking Tool Agent Sample
+
+## Overview
+
+This sample provides a minimal agent to demonstrate non-blocking tool execution in ADK Live mode (`adk web` / `run_live`).
+
+When a tool declaration is configured with `response_scheduling` set to `WHEN_IDLE`, `SILENT`, or `INTERRUPT`, it indicates to the model that response handling can occur asynchronously.
+
+## Sample Inputs
+
+- `Please start a slow background task for data processing, and then let's keep talking.`
+
+  *Triggers `slow_background_task` which sleeps for 10 seconds. While it runs, continue speaking to the agent.*
+
+## Reproduction Instructions
+
+1. Run the sample via `adk web`:
+   ```bash
+   uv run adk web contributing/samples/live/live_non_blocking_tool_agent
+   ```
+1. Open the ADK web interface and start a Live Session with the agent.
+1. Trigger the tool by saying: *"Please start a slow background task and keep talking with me."*
+1. Continue speaking to the agent while the background task runs in console (`[Tool] Starting slow background task...`).
+
+### Expected Behavior
+
+The model should continue conversing and generating audio/transcription responses immediately while the tool executes in the background. The tool result is delivered later per the `response_scheduling` mode.

contributing/samples/live/live_non_blocking_tool_agent/__init__.py

@@ -0,0 +1,15 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from . import agent

contributing/samples/live/live_non_blocking_tool_agent/agent.py

@@ -0,0 +1,72 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import asyncio
+from typing import Any
+from typing import Dict
+
+from google.adk.agents.llm_agent import Agent
+from google.adk.tools.function_tool import FunctionTool
+from google.genai import types
+
+
+async def slow_background_task(task_description: str) -> Dict[str, Any]:
+  """Performs a long-running background computation or data retrieval.
+
+  Args:
+    task_description: Description of the task to run in the background.
+
+  Returns:
+    A dictionary containing the completion status and task result.
+  """
+  print(f"[Tool] Starting slow background task: {task_description}")
+  # Simulate a 10-second non-blocking background operation
+  await asyncio.sleep(10)
+  print(f"[Tool] Completed slow background task: {task_description}")
+  return {
+      "status": "completed",
+      "task": task_description,
+      "result": "Background task finished successfully after 10 seconds.",
+  }
+
+
+# Create a FunctionTool wrapping the long-running async function
+non_blocking_tool = FunctionTool(slow_background_task)
+
+# Configure response_scheduling to indicate non-blocking behavior for Live mode.
+# Options: WHEN_IDLE, SILENT, or INTERRUPT.
+non_blocking_tool.response_scheduling = (
+    types.FunctionResponseScheduling.WHEN_IDLE
+)
+
+
+root_agent = Agent(
+    model="gemini-live-2.5-flash-native-audio",
+    name="non_blocking_tool_agent",
+    description=(
+        "Agent demonstrating non-blocking tool execution in ADK Live mode."
+    ),
+    instruction="""
+      You are a helpful assistant for testing live mode non-blocking tool execution.
+
+      You have access to a tool `slow_background_task` which is configured with
+      NON_BLOCKING response scheduling (WHEN_IDLE).
+
+      When the user asks you to run a long-running or background task, call the `slow_background_task` tool.
+      Inform the user that the task has started and continue conversing with them normally while the task runs in the background.
+    """,
+    tools=[
+        non_blocking_tool,
+    ],
+)

src/google/adk/agents/invocation_context.py

@@ -210,6 +210,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/functions.py

@@ -678,8 +678,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 = [
@@ -689,7 +689,7 @@ async def handle_function_calls_live(
               function_call,
               tools_dict,
               agent,
-              streaming_lock,
+              active_tools_lock,
           )
       )
       for function_call in function_calls
@@ -737,7 +737,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."""
 
@@ -800,7 +800,16 @@ async def _run_on_tool_error_callbacks(
       )
     raise tool_error
 
-  async def _run_with_trace():
+  async def _run_with_trace() -> Optional[Event]:
+    """Executes the tool with full lifecycle management and telemetry.
+
+    This function orchestrates the tool execution pipeline, including:
+    1. Running plugin and canonical before-tool callbacks.
+    2. Executing the actual tool logic.
+    3. Running plugin and canonical after-tool callbacks.
+    4. Detecting error types for telemetry.
+    5. Building the final FunctionResponse Event to be returned.
+    """
     nonlocal function_args, detected_error_type
 
     # Do not use "args" as the variable name, because it is a reserved keyword
@@ -837,7 +846,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(
@@ -904,12 +913,59 @@ async def _run_with_trace():
     )
     return function_response_event
 
-  async with _instrumentation.record_tool_execution(
-      tool, agent, function_args, invocation_context=invocation_context
-  ) as tel_ctx:
-    tel_ctx.function_response_event = await _run_with_trace()
-    tel_ctx.error_type = detected_error_type
-    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, invocation_context=invocation_context
+        ) as tel_ctx:
+          function_response_event = await _run_with_trace()
+          tel_ctx.function_response_event = function_response_event
+          tel_ctx.error_type = detected_error_type
+
+          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:
+        logger.exception('Error running non-blocking tool %s', tool.name)
+      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, invocation_context=invocation_context
+    ) as tel_ctx:
+      tel_ctx.function_response_event = await _run_with_trace()
+      tel_ctx.error_type = detected_error_type
+      return tel_ctx.function_response_event
 
 
 async def _process_function_live_helper(
@@ -918,7 +974,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
@@ -928,7 +984,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
@@ -961,7 +1017,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
@@ -1005,7 +1061,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 = {}