Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/google/adk/tools/_function_tool_declarations.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ def _get_function_fields(

# Get type hints with forward reference resolution
try:
type_hints = get_type_hints(func)
type_hints = get_type_hints(func, include_extras=True)
except TypeError:
# Can happen with mock objects or complex annotations
type_hints = {}
Expand Down Expand Up @@ -160,7 +160,7 @@ def _build_response_json_schema(
# Handle string annotations (forward references)
if isinstance(return_annotation, str):
try:
type_hints = get_type_hints(func)
type_hints = get_type_hints(func, include_extras=True)
return_annotation = type_hints.get('return', return_annotation)
except TypeError:
pass
Expand Down
12 changes: 9 additions & 3 deletions src/google/adk/tools/function_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import inspect
import logging
from types import UnionType
from typing import Annotated
from typing import Any
from typing import Awaitable
from typing import Callable
Expand Down Expand Up @@ -189,15 +190,20 @@ def _preprocess_args(self, args: dict[str, Any]) -> dict[str, Any]:
for param_name, param in signature.parameters.items():
if param_name in args:
target_type = type_hints.get(param_name, param.annotation)
if get_origin(target_type) is Annotated:
# Strip Annotated to get actual type
target_type = get_args(target_type)[0]
if target_type != inspect.Parameter.empty:

# Handle Optional/Union types (e.g. Optional[PydanticModel], PydanticModel | None)
origin = get_origin(target_type)
if origin is Union or origin is UnionType:
union_args = get_args(target_type)
# Find the non-None type in Optional[T] (which is Union[T, None])
# Find the non-None type in Optional[T] (which is Union[T, None]).
# Handle Optional[Annotated(...)]
non_none_types = [
arg for arg in union_args if arg is not type(None)
get_args(arg)[0] if get_origin(arg) is Annotated else arg
for arg in union_args
if arg is not type(None)
]
if len(non_none_types) == 1:
target_type = non_none_types[0]
Expand Down
34 changes: 34 additions & 0 deletions tests/unittests/tools/test_build_function_declaration.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
# limitations under the License.

from enum import Enum
from typing import Annotated
from typing import Any

from google.adk.features import FeatureName
Expand All @@ -25,6 +26,7 @@
# TODO: crewai requires python 3.10 as minimum
# from crewai_tools import FileReadTool
from pydantic import BaseModel
from pydantic import Field
import pytest


Expand Down Expand Up @@ -648,6 +650,22 @@ def __call__(self, a: int, b: int):
assert function_decl.name == 'Calc'
assert function_decl.response is not None

def test_annotated_field_metadata_preserved(self):
"""Test Annotated[T, Field(...)] metadata reaches the schema."""

def legacy_annotated_function(
count: Annotated[int, Field(description='How many widgets', ge=1)],
) -> str:
return str(count)

function_decl = _automatic_function_calling_util.build_function_declaration(
func=legacy_annotated_function
)

count_schema = function_decl.parameters.properties['count']
assert count_schema.description == 'How many widgets'
assert count_schema.minimum == 1


class TestBuildFunctionDeclarationWithJsonSchema:
"""Tests for build_function_declaration when JSON_SCHEMA_FOR_FUNC_DECL is enabled."""
Expand Down Expand Up @@ -918,6 +936,22 @@ def greet(name: str = 'World') -> str:
assert schema['properties']['name']['default'] == 'World'
assert 'name' not in schema.get('required', [])

def test_annotated_field_metadata_preserved(self):
"""Test Annotated[T, Field(...)] metadata reaches the schema."""

def json_schema_annotated_function(
count: Annotated[int, Field(description='How many widgets', ge=1)],
) -> str:
return str(count)

function_decl = _automatic_function_calling_util.build_function_declaration(
func=json_schema_annotated_function
)

count_schema = function_decl.parameters_json_schema['properties']['count']
assert count_schema['description'] == 'How many widgets'
assert count_schema['minimum'] == 1


class TestBuildFunctionDeclarationFromSchemaDict:
"""Tests for the declaration builders that take a JSON schema dict.
Expand Down
98 changes: 98 additions & 0 deletions tests/unittests/tools/test_function_tool_declarations.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
from collections.abc import Sequence
import dataclasses
from enum import Enum
from typing import Annotated
from typing import Any
from typing import AsyncGenerator
from typing import Generator
Expand Down Expand Up @@ -639,6 +640,103 @@ def get_status() -> StandardReturnDataclass:
self.assertIn("status", decl.response_json_schema["properties"])


class TestAnnotatedMetadata(parameterized.TestCase):
"""Tests that Annotated[T, Field(...)] metadata reaches the schema."""

def test_annotated_field_metadata_in_schema(self):
"""Test descriptions, constraints and defaults attached via Annotated."""

def configure(
count: Annotated[
int, Field(description="How many widgets", ge=1, le=10)
],
zip_code: Annotated[str, Field(pattern=r"^\d{5}$")],
retries: Annotated[int, Field(description="Retry attempts")] = 3,
) -> str:
return "ok"

decl = build_function_declaration_with_json_schema(configure)
schema = decl.parameters_json_schema

self.assertEqual(
schema["properties"],
{
"count": {
"description": "How many widgets",
"maximum": 10,
"minimum": 1,
"title": "Count",
"type": "integer",
},
"zip_code": {
"pattern": r"^\d{5}$",
"title": "Zip Code",
"type": "string",
},
"retries": {
"default": 3,
"description": "Retry attempts",
"title": "Retries",
"type": "integer",
},
},
)
self.assertEqual(set(schema["required"]), {"count", "zip_code"})

def test_annotated_optional_model(self):
"""Test Annotated[Optional[Model], Field(...)] keeps its description."""

def save(
address: Annotated[
Optional[Address], Field(description="Where to ship")
] = None,
) -> str:
return "ok"

decl = build_function_declaration_with_json_schema(save)
address_schema = decl.parameters_json_schema["properties"]["address"]

self.assertEqual(address_schema["description"], "Where to ship")
self.assertIsNone(address_schema["default"])
self.assertEqual(
address_schema["anyOf"],
[{"$ref": "#/$defs/Address"}, {"type": "null"}],
)

def test_annotated_nested_list_constraints(self):
"""Test Annotated metadata on both a list and its item type."""

def tally(
scores: Annotated[
list[Annotated[int, Field(ge=0)]], Field(description="Scores")
],
) -> int:
return sum(scores)

decl = build_function_declaration_with_json_schema(tally)
scores_schema = decl.parameters_json_schema["properties"]["scores"]

self.assertEqual(scores_schema["description"], "Scores")
self.assertEqual(scores_schema["type"], "array")
self.assertEqual(scores_schema["items"]["type"], "integer")
self.assertEqual(scores_schema["items"]["minimum"], 0)

def test_annotated_return_type_metadata(self):
"""Test Annotated metadata on the return type."""

def count_items(
items: list[str],
) -> Annotated[int, Field(description="Item count")]:
return len(items)

decl = build_function_declaration_with_json_schema(count_items)

self.assertEqual(
decl.response_json_schema,
{"description": "Item count", "type": "integer"},
)


class TestSpecialCases(parameterized.TestCase):
"""Tests for special cases and edge cases."""

Expand Down
Loading