Compare commits

..
2 changed files with 177 additions and 2 deletions

No files matched your search

@@ -14,10 +14,13 @@ Usage:
# {"type": "string", "enum": ["fast", "deep"]}
"""
import collections.abc as _cabc
import contextvars
import dataclasses
import enum
import inspect
import logging
from typing import Any, Union, Literal, get_origin, get_args, get_type_hints, Callable, Dict, Optional, Set
from typing import Any, Union, Literal, get_origin, get_args, get_type_hints, Callable, Dict, Optional, Set, Tuple
def annotation_to_json_schema(annotation: Any) -> dict:
@@ -97,6 +100,44 @@ def annotation_to_json_schema(annotation: Any) -> dict:
return {"type": "array", "items": items_schema}
else:
return {"type": "array"}
# Handle other sequence/set-like containers (tuple, set, Sequence, ...)
if origin in _ARRAY_TYPES or annotation in _ARRAY_TYPES:
if annotation == Tuple[()] or annotation == tuple[()]:
return {"type": "array", "maxItems": 0}
item_args = [a for a in args if a is not Ellipsis]
if origin is tuple and item_args and Ellipsis not in args:
# Fixed-length tuple. prefixItems is not accepted by every provider
# (e.g. strict mode), so items stay a union; casting is positional.
positional = [annotation_to_json_schema(a) for a in item_args]
schema = {"type": "array",
"items": positional[0] if len(positional) == 1 else {"anyOf": positional},
"minItems": len(item_args), "maxItems": len(item_args)}
else:
schema = {"type": "array"}
if item_args:
schema["items"] = annotation_to_json_schema(item_args[0])
if origin in _SET_TYPES or annotation in _SET_TYPES:
schema["uniqueItems"] = True
return schema
# Handle Mapping types
if origin in _MAPPING_TYPES or annotation in _MAPPING_TYPES:
if len(args) >= 2:
return {"type": "object", "additionalProperties": annotation_to_json_schema(args[1])}
return {"type": "object"}
# Handle Pydantic models, dataclasses and TypedDicts
if isinstance(annotation, type):
if hasattr(annotation, "model_json_schema"):
schema = annotation.model_json_schema()
if "$defs" in schema:
# Nested $refs point at the root document; keep it a plain object
return {"type": "object"}
schema.pop("title", None)
return schema
if dataclasses.is_dataclass(annotation) or hasattr(annotation, "__required_keys__"):
return _object_schema_from_hints(annotation)
# Handle Dict types
if origin is dict:
@@ -120,6 +161,52 @@ def annotation_to_json_schema(annotation: Any) -> dict:
return {"type": type_map.get(annotation, "string")}
_SET_TYPES = (set, frozenset, _cabc.Set, _cabc.MutableSet)
_ARRAY_TYPES = (tuple, _cabc.Sequence, _cabc.MutableSequence, _cabc.Iterable,
_cabc.Collection) + _SET_TYPES
_MAPPING_TYPES = (_cabc.Mapping, _cabc.MutableMapping)
# Classes currently being expanded, so recursive types stop at a plain object
_EXPANDING: contextvars.ContextVar = contextvars.ContextVar("_schema_expanding", default=frozenset())
def _object_schema_from_hints(cls: type) -> dict:
"""Build an object schema for a dataclass or TypedDict from its type hints."""
expanding = _EXPANDING.get()
if cls in expanding:
return {"type": "object"}
hints = get_type_hints(cls)
if dataclasses.is_dataclass(cls):
# fields() already drops ClassVar/InitVar; keep only constructor inputs
fields = [f for f in dataclasses.fields(cls) if f.init]
names = [f.name for f in fields]
required = [
f.name for f in fields
if f.default is dataclasses.MISSING and f.default_factory is dataclasses.MISSING
]
else:
names = list(hints)
required_keys = set(getattr(cls, "__required_keys__", ()))
# __required_keys__ can be wrong under postponed annotations; trust markers
for k, v in get_type_hints(cls, include_extras=True).items():
marker = getattr(get_origin(v), "_name", None)
if marker == "Required":
required_keys.add(k)
elif marker == "NotRequired":
required_keys.discard(k)
required = [k for k in names if k in required_keys]
token = _EXPANDING.set(expanding | {cls})
try:
properties = {k: annotation_to_json_schema(hints[k]) for k in names}
finally:
_EXPANDING.reset(token)
schema = {"type": "object", "properties": properties}
if required:
schema["required"] = required
return schema
def fix_array_schemas(schema: Dict[str, Any]) -> Dict[str, Any]:
"""Recursively fix array schemas by adding a missing 'items' attribute.
@@ -361,7 +448,7 @@ def build_tool_definition(
try:
prop_schema = annotation_to_json_schema(annotation)
except Exception as e:
logging.debug(f"Could not generate schema for {function_name}.{name}: {e}")
logging.warning(f"Could not generate schema for {function_name}.{name}: {e}")
prop_schema = {"type": "string"}
if name in param_descriptions:
@@ -0,0 +1,88 @@
"""Regression tests for issue #6294: container/model tool schemas."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import ClassVar, List, Mapping, Sequence, Set, Tuple, TypedDict
from typing_extensions import NotRequired, Required
import pytest
from pydantic import BaseModel
from praisonaiagents.tools.schema import annotation_to_json_schema
class Addr(BaseModel):
city: str
@dataclass
class Point:
x: int
class TD(TypedDict):
a: int
@pytest.mark.parametrize("ann,expected_type", [
(Sequence[str], "array"), (Set[str], "array"), (Tuple[int, int], "array"),
(tuple, "array"), (set, "array"), (Mapping[str, int], "object"),
(Addr, "object"), (Point, "object"), (TD, "object"),
])
def test_container_and_model_schemas(ann, expected_type):
assert annotation_to_json_schema(ann)["type"] == expected_type
def test_nested_schema_items():
assert annotation_to_json_schema(List[Addr])["items"]["type"] == "object"
assert annotation_to_json_schema(Set[str])["uniqueItems"] is True
assert annotation_to_json_schema(Point)["properties"] == {"x": {"type": "integer"}}
def test_tuple_schema_fixed_length():
assert annotation_to_json_schema(Tuple[int])["maxItems"] == 1
assert annotation_to_json_schema(Tuple[int, str])["minItems"] == 2
def test_empty_tuple_allows_no_items():
assert annotation_to_json_schema(Tuple[()])["maxItems"] == 0
@dataclass
class WithExtras:
x: int
k: ClassVar[int] = 0
computed: int = field(init=False)
def test_dataclass_only_constructor_fields():
schema = annotation_to_json_schema(WithExtras)
assert schema["properties"] == {"x": {"type": "integer"}}
assert schema["required"] == ["x"]
@dataclass
class Node:
name: str
children: List["Node"]
def test_recursive_dataclass_stops_expanding():
items = annotation_to_json_schema(Node)["properties"]["children"]["items"]
assert items == {"type": "object"}
class TotalTD(TypedDict):
a: NotRequired[int]
b: int
class PartialTD(TypedDict, total=False):
c: Required[int]
d: int
def test_typeddict_required_markers():
assert annotation_to_json_schema(TotalTD)["required"] == ["b"]
assert annotation_to_json_schema(PartialTD)["required"] == ["c"]