fix(tools): handle optional and union pydantic models with string annotations in FunctionTool

src/google/adk/tools/function_tool.py

@@ -25,6 +25,11 @@
 from typing import Optional
 from typing import Union
 
+try:
+  from types import UnionType
+except ImportError:
+  UnionType = None
+
 from google.genai import types
 import pydantic
 from typing_extensions import override
@@ -141,9 +146,10 @@ def _preprocess_args(self, args: dict[str, Any]) -> dict[str, Any]:
         target_type = type_hints.get(param_name, param.annotation)
         if target_type != inspect.Parameter.empty:
 
-          # Handle Optional[PydanticModel] types
-          if get_origin(param.annotation) is Union:
-            union_args = get_args(param.annotation)
+          # Handle Optional/Union types (e.g. Optional[PydanticModel], PydanticModel | None)
+          origin = get_origin(target_type)
+          if origin is Union or (UnionType is not None and origin is UnionType):
+            union_args = get_args(target_type)
             # Find the non-None type in Optional[T] (which is Union[T, None])
             non_none_types = [
                 arg for arg in union_args if arg is not type(None)
@@ -160,12 +166,12 @@ def _preprocess_args(self, args: dict[str, Any]) -> dict[str, Any]:
                 continue
               try:
                 converted_args[param_name] = pydantic.TypeAdapter(
-                    param.annotation
+                    target_type
                 ).validate_python(args[param_name])
               except Exception as e:
                 logger.warning(
                     f"Failed to convert argument '{param_name}' to"
-                    f' {param.annotation}: {e}'
+                    f' {target_type}: {e}'
                 )
               continue
 

tests/unittests/tools/test_function_tool_with_import_annotations.py

@@ -16,6 +16,7 @@
 
 from typing import Any
 from typing import Dict
+from typing import Optional
 
 from google.adk.tools import _automatic_function_calling_util
 from google.adk.tools.function_tool import FunctionTool
@@ -229,3 +230,57 @@ def function_with_list(items: list[ItemModel]) -> int:
   assert processed_args['items'][0].name == 'Burger'
   assert processed_args['items'][0].quantity == 10
   assert processed_args['items'][1].quantity == 5
+
+
+def test_preprocess_args_with_optional_pydantic_model_and_annotations():
+  """Test _preprocess_args converts dict to Optional[Pydantic] model with string annotations."""
+
+  def function_with_optional(item: Optional[ItemModel] = None) -> int:
+    return item.quantity if item else 0
+
+  tool = FunctionTool(function_with_optional)
+  input_args = {'item': {'name': 'Burger', 'quantity': 10}}
+  processed_args = tool._preprocess_args(input_args)
+
+  assert isinstance(processed_args['item'], ItemModel)
+  assert processed_args['item'].name == 'Burger'
+  assert processed_args['item'].quantity == 10
+
+
+def test_preprocess_args_with_pipe_union_pydantic_model_and_annotations():
+  """Test _preprocess_args converts dict to BaseModel | None with string annotations."""
+
+  def function_with_pipe_union(item: ItemModel | None = None) -> int:
+    return item.quantity if item else 0
+
+  tool = FunctionTool(function_with_pipe_union)
+  input_args = {'item': {'name': 'Pizza', 'quantity': 5}}
+  processed_args = tool._preprocess_args(input_args)
+
+  assert isinstance(processed_args['item'], ItemModel)
+  assert processed_args['item'].name == 'Pizza'
+  assert processed_args['item'].quantity == 5
+
+
+def test_preprocess_args_with_optional_list_of_pydantic_models_and_annotations():
+  """Test _preprocess_args converts dicts in Optional[list[BaseModel]] with string annotations."""
+
+  def function_with_optional_list(
+      items: Optional[list[ItemModel]] = None,
+  ) -> int:
+    return sum(item.quantity for item in items) if items else 0
+
+  tool = FunctionTool(function_with_optional_list)
+  input_args = {
+      'items': [
+          {'name': 'Burger', 'quantity': 10},
+          {'name': 'Pizza', 'quantity': 5},
+      ]
+  }
+  processed_args = tool._preprocess_args(input_args)
+
+  assert isinstance(processed_args['items'], list)
+  assert len(processed_args['items']) == 2
+  assert all(isinstance(item, ItemModel) for item in processed_args['items'])
+  assert processed_args['items'][0].quantity == 10
+  assert processed_args['items'][1].quantity == 5