diff --git a/src/google/adk/tools/function_tool.py b/src/google/adk/tools/function_tool.py index 9514c9dd21..f541d7e345 100644 --- a/src/google/adk/tools/function_tool.py +++ b/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 diff --git a/tests/unittests/tools/test_function_tool_with_import_annotations.py b/tests/unittests/tools/test_function_tool_with_import_annotations.py index 0d171628d3..c917bc2ca6 100644 --- a/tests/unittests/tools/test_function_tool_with_import_annotations.py +++ b/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