ADK changes

PiperOrigin-RevId: 949756197

Author: google-genai-bot

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

@@ -13,6 +13,7 @@
 # limitations under the License.
 from __future__ import annotations
 
+import json
 import logging
 from typing import Any
 from typing import AsyncGenerator
@@ -26,157 +27,68 @@
 from ...agents.readonly_context import ReadonlyContext
 from ...events.event import Event
 from ...models.llm_request import LlmRequest
-from ...tools.base_tool import BaseTool
 from ...tools.tool_confirmation import ToolConfirmation
-from ...tools.tool_context import ToolContext
 from ._base_llm_processor import BaseLlmRequestProcessor
 from .functions import REQUEST_CONFIRMATION_FUNCTION_CALL_NAME
 
 if TYPE_CHECKING:
   pass
 
 
-logger = logging.getLogger("google_adk." + __name__)
+logger = logging.getLogger('google_adk.' + __name__)
 
 
 def _parse_tool_confirmation(response: dict[str, Any]) -> ToolConfirmation:
-  """Parses ToolConfirmation from a function response dict."""
-  return ToolConfirmation.from_response_dict(response)
+  """Parse ToolConfirmation from a function response dict.
 
+  Handles both the direct dict format and the ADK client's
+  ``{'response': json_string}`` wrapper format.
 
-async def _resolve_confirmation_targets(
-    invocation_context: InvocationContext,
+  """
+  if response and len(response.values()) == 1 and 'response' in response.keys():
+    return ToolConfirmation.model_validate(json.loads(response['response']))
+  return ToolConfirmation.model_validate(response)
+
+
+def _resolve_confirmation_targets(
     events: list[Event],
     confirmation_fc_ids: set[str],
     confirmations_by_fc_id: dict[str, ToolConfirmation],
-    tools_dict: dict[str, BaseTool],
 ) -> tuple[dict[str, ToolConfirmation], dict[str, types.FunctionCall]]:
-  """Find original function calls for confirmed tools and validate them.
+  """Find original function calls for confirmed tools.
 
   Scans events for ``adk_request_confirmation`` function calls whose IDs
   are in *confirmation_fc_ids*, extracts the ``originalFunctionCall`` from
-  their args, validates that they are registered, actually require confirmation,
-  and match the original function calls in history, and maps each confirmation
-  to the original FC ID.
+  their args, and maps each confirmation to the original FC ID.
 
   Args:
-    invocation_context: Current invocation context.
     events: Session events to scan.
     confirmation_fc_ids: IDs of ``adk_request_confirmation`` function calls.
     confirmations_by_fc_id: Mapping of confirmation FC ID ->
       ``ToolConfirmation``.
-    tools_dict: Dictionary of registered tools.
 
   Returns:
     Tuple of ``(tool_confirmation_dict, original_fcs_dict)`` where both
     are keyed by the ORIGINAL function call IDs.
