diff --git a/src/agents/function_schema.py b/src/agents/function_schema.py index 8860c15180..b11a76ee74 100644 --- a/src/agents/function_schema.py +++ b/src/agents/function_schema.py @@ -6,7 +6,7 @@ import re from collections.abc import Callable from dataclasses import dataclass -from typing import Annotated, Any, Literal, get_args, get_origin, get_type_hints +from typing import Annotated, Any, Literal, cast, get_args, get_origin, get_type_hints # griffelib exposes the `griffe` package at runtime but currently does not ship typing markers. from griffe import Docstring, DocstringSectionKind # type: ignore[import-untyped] @@ -433,6 +433,13 @@ def function_schema( # If a docstring param description exists, use it field_description = param_descs.get(name, None) + value_ann = ann + if param.kind in (param.VAR_POSITIONAL, param.VAR_KEYWORD): + field_info = _extract_field_info_from_metadata(param_metadata.get(name, ())) + if field_info is not None and field_info.metadata: + # Constraints apply to each value, not the collected container or its defaults. + value_ann = Annotated[(ann, *cast(Any, field_info).metadata)] + # Handle different parameter kinds if param.kind == param.VAR_POSITIONAL: # e.g. *args: extend positional args @@ -440,7 +447,7 @@ def function_schema( # Preserve a homogeneous tuple as the type of each positional argument. args_of_tuple = get_args(ann) if len(args_of_tuple) == 2 and args_of_tuple[1] is Ellipsis: - ann = list[ann] # type: ignore + ann = list[value_ann] # type: ignore # tuple[()] parameterizes an empty tuple and reports no args, while a bare # typing.Tuple is unparameterized and carries no element type to reject. elif hasattr(ann, "__args__"): @@ -453,7 +460,7 @@ def function_schema( ann = list[Any] else: # If user wrote *args: int, treat as List[int] - ann = list[ann] # type: ignore + ann = list[value_ann] # type: ignore # Default factory to empty list fields[name] = ( @@ -467,7 +474,7 @@ def function_schema( # annotation as the value type -- mirroring the variadic-positional handling above, # where ``*args: X`` becomes ``list[X]`` (see #4655). A bare ``**kwargs`` has ``ann`` # set to ``Any`` above, yielding ``dict[str, Any]``. - ann = dict[str, ann] # type: ignore + ann = dict[str, value_ann] # type: ignore fields[name] = ( ann, diff --git a/tests/test_function_schema.py b/tests/test_function_schema.py index 1d8325d1a0..a9b5a6c6b4 100644 --- a/tests/test_function_schema.py +++ b/tests/test_function_schema.py @@ -554,6 +554,88 @@ def func(**kwargs: int) -> int: assert func(*args, **kwargs) == 5 +def _var_positional_field_constraints(*scores: Annotated[int, Field(ge=0, le=10)]) -> int: + return sum(scores) + + +def _var_keyword_field_constraints(**scores: Annotated[int, Field(ge=0, le=10)]) -> int: + return sum(scores.values()) + + +@pytest.mark.parametrize( + "func, strict, container_type, value_key", + [ + (_var_positional_field_constraints, True, "array", "items"), + (_var_keyword_field_constraints, False, "object", "additionalProperties"), + ], +) +def test_variadic_field_constraints_in_schema(func, strict, container_type, value_key): + fs = function_schema(func, strict_json_schema=strict) + scores = fs.params_json_schema["properties"]["scores"] + + assert scores["type"] == container_type + assert scores[value_key] == {"type": "integer", "minimum": 0, "maximum": 10} + assert "minimum" not in scores + assert "maximum" not in scores + + +def test_variadic_field_constraints_apply_to_each_string(): + def func(*names: Annotated[str, Field(min_length=2)]) -> str: + return ",".join(names) + + fs = function_schema(func) + names = fs.params_json_schema["properties"]["names"] + assert names["items"]["minLength"] == 2 + assert "minItems" not in names + + parsed = fs.params_pydantic_model.model_validate({"names": ["ok"]}) + args, kwargs = fs.to_call_args(parsed) + assert func(*args, **kwargs) == "ok" + with pytest.raises(ValidationError): + fs.params_pydantic_model.model_validate({"names": ["ok", "x"]}) + + +def test_variadic_field_constraints_preserve_collection_defaults(): + def func( + *scores: Annotated[int, Field(default=5, alias="values", ge=0)], + **extras: Annotated[int, Field(..., alias="options", ge=0)], + ) -> int: + return sum(scores) + sum(extras.values()) + + fs = function_schema(func, strict_json_schema=False) + assert set(fs.params_json_schema["properties"]) == {"scores", "extras"} + assert not fs.params_json_schema.get("required") + for payload in ({}, {"scores": [], "extras": {}}): + parsed = fs.params_pydantic_model.model_validate(payload) + assert parsed.model_dump() == {"scores": [], "extras": {}} + args, kwargs = fs.to_call_args(parsed) + assert func(*args, **kwargs) == 0 + + +def test_variadic_field_constraints_preserve_homogeneous_tuple_values(): + def func(*pairs: Annotated[tuple[int, ...], Field(min_length=2)]) -> int: + return sum(sum(pair) for pair in pairs) + + fs = function_schema(func) + pairs = fs.params_json_schema["properties"]["pairs"] + assert pairs["items"]["minItems"] == 2 + assert pairs["items"]["items"] == {"type": "integer"} + parsed = fs.params_pydantic_model.model_validate({"pairs": [[1, 2]]}) + args, kwargs = fs.to_call_args(parsed) + assert args == [(1, 2)] + assert func(*args, **kwargs) == 3 + with pytest.raises(ValidationError): + fs.params_pydantic_model.model_validate({"pairs": [[1]]}) + + +def test_variadic_field_constraints_do_not_bypass_fixed_tuple_rejection(): + def func(*pairs: Annotated[tuple[int, str], Field(min_length=2)]) -> int: + return len(pairs) + + with pytest.raises(UserError, match=r"use tuple\[T, \.\.\.\] or list\[T\] instead"): + function_schema(func) + + def test_schema_with_mapping_raises_strict_mode_error(): """A mapping type is not allowed in strict mode. Same for dicts. Ensure we raise a UserError.""" diff --git a/tests/test_function_tool.py b/tests/test_function_tool.py index de03b2e6bc..114caa459b 100644 --- a/tests/test_function_tool.py +++ b/tests/test_function_tool.py @@ -6,10 +6,10 @@ import logging import time from collections.abc import Callable -from typing import Any, cast +from typing import Annotated, Any, cast import pytest -from pydantic import BaseModel +from pydantic import BaseModel, Field from typing_extensions import TypedDict import agents._debug as _debug @@ -175,6 +175,44 @@ async def test_simple_function(): ) +@pytest.mark.asyncio +@pytest.mark.parametrize("keyword_values", [False, True], ids=["args", "kwargs"]) +async def test_variadic_field_constraints_validate_before_invocation(keyword_values: bool) -> None: + calls: list[int] = [] + + def positional(*scores: Annotated[int, Field(..., ge=0, le=10)]) -> int: + calls.append(sum(scores)) + return sum(scores) + + def keywords(**scores: Annotated[int, Field(..., ge=0, le=10)]) -> int: + calls.append(sum(scores.values())) + return sum(scores.values()) + + tool = function_tool( + keywords if keyword_values else positional, + strict_mode=not keyword_values, + failure_error_function=None, + ) + + async def invoke(payload: dict[str, Any]) -> Any: + arguments = json.dumps(payload) + context = ToolContext( + context=None, tool_name=tool.name, tool_call_id="1", tool_arguments=arguments + ) + return await tool.on_invoke_tool(context, arguments) + + for invalid in (-1, 11): + values = {"first": invalid} if keyword_values else [invalid] + with pytest.raises(ModelBehaviorError): + await invoke({"scores": values}) + assert calls == [] + + assert await invoke({"scores": {"a": 0, "b": 10} if keyword_values else [0, 10]}) == 10 + assert await invoke({}) == 0 + assert await invoke({"scores": {} if keyword_values else []}) == 0 + assert calls == [10, 0, 0] + + @pytest.mark.asyncio async def test_sync_function_runs_via_to_thread(monkeypatch: pytest.MonkeyPatch) -> None: calls = {"to_thread": 0, "func": 0}