1  """
  2  Test-specific ADK Tools.
  3  """
  4  
  5  import logging
  6  import asyncio
  7  from typing import Dict, Optional, Any
  8  
  9  from google.adk.tools import ToolContext
 10  
 11  from google.genai import types as adk_types
 12  from .tool_definition import BuiltinTool
 13  from .registry import tool_registry
 14  
 15  log = logging.getLogger(__name__)
 16  
 17  async def time_delay(
 18      seconds: float,
 19      tool_context: ToolContext = None,
 20      tool_config: Optional[Dict[str, Any]] = None,
 21  ) -> Dict[str, any]:
 22      """
 23      Pauses execution for a specified number of seconds.
 24      Useful for testing timeouts and asynchronous behavior.
 25  
 26      Args:
 27          seconds: The duration of the delay in seconds.
 28          tool_context: The context provided by the ADK framework.
 29  
 30      Returns:
 31          A dictionary indicating the status and duration of the delay.
 32      """
 33      log_identifier = "[TestTool:time_delay]"
 34      log.info("%s Requesting delay for %.2f seconds.", log_identifier, seconds)
 35  
 36      if not tool_context:
 37          log.warning("%s ToolContext is missing.", log_identifier)
 38  
 39      try:
 40          if not isinstance(seconds, (int, float)) or seconds < 0:
 41              log.error(
 42                  "%s Invalid 'seconds' argument: %s. Must be a non-negative number.",
 43                  log_identifier,
 44                  seconds,
 45              )
 46              return {
 47                  "status": "error",
 48                  "message": f"Invalid 'seconds' argument: {seconds}. Must be a non-negative number.",
 49              }
 50  
 51          await asyncio.sleep(float(seconds))
 52          log.info("%s Successfully delayed for %.2f seconds.", log_identifier, seconds)
 53          return {
 54              "status": "success",
 55              "message": f"Delayed for {seconds} seconds.",
 56              "delayed_seconds": seconds,
 57          }
 58      except Exception as e:
 59          log.exception("%s Error during time_delay: %s", log_identifier, e)
 60          return {
 61              "status": "error",
 62              "message": f"Error during delay: {e}",
 63              "requested_seconds": seconds,
 64          }
 65  
 66  
 67  time_delay_tool_def = BuiltinTool(
 68      name="time_delay",
 69      implementation=time_delay,
 70      description="Pauses execution for a specified number of seconds. Useful for testing timeouts and asynchronous behavior.",
 71      category="test",
 72      required_scopes=["tool:test:delay"],
 73      parameters=adk_types.Schema(
 74          type=adk_types.Type.OBJECT,
 75          properties={
 76              "seconds": adk_types.Schema(
 77                  type=adk_types.Type.NUMBER,
 78                  description="The duration of the delay in seconds.",
 79              ),
 80          },
 81          required=["seconds"],
 82      ),
 83      examples=[],
 84  )
 85  
 86  
 87  def always_fail_tool() -> dict:
 88      """This tool is designed to always raise an exception for testing error handling."""
 89      raise ValueError("This tool is designed to fail.")
 90  
 91  
 92  def dangling_tool_call_test_tool() -> None:
 93      """
 94      This tool is designed to return None, which simulates a silent failure
 95      where the ADK does not create a function_response, leaving a dangling
 96      function_call in the history. This is used to test the proactive
 97      history repair callback ("the suspenders").
 98      """
 99      log.info(
100          "[TestTool:dangling_tool_call] Executing and returning None to create a dangling tool call."
101      )
102      return None
103  
104  
105  always_fail_tool_def = BuiltinTool(
106      name="always_fail_tool",
107      implementation=always_fail_tool,
108      description="This tool is designed to always raise an exception for testing error handling.",
109      category="test",
110      required_scopes=["tool:test:fail"],
111      parameters=adk_types.Schema(
112          type=adk_types.Type.OBJECT,
113          properties={},
114          required=[],
115      ),
116      examples=[],
117  )
118  
119  dangling_tool_call_test_tool_def = BuiltinTool(
120      name="dangling_tool_call_test_tool",
121      implementation=dangling_tool_call_test_tool,
122      description="A test tool that returns None to create a dangling tool call in the history.",
123      category="test",
124      required_scopes=["tool:test:dangle"],
125      parameters=adk_types.Schema(
126          type=adk_types.Type.OBJECT,
127          properties={},
128          required=[],
129      ),
130      examples=[],
131  )
132  
133  
134  tool_registry.register(time_delay_tool_def)
135  tool_registry.register(always_fail_tool_def)
136  tool_registry.register(dangling_tool_call_test_tool_def)