fix: preserve Field constraints on variadic tool parameters

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,14 +433,21 @@ 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
             if get_origin(ann) is tuple:
                 # 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,

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."""
 

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}