-
-  Raises:
-    ValueError: If validation of any confirmation target fails.
   """
   tool_confirmation_dict: dict[str, ToolConfirmation] = {}
   original_fcs_dict: dict[str, types.FunctionCall] = {}
 
-  history_fcs = {
-      fc.id: (fc, ev)
-      for ev in events
-      for fc in ev.get_function_calls()
-      if fc.id and fc.name != REQUEST_CONFIRMATION_FUNCTION_CALL_NAME
-  }
-  history_fr_events = {
-      fr.id: ev for ev in events for fr in ev.get_function_responses() if fr.id
-  }
-
   for event in events:
     event_function_calls = event.get_function_calls()
     if not event_function_calls:
       continue
 
     for function_call in event_function_calls:
-      if not function_call.id or function_call.id not in confirmation_fc_ids:
+      if function_call.id not in confirmation_fc_ids:
         continue
 
       args = function_call.args
-      if not args or "originalFunctionCall" not in args:
+      if 'originalFunctionCall' not in args:
         continue
       original_function_call = types.FunctionCall(
-          **args["originalFunctionCall"]
-      )
-      if not original_function_call.id:
-        raise ValueError("Original function call ID is missing.")
-      tool_name = original_function_call.name
-      if not tool_name:
-        raise ValueError("Original function call name is missing.")
-
-      # Check 1: Is the tool registered?
-      original_fc_info = history_fcs.get(original_function_call.id)
-      if not original_fc_info:
-        raise ValueError(
-            f"Original function call for ID '{original_function_call.id}' not"
-            " found in session history."
-        )
-      original_fc_in_history, original_fc_event = original_fc_info
-
-      # If this tool call was authored by another agent, skip it to let that
-      # agent's processor handle it.
-      agent = invocation_context.agent
-      if agent and original_fc_event.author != agent.name:
-        continue
-
-      tool = tools_dict.get(tool_name)
-      if not tool:
-        raise ValueError(
-            f"Tool '{original_function_call.name}' is not registered."
-        )
-
-      # Check 2: Does the tool require confirmation for these arguments?
-      # We check if it is either statically required, or if it was dynamically
-      # requested in the session history.
-      temp_tool_context = ToolContext(
-          invocation_context=invocation_context,
-          function_call_id=original_function_call.id,
+          **args['originalFunctionCall']
       )
-      requires_confirmation = await tool.check_require_confirmation(
-          original_function_call.args or {}, temp_tool_context
-      )
-
-      requested_in_history = False
-      if not requires_confirmation:
-        # Search the history for the response event of the original tool call
-        original_response_event = history_fr_events.get(
-            original_function_call.id
-        )
-        if (
-            original_response_event
-            and original_response_event.actions.requested_tool_confirmations
-        ):
-          requested_in_history = (
-              original_function_call.id
-              in original_response_event.actions.requested_tool_confirmations
-          )
-
-      if not requires_confirmation and not requested_in_history:
-        raise ValueError(
-            f"Tool '{original_function_call.name}' does not require"
-            " confirmation."
-        )
-
-      # Check 3: Does the original function call match name and arguments?
-      if original_fc_in_history.name != original_function_call.name:
-        raise ValueError(
-            f"Function call name mismatch for ID '{original_function_call.id}':"
-            f" history has '{original_fc_in_history.name}', confirmation has"
-            f" '{original_function_call.name}'."
-        )
-
-      hist_args = original_fc_in_history.args or {}
-      conf_args = original_function_call.args or {}
-      if hist_args != conf_args:
-        raise ValueError(
-            "Function call arguments mismatch for ID"
-            f" '{original_function_call.id}'."
-        )
-
       tool_confirmation_dict[original_function_call.id] = (
           confirmations_by_fc_id[function_call.id]
       )
@@ -203,9 +115,10 @@ async def run_async(
     # Step 1: Find the last user-authored event and parse confirmation
     # responses from it.
     confirmations_by_fc_id: dict[str, ToolConfirmation] = {}
+    confirmation_event_index = -1
     for k in range(len(events) - 1, -1, -1):
       event = events[k]
-      if not event.author or event.author != "user":
+      if not event.author or event.author != 'user':
         continue
       responses = event.get_function_responses()
       if not responses:
@@ -214,45 +127,29 @@ async def run_async(
       for function_response in responses:
         if function_response.name != REQUEST_CONFIRMATION_FUNCTION_CALL_NAME:
           continue
-        if not function_response.id or function_response.response is None:
-          continue
         confirmations_by_fc_id[function_response.id] = _parse_tool_confirmation(
             function_response.response
         )
+      confirmation_event_index = k
       break
 
     if not confirmations_by_fc_id:
       return
 
-    # Resolve all canonical tools and build tools_dict
-    tools_dict = {}
-    if agent is not None and hasattr(agent, "canonical_tools"):
-      tools_dict = {
-          tool.name: tool
-          for tool in await agent.canonical_tools(
-              ReadonlyContext(invocation_context)
-          )
-      }
-
     # Step 2: Resolve confirmation targets using extracted helper.
     confirmation_fc_ids = set(confirmations_by_fc_id.keys())
     tools_to_resume_with_confirmation, tools_to_resume_with_args = (
-        await _resolve_confirmation_targets(
-            invocation_context,
-            events,
-            confirmation_fc_ids,
-            confirmations_by_fc_id,
-            tools_dict,
+        _resolve_confirmation_targets(
+            events, confirmation_fc_ids, confirmations_by_fc_id
         )
     )
 
     if not tools_to_resume_with_confirmation:
       return
 
     # Step 3: Remove tools that have already been confirmed (dedup).
-    for event in reversed(events):
-      if event.author == "user":
-        break
+    for i in range(len(events) - 1, confirmation_event_index, -1):
+      event = events[i]
       fr_list = event.get_function_responses()
       if not fr_list:
         continue
@@ -270,9 +167,14 @@ async def run_async(
     # Step 4: Re-execute the confirmed tools.
     if function_response_event := await functions.handle_function_call_list_async(
         invocation_context,
-        list(tools_to_resume_with_args.values()),
-        tools_dict,
-        set(tools_to_resume_with_confirmation.keys()),
+        tools_to_resume_with_args.values(),
+        {
+            tool.name: tool
+            for tool in await agent.canonical_tools(
+                ReadonlyContext(invocation_context)
+            )
+        },
+        tools_to_resume_with_confirmation.keys(),
         tools_to_resume_with_confirmation,
     ):
       yield function_response_event

src/google/adk/tools/base_tool.py

@@ -168,12 +168,6 @@ async def process_llm_request(
     # Use the consolidated logic in LlmRequest.append_tools
     llm_request.append_tools([self])
 
-  async def check_require_confirmation(
-      self, args: dict[str, Any], tool_context: ToolContext
-  ) -> bool:
-    """Returns whether the tool requires confirmation for the given args."""
-    return False
-
   @property
   def _api_variant(self) -> GoogleLLMVariant:
     return get_google_llm_variant()

src/google/adk/tools/function_tool.py

@@ -18,7 +18,6 @@
 import logging
 from typing import Any
 from typing import Callable
-from typing import cast
 from typing import get_args
 from typing import get_origin
 from typing import get_type_hints
@@ -195,52 +194,36 @@ def _preprocess_args(self, args: dict[str, Any]) -> dict[str, Any]:
               args[param_name], list
           ):
             item_type = get_list_inner_type(target_type)
-            if item_type is not None:
-              try:
-                converted_args[param_name] = [
-                    item_type.model_validate(item)
-                    if isinstance(item, dict)
-                    else item
-                    for item in args[param_name]
-                ]
-              except Exception as e:
-                logger.warning(
-                    f"Failed to convert argument '{param_name}' to"
-                    f' list[{item_type.__name__}]: {e}'
-                )
-                pass
+            try:
+              converted_args[param_name] = [
+                  item_type.model_validate(item)
+                  if isinstance(item, dict)
+                  else item
+                  for item in args[param_name]
+              ]
+            except Exception as e:
+              logger.warning(
+                  f"Failed to convert argument '{param_name}' to"
+                  f' list[{item_type.__name__}]: {e}'
+              )
+              pass
 
     return converted_args
 
-  def _prepare_invocation_args(
-      self, args: dict[str, Any], tool_context: ToolContext
-  ) -> dict[str, Any]:
-    """Prepare args for function invocation (preprocesses, injects context and filters)."""
-    args_to_call = self._preprocess_args(args)
-    signature = inspect.signature(self.func)
-    valid_params = set(signature.parameters.keys())
-    if self._context_param_name in valid_params:
-      args_to_call[self._context_param_name] = tool_context
-    return {k: v for k, v in args_to_call.items() if k in valid_params}
-
-  @override
-  async def check_require_confirmation(
-      self, args: dict[str, Any], tool_context: ToolContext
-  ) -> bool:
-    if callable(self._require_confirmation):
-      args_to_call = self._prepare_invocation_args(args, tool_context)
-      return cast(
-          bool,
-          await self._invoke_callable(self._require_confirmation, args_to_call),
-      )
-    return bool(self._require_confirmation)
-
   @override
   async def run_async(
       self, *, args: dict[str, Any], tool_context: ToolContext
   ) -> Any:
     # Preprocess arguments (includes Pydantic model conversion)
-    args_to_call = self._prepare_invocation_args(args, tool_context)
+    args_to_call = self._preprocess_args(args)
+
+    signature = inspect.signature(self.func)
+    valid_params = {param for param in signature.parameters}
+    if self._context_param_name in valid_params:
+      args_to_call[self._context_param_name] = tool_context
+
+    # Filter args_to_call to only include valid parameters for the function
+    args_to_call = {k: v for k, v in args_to_call.items() if k in valid_params}
 
     # Before invoking the function, we check for if the list of args passed in
     # has all the mandatory arguments or not.
@@ -259,9 +242,12 @@ async def run_async(
 You could retry calling this tool, but it is IMPORTANT for you to provide all the mandatory parameters."""
       return {'error': error_str}
 
-    require_confirmation = await self.check_require_confirmation(
-        args, tool_context
-    )
+    if isinstance(self._require_confirmation, Callable):
+      require_confirmation = await self._invoke_callable(
+          self._require_confirmation, args_to_call
+      )
+    else:
+      require_confirmation = bool(self._require_confirmation)
 
     if require_confirmation:
       if not tool_context.tool_confirmation: