fix: validate and coerce function tool argument types (#4612)

_preprocess_args now uses pydantic.TypeAdapter to validate and coerce
all annotated argument types (primitives, enums, containers), not just
Pydantic models. Invalid arguments return a descriptive error to the
LLM so it can self-correct and retry, matching the existing pattern
for missing mandatory args.

- Coerces compatible types (e.g. str "42" -> int 42, str "red" -> Color.RED)
- Returns validation errors to LLM for incompatible types (e.g. "foobar" -> int)
- Existing Pydantic BaseModel handling unchanged (graceful failure)
- Updated all 3 call sites (FunctionTool, CrewAITool, sync tool path)

Author: yuvrajangadsingh

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

@@ -147,7 +147,16 @@ async def _call_tool_in_thread_pool(
     # For sync FunctionTool, call the underlying function directly
     def run_sync_tool():
       if isinstance(tool, FunctionTool):
-        args_to_call = tool._preprocess_args(args)
+        args_to_call, validation_errors = tool._preprocess_args(args)
+        if validation_errors:
+          validation_errors_str = '\n'.join(validation_errors)
+          return {
+              'error': (
+                  f'Invoking `{tool.name}()` failed due to argument'
+                  f' validation errors:\n{validation_errors_str}\nYou could'
+                  ' retry calling this tool with corrected argument types.'
+              )
+          }
         signature = inspect.signature(tool.func)
         valid_params = {param for param in signature.parameters}
         if tool._context_param_name in valid_params:

src/google/adk/tools/crewai_tool.py

@@ -73,8 +73,16 @@ async def run_async(
     duplicates, but is re-added if the function signature explicitly requires it
     as a parameter.
     """
-    # Preprocess arguments (includes Pydantic model conversion)
-    args_to_call = self._preprocess_args(args)
+    # Preprocess arguments (includes Pydantic model conversion and type
+    # validation)
+    args_to_call, validation_errors = self._preprocess_args(args)
+
+    if validation_errors:
+      validation_errors_str = '\n'.join(validation_errors)
+      error_str = f"""Invoking `{self.name}()` failed due to argument validation errors:
+{validation_errors_str}
+You could retry calling this tool with corrected argument types."""
+      return {'error': error_str}
 
     signature = inspect.signature(self.func)
     valid_params = {param for param in signature.parameters}

src/google/adk/tools/function_tool.py

@@ -100,68 +100,101 @@ def _get_declaration(self) -> Optional[types.FunctionDeclaration]:
 
     return function_decl
 
-  def _preprocess_args(self, args: dict[str, Any]) -> dict[str, Any]:
-    """Preprocess and convert function arguments before invocation.
+  def _preprocess_args(
+      self, args: dict[str, Any]
+  ) -> tuple[dict[str, Any], list[str]]:
+    """Preprocess, validate, and convert function arguments before invocation.
 
-    Currently handles:
+    Handles:
     - Converting JSON dictionaries to Pydantic model instances where expected
-
-    Future extensions could include:
-    - Type coercion for other complex types
-    - Validation and sanitization
-    - Custom conversion logic
+    - Validating and coercing primitive types (int, float, str, bool)
+    - Validating enum values
+    - Validating container types (list[int], dict[str, float], etc.)
 
     Args:
       args: Raw arguments from the LLM tool call
 
     Returns:
-      Processed arguments ready for function invocation
+      A tuple of (processed_args, validation_errors). If validation_errors is
+      non-empty, the caller should return the errors to the LLM instead of
+      invoking the function.
     """
     signature = inspect.signature(self.func)
     converted_args = args.copy()
+    validation_errors = []
 
     for param_name, param in signature.parameters.items():
-      if param_name in args and param.annotation != inspect.Parameter.empty:
-        target_type = param.annotation
-
-        # Handle Optional[PydanticModel] types
-        if get_origin(param.annotation) is Union:
-          union_args = get_args(param.annotation)
-          # 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)]
-          if len(non_none_types) == 1:
-            target_type = non_none_types[0]
-
-        # Check if the target type is a Pydantic model
-        if inspect.isclass(target_type) and issubclass(
-            target_type, pydantic.BaseModel
-        ):
-          # Skip conversion if the value is None and the parameter is Optional
-          if args[param_name] is None:
-            continue
-
-          # Convert to Pydantic model if it's not already the correct type
-          if not isinstance(args[param_name], target_type):
-            try:
-              converted_args[param_name] = target_type.model_validate(
-                  args[param_name]
-              )
-            except Exception as e:
-              logger.warning(
-                  f"Failed to convert argument '{param_name}' to Pydantic model"
-                  f' {target_type.__name__}: {e}'
-              )
-              # Keep the original value if conversion fails
-              pass
-
-    return converted_args
+      if param_name not in args or param.annotation is inspect.Parameter.empty:
+        continue
+
+      target_type = param.annotation
+      is_optional = False
+
+      # Handle Optional[T] (Union[T, None]) - unwrap to get inner type
+      if get_origin(param.annotation) is Union:
+        union_args = get_args(param.annotation)
+        non_none_types = [arg for arg in union_args if arg is not type(None)]
+        if len(non_none_types) == 1:
+          target_type = non_none_types[0]
+          is_optional = len(union_args) != len(non_none_types)
+
+      # Pydantic models: keep existing graceful-failure behavior
+      if inspect.isclass(target_type) and issubclass(
+          target_type, pydantic.BaseModel
+      ):
+        if args[param_name] is None:
+          continue
+        if not isinstance(args[param_name], target_type):
+          try:
+            converted_args[param_name] = target_type.model_validate(
+                args[param_name]
+            )
+          except Exception as e:
+            logger.warning(
+                f"Failed to convert argument '{param_name}' to Pydantic model"
+                f' {target_type.__name__}: {e}'
+            )
+        continue
+
+      # Skip None values only for Optional params
+      if args[param_name] is None and is_optional:
+        continue
+
+      # Validate and coerce all other annotated types using TypeAdapter.
+      # This handles primitives (int, float, str, bool), enums, and
+      # container types (list[int], dict[str, float], etc.).
+      try:
+        adapter = pydantic.TypeAdapter(target_type)
+        converted_args[param_name] = adapter.validate_python(
+            args[param_name]
+        )
+      except pydantic.ValidationError as e:
+        validation_errors.append(
+            f"Parameter '{param_name}': expected type '{target_type}',"
+            f' validation error: {e}'
+        )
+      except Exception:
+        # TypeAdapter could not handle this annotation (e.g. a forward
+        # reference string). Skip validation silently.
+        pass
+
+    return converted_args, validation_errors
 
   @override
   async def run_async(
       self, *, args: dict[str, Any], tool_context: ToolContext
   ) -> Any:
-    # Preprocess arguments (includes Pydantic model conversion)
-    args_to_call = self._preprocess_args(args)
+    # Preprocess arguments (includes Pydantic model conversion and type
+    # validation). Validation errors are returned to the LLM so it can
+    # self-correct and retry with proper argument types.
+    args_to_call, validation_errors = self._preprocess_args(args)
+
+    if validation_errors:
+      validation_errors_str = '\n'.join(validation_errors)
+      error_str = f"""Invoking `{self.name}()` failed due to argument validation errors:
+{validation_errors_str}
+You could retry calling this tool with corrected argument types."""
+      return {'error': error_str}
 
     signature = inspect.signature(self.func)
     valid_params = {param for param in signature.parameters}

tests/unittests/tools/test_function_tool_arg_validation.py

@@ -0,0 +1,211 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tests for FunctionTool argument type validation and coercion."""
+
+from enum import Enum
+from typing import Optional
+from unittest.mock import MagicMock
+
+from google.adk.agents.invocation_context import InvocationContext
+from google.adk.sessions.session import Session
+from google.adk.tools.function_tool import FunctionTool
+from google.adk.tools.tool_context import ToolContext
+import pytest
+
+
+class Color(Enum):
+  RED = "red"
+  GREEN = "green"
+  BLUE = "blue"
+
+
+def int_func(num: int) -> int:
+  return num
+
+
+def float_func(val: float) -> float:
+  return val
+
+
+def bool_func(flag: bool) -> bool:
+  return flag
+
+
+def enum_func(color: Color) -> str:
+  return color.value
+
+
+def list_int_func(nums: list[int]) -> list[int]:
+  return nums
+
+
+def optional_int_func(num: Optional[int] = None) -> Optional[int]:
+  return num
+
+
+def multi_param_func(name: str, count: int, flag: bool) -> dict:
+  return {"name": name, "count": count, "flag": flag}
+
+
+# --- _preprocess_args coercion tests ---
+
+
+class TestArgCoercion:
+
+  def test_string_to_int(self):
+    tool = FunctionTool(int_func)
+    args, errors = tool._preprocess_args({"num": "42"})
+    assert errors == []
+    assert args["num"] == 42
+    assert isinstance(args["num"], int)
+
+  def test_float_to_int(self):
+    """Pydantic lax mode truncates float to int."""
+    tool = FunctionTool(int_func)
+    args, errors = tool._preprocess_args({"num": 3.0})
+    assert errors == []
+    assert args["num"] == 3
+    assert isinstance(args["num"], int)
+
+  def test_string_to_float(self):
+    tool = FunctionTool(float_func)
+    args, errors = tool._preprocess_args({"val": "3.14"})
+    assert errors == []
+    assert abs(args["val"] - 3.14) < 1e-9
+
+  def test_int_to_float(self):
+    tool = FunctionTool(float_func)
+    args, errors = tool._preprocess_args({"val": 5})
+    assert errors == []
+    assert args["val"] == 5.0
+    assert isinstance(args["val"], float)
+
+  def test_enum_valid_value(self):
+    tool = FunctionTool(enum_func)
+    args, errors = tool._preprocess_args({"color": "red"})
+    assert errors == []
+    assert args["color"] == Color.RED
+
+  def test_enum_invalid_value(self):
+    tool = FunctionTool(enum_func)
+    args, errors = tool._preprocess_args({"color": "purple"})
+    assert len(errors) == 1
+    assert "color" in errors[0]
+
+  def test_list_int_coercion(self):
+    tool = FunctionTool(list_int_func)
+    args, errors = tool._preprocess_args({"nums": ["1", "2", "3"]})
+    assert errors == []
+    assert args["nums"] == [1, 2, 3]
+
+  def test_optional_none_skipped(self):
+    tool = FunctionTool(optional_int_func)
+    args, errors = tool._preprocess_args({"num": None})
+    assert errors == []
+    assert args["num"] is None
+
+  def test_optional_value_coerced(self):
+    tool = FunctionTool(optional_int_func)
+    args, errors = tool._preprocess_args({"num": "7"})
+    assert errors == []
+    assert args["num"] == 7
+
+  def test_bool_from_int(self):
+    tool = FunctionTool(bool_func)
+    args, errors = tool._preprocess_args({"flag": 1})
+    assert errors == []
+    assert args["flag"] is True
+
+
+# --- _preprocess_args validation error tests ---
+
+
+class TestArgValidationErrors:
+
+  def test_string_for_int_returns_error(self):
+    tool = FunctionTool(int_func)
+    args, errors = tool._preprocess_args({"num": "foobar"})
+    assert len(errors) == 1
+    assert "num" in errors[0]
+
+  def test_none_for_required_int_returns_error(self):
+    """None for a non-Optional int should be flagged."""
+    tool = FunctionTool(int_func)
+    # None passed for a required int param. The Optional unwrap won't
+    # trigger because the annotation is plain `int`, not Optional[int].
+    # TypeAdapter(int).validate_python(None) raises ValidationError.
+    args, errors = tool._preprocess_args({"num": None})
+    assert len(errors) == 1
+    assert "num" in errors[0]
+
+  def test_multiple_param_errors(self):
+    tool = FunctionTool(multi_param_func)
+    args, errors = tool._preprocess_args(
+        {"name": 123, "count": "not_a_number", "flag": "not_a_bool"}
+    )
+    # name: int->str coercion might fail in strict, but lax mode might
+    # handle it. count: "not_a_number"->int will fail. flag: depends on
+    # pydantic behavior.
+    assert any("count" in e for e in errors)
+
+
+# --- run_async integration tests ---
+
+
+def _make_tool_context():
+  tool_context_mock = MagicMock(spec=ToolContext)
+  invocation_context_mock = MagicMock(spec=InvocationContext)
+  session_mock = MagicMock(spec=Session)
+  invocation_context_mock.session = session_mock
+  tool_context_mock.invocation_context = invocation_context_mock
+  return tool_context_mock
+
+
+class TestRunAsyncValidation:
+
+  @pytest.mark.asyncio
+  async def test_invalid_arg_returns_error_to_llm(self):
+    tool = FunctionTool(int_func)
+    result = await tool.run_async(
+        args={"num": "foobar"}, tool_context=_make_tool_context()
+    )
+    assert isinstance(result, dict)
+    assert "error" in result
+    assert "validation error" in result["error"].lower()
+
+  @pytest.mark.asyncio
+  async def test_valid_coercion_invokes_function(self):
+    tool = FunctionTool(int_func)
+    result = await tool.run_async(
+        args={"num": "42"}, tool_context=_make_tool_context()
+    )
+    assert result == 42
+
+  @pytest.mark.asyncio
+  async def test_enum_invalid_returns_error(self):
+    tool = FunctionTool(enum_func)
+    result = await tool.run_async(
+        args={"color": "purple"}, tool_context=_make_tool_context()
+    )
+    assert isinstance(result, dict)
+    assert "error" in result
+
+  @pytest.mark.asyncio
+  async def test_enum_valid_invokes_function(self):
+    tool = FunctionTool(enum_func)
+    result = await tool.run_async(
+        args={"color": "green"}, tool_context=_make_tool_context()
+    )
+    assert result == "green"