Merge branch 'main' into feat/app-name-override

Author: amitjborate

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

@@ -13,7 +13,6 @@
 # limitations under the License.
 from __future__ import annotations
 
-import json
 import logging
 from typing import Any
 from typing import AsyncGenerator
@@ -27,68 +26,157 @@
 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:
-  """Parse ToolConfirmation from a function response dict.
+  """Parses ToolConfirmation from a function response dict."""
+  return ToolConfirmation.from_response_dict(response)
 
-  Handles both the direct dict format and the ADK client's
-  ``{'response': json_string}`` wrapper format.
 
-  """
-  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(
+async def _resolve_confirmation_targets(
+    invocation_context: InvocationContext,
     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.
+  """Find original function calls for confirmed tools and validate them.
 
   Scans events for ``adk_request_confirmation`` function calls whose IDs
   are in *confirmation_fc_ids*, extracts the ``originalFunctionCall`` from
-  their args, and maps each confirmation to the original FC ID.
+  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.
 
   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 function_call.id not in confirmation_fc_ids:
+      if not function_call.id or function_call.id not in confirmation_fc_ids:
         continue
 
       args = function_call.args
-      if 'originalFunctionCall' not in args:
+      if not args or "originalFunctionCall" not in args:
         continue
       original_function_call = types.FunctionCall(
-          **args['originalFunctionCall']
+          **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,
       )
+      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]
       )
@@ -115,10 +203,9 @@ 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:
@@ -127,29 +214,45 @@ 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 = (
-        _resolve_confirmation_targets(
-            events, confirmation_fc_ids, confirmations_by_fc_id
+        await _resolve_confirmation_targets(
+            invocation_context,
+            events,
+            confirmation_fc_ids,
+            confirmations_by_fc_id,
+            tools_dict,
         )
     )
 
     if not tools_to_resume_with_confirmation:
       return
 
     # Step 3: Remove tools that have already been confirmed (dedup).
-    for i in range(len(events) - 1, confirmation_event_index, -1):
-      event = events[i]
+    for event in reversed(events):
+      if event.author == "user":
+        break
       fr_list = event.get_function_responses()
       if not fr_list:
         continue
@@ -167,14 +270,9 @@ async def run_async(
     # Step 4: Re-execute the confirmed tools.
     if function_response_event := await functions.handle_function_call_list_async(
         invocation_context,
-        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(),
+        list(tools_to_resume_with_args.values()),
+        tools_dict,
+        set(tools_to_resume_with_confirmation.keys()),
         tools_to_resume_with_confirmation,
     ):
       yield function_response_event

src/google/adk/tools/base_tool.py

@@ -168,6 +168,12 @@ 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,6 +18,7 @@
 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
@@ -194,36 +195,52 @@ def _preprocess_args(self, args: dict[str, Any]) -> dict[str, Any]:
               args[param_name], list
           ):
             item_type = get_list_inner_type(target_type)
-            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
+            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
 
     return converted_args
 
-  @override
-  async def run_async(
-      self, *, args: dict[str, Any], tool_context: ToolContext
-  ) -> Any:
-    # Preprocess arguments (includes Pydantic model conversion)
+  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 = {param for param in signature.parameters}
+    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}
 
-    # 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}
+  @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)
 
     # Before invoking the function, we check for if the list of args passed in
     # has all the mandatory arguments or not.
@@ -242,12 +259,9 @@ 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}
 
-    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)
+    require_confirmation = await self.check_require_confirmation(
+        args, tool_context
+    )
 
     if require_confirmation:
       if not tool_context.tool_confirmation: