refactor(types): type the tool and toolset base classes for strict mypy

Co-authored-by: George Weale <gweale@google.com>
PiperOrigin-RevId: 974055066

Author: GWeale

src/google/adk/tools/agent_tool.py

@@ -404,6 +404,7 @@ async def run_async(
       tool_context: ToolContext,
   ) -> Any:
     input_schema = _get_input_schema(self.agent)
+    node_input: object
     if input_schema:
       try:
         node_input = input_schema.model_validate(args)

src/google/adk/tools/authenticated_function_tool.py

@@ -45,7 +45,7 @@ def __init__(
       self,
       *,
       func: Callable[..., Any],
-      auth_config: AuthConfig = None,
+      auth_config: Optional[AuthConfig] = None,
       response_for_auth_required: Optional[Union[dict[str, Any], str]] = None,
   ):
     """Initializes the AuthenticatedFunctionTool.
@@ -65,6 +65,7 @@ def __init__(
     super().__init__(func=func)
     self._ignore_params.append("credential")
 
+    self._credentials_manager: Optional[CredentialManager]
     if auth_config and auth_config.auth_scheme:
       self._credentials_manager = CredentialManager(auth_config=auth_config)
     else:
@@ -98,7 +99,7 @@ async def _run_async_impl(
       *,
       args: dict[str, Any],
       tool_context: ToolContext,
-      credential: AuthCredential,
+      credential: Optional[AuthCredential],
   ) -> Any:
     args_to_call = args.copy()
     signature = inspect.signature(self.func)

src/google/adk/tools/base_authenticated_tool.py

@@ -17,8 +17,7 @@
 from abc import abstractmethod
 import logging
 from typing import Any
-from typing import Optional
-from typing import Union
+from typing import cast
 
 from typing_extensions import override
 
@@ -43,11 +42,11 @@ class BaseAuthenticatedTool(BaseTool):
   def __init__(
       self,
       *,
-      name,
-      description,
-      auth_config: AuthConfig = None,
-      response_for_auth_required: Optional[Union[dict[str, Any], str]] = None,
-  ):
+      name: str,
+      description: str,
+      auth_config: AuthConfig | None = None,
+      response_for_auth_required: dict[str, Any] | str | None = None,
+  ) -> None:
     """
     Args:
       name: The name of the tool.
@@ -67,6 +66,7 @@ def __init__(
         description=description,
     )
     self._auth_config = auth_config
+    self._credentials_manager: CredentialManager | None
 
     if auth_config and auth_config.auth_scheme:
       self._credentials_manager = CredentialManager(auth_config=auth_config)
@@ -94,7 +94,10 @@ async def run_async(
     return await self._run_async_impl(
         args=args,
         tool_context=tool_context,
-        credential=credential,
+        # A tool with no credentials manager runs unauthenticated, so this is
+        # None on that path. Widening the abstract signature instead would
+        # invalidate every subclass that declares the narrower type.
+        credential=cast(AuthCredential, credential),
     )
 
   @abstractmethod

src/google/adk/tools/base_tool.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 from abc import ABC
+from collections.abc import Callable as CallableABC
 import inspect
 import logging
 from typing import Any
@@ -48,6 +49,11 @@
 SelfTool = TypeVar("SelfTool", bound="BaseTool")
 
 
+def _is_callable_annotation(annotation: object) -> bool:
+  """Returns whether a resolved annotation describes a callable."""
+  return annotation is Callable or get_origin(annotation) is CallableABC
+
+
 class BaseTool(ABC):
   """The base class for all tools."""
 
@@ -110,8 +116,8 @@ class BaseTool(ABC):
   def __init__(
       self,
       *,
-      name,
-      description,
+      name: str,
+      description: str,
       is_long_running: bool = False,
       custom_metadata: Optional[dict[str, Any]] = None,
       response_scheduling: Optional[types.FunctionResponseScheduling] = None,
@@ -235,21 +241,23 @@ def from_config(
             and value is not None
         ):
           kwargs[param_name] = param_type.model_validate(value)
-        elif param_type is Callable or get_origin(param_type) is Callable:
+        elif _is_callable_annotation(param_type):
           kwargs[param_name] = config_agent_utils.resolve_fully_qualified_name(
               value
           )
         elif param_type in (list, set, dict):
           kwargs[param_name] = param_type(value)
         elif get_origin(param_type) is list:
           list_args = get_args(param_type)
-          if issubclass(list_args[0], BaseModel):
+          if inspect.isclass(list_args[0]) and issubclass(
+              list_args[0], BaseModel
+          ):
             kwargs[param_name] = [
                 list_args[0].model_validate(item) for item in value
             ]
           elif list_args[0] in (int, str, bool, float):
             kwargs[param_name] = value
-          elif list_args[0] is Callable or get_origin(list_args[0]) is Callable:
+          elif _is_callable_annotation(list_args[0]):
             kwargs[param_name] = [
                 config_agent_utils.resolve_fully_qualified_name(item)
                 for item in value

src/google/adk/tools/base_toolset.py

@@ -17,6 +17,7 @@
 from abc import ABC
 from abc import abstractmethod
 import copy
+from typing import Callable
 from typing import final
 from typing import List
 from typing import Optional
@@ -27,6 +28,8 @@
 from typing import TypeVar
 from typing import Union
 
+from google.genai import types
+
 from ..agents.readonly_context import ReadonlyContext
 from ..auth.auth_tool import AuthConfig
 from .base_tool import BaseTool
@@ -146,10 +149,12 @@ async def get_tools_with_prefix(
       # Also update the function declaration name if the tool has one
       # Use default parameters to capture the current values in the closure
       def _create_prefixed_declaration(
-          original_get_declaration=tool._get_declaration,
-          prefixed_name=prefixed_name,
-      ):
-        def _get_prefixed_declaration():
+          original_get_declaration: Callable[
+              [], Optional[types.FunctionDeclaration]
+          ] = tool._get_declaration,
+          prefixed_name: str = prefixed_name,
+      ) -> Callable[[], Optional[types.FunctionDeclaration]]:
+        def _get_prefixed_declaration() -> Optional[types.FunctionDeclaration]:
           declaration = original_get_declaration()
           if declaration is not None:
             declaration.name = prefixed_name
@@ -158,7 +163,9 @@ def _get_prefixed_declaration():
 
         return _get_prefixed_declaration
 
-      tool_copy._get_declaration = _create_prefixed_declaration()
+      tool_copy._get_declaration = (  # type: ignore[method-assign]
+          _create_prefixed_declaration()
+      )
       prefixed_tools.append(tool_copy)
 
     self._cached_invocation_id = invocation_id

src/google/adk/tools/long_running_tool.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+from typing import Any
 from typing import Callable
 from typing import Optional
 
@@ -40,7 +41,7 @@ class LongRunningFunctionTool(FunctionTool):
     is_long_running: Whether the tool is a long running operation.
   """
 
-  def __init__(self, func: Callable):
+  def __init__(self, func: Callable[..., Any]):
     super().__init__(func)
     self.is_long_running = True
 

tests/unittests/tools/test_base_tool.py

@@ -12,7 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import Any
+from typing import Callable
 from typing import Optional
+from typing import Union
 
 from google.adk.agents.invocation_context import InvocationContext
 from google.adk.agents.sequential_agent import SequentialAgent
@@ -157,6 +160,54 @@ async def run_async(self, **kwargs):
   assert t2._defers_response is True
 
 
+def _sample_handler(value: str) -> str:
+  return value
+
+
+def test_from_config_resolves_parametrized_callable():
+  """A ``Callable[..., ...]`` arg is resolved, not skipped as unsupported."""
+  from google.adk.tools.tool_configs import ToolArgsConfig
+
+  class CallbackTool(BaseTool):
+
+    def __init__(self, handler: Callable[[str], str]):
+      super().__init__(name='callback_tool', description='desc')
+      self.handler = handler
+
+    async def run_async(self, **kwargs):
+      pass
+
+  config = ToolArgsConfig(handler=f'{__name__}._sample_handler')
+  tool = CallbackTool.from_config(config, '')
+
+  assert tool.handler is _sample_handler
+
+
+def test_from_config_skips_list_of_non_class():
+  """Non-class ``list`` element types are skipped, not an issubclass error."""
+  from google.adk.tools.tool_configs import ToolArgsConfig
+
+  class ListTool(BaseTool):
+
+    def __init__(
+        self,
+        unions: Optional[list[Union[int, str]]] = None,
+        anys: Optional[list[Any]] = None,
+    ):
+      super().__init__(name='list_tool', description='desc')
+      self.unions = unions
+      self.anys = anys
+
+    async def run_async(self, **kwargs):
+      pass
+
+  config = ToolArgsConfig(unions=[1, 'two'], anys=[1, 'two'])
+  tool = ListTool.from_config(config, '')
+
+  assert tool.unions is None
+  assert tool.anys is None
+
+
 def test_response_scheduling_defaults_to_none():
   """response_scheduling defaults to None, preserving existing behavior."""