Python: resolve forward references nested in generic aliases when building schemas

python/semantic_kernel/functions/kernel_function_decorator.py

@@ -6,6 +6,8 @@
 from inspect import Parameter, Signature, isasyncgenfunction, isclass, isgeneratorfunction, signature
 from typing import Annotated, Any, ForwardRef, Union, get_args, get_origin
 
+from semantic_kernel.schema.kernel_json_schema_builder import KernelJsonSchemaBuilder
+
 NoneType = type(None)
 logger = logging.getLogger(__name__)
 
@@ -65,7 +67,7 @@ def decorator(func: Callable[..., object]) -> Callable[..., object]:
         logger.debug(f"Parsing decorator for function: {getattr(func, '__kernel_function_name__')}")
         func_sig = signature(func, eval_str=True)
 
-        annotations = _process_signature(func_sig)
+        annotations = _process_signature(func_sig, getattr(func, "__globals__", None))
         logger.debug(f"{annotations=}")
 
         setattr(func, "__kernel_function_parameters__", annotations)
@@ -114,8 +116,15 @@ def _get_underlying_type(annotation: Any) -> Any:
     return annotation
 
 
-def _process_signature(func_sig: Signature) -> list[dict[str, Any]]:
-    """Process the signature of the function."""
+def _process_signature(func_sig: Signature, globalns: dict[str, Any] | None = None) -> list[dict[str, Any]]:
+    """Process the signature of the function.
+
+    Args:
+        func_sig: The signature of the function.
+        globalns: The function's module globals, used to resolve forward references such as
+            ``list["Model"]`` in parameter annotations. ``signature(..., eval_str=True)`` evaluates an
+            annotation that is a string, but not a string nested inside a generic alias.
+    """
     annotations = []
 
     for arg in func_sig.parameters.values():
@@ -129,6 +138,8 @@ def _process_signature(func_sig: Signature) -> list[dict[str, Any]]:
             underlying_type = _get_underlying_type(annotation)
         else:
             underlying_type = annotation
+        if globalns is not None:
+            underlying_type = KernelJsonSchemaBuilder.resolve_forward_refs(underlying_type, globalns)
         parsed_annotation["type_object"] = underlying_type
         annotations.append(parsed_annotation)
 

python/semantic_kernel/schema/kernel_json_schema_builder.py

@@ -3,7 +3,7 @@
 import sys
 import types
 from enum import Enum
-from typing import Any, Literal, Union, get_args, get_origin, get_type_hints
+from typing import Annotated, Any, ForwardRef, Literal, Union, get_args, get_origin, get_type_hints
 
 from semantic_kernel.const import PARSED_ANNOTATION_UNION_DELIMITER
 from semantic_kernel.exceptions.function_exceptions import FunctionInvalidParameterConfiguration
@@ -86,6 +86,7 @@ def build_model_schema(
         hints = get_type_hints(model, globalns=model_module_globals, localns={})
 
         for field_name, field_type in hints.items():
+            field_type = cls.resolve_forward_refs(field_type, model_module_globals)
             field_description = None
             if hasattr(model, "model_fields") and field_name in model.model_fields:
                 field_info = model.model_fields[field_name]
@@ -109,6 +110,88 @@ def build_model_schema(
 
         return schema
 
+    @classmethod
+    def resolve_forward_refs(cls, annotation: Any, globalns: dict[str, Any]) -> Any:
+        """Resolves forward references sitting inside a generic alias.
+
+        On Python 3.10, ``get_type_hints`` evaluates an annotation that *is* a string and
+        ``ForwardRef`` objects nested inside a generic alias, but not a bare ``str`` sitting in a
+        PEP 585 alias's ``__args__``. ``list["Inner"]`` goes through ``list.__class_getitem__``,
+        which stores ``"Inner"`` verbatim with no ``ForwardRef`` wrapper, so nothing resolves it and
+        the element type falls through to the ``{"type": "object"}`` placeholder.
+        ``typing.Optional["Inner"]`` does wrap the string, which is why the top-level and
+        ``Optional`` forms already work.
+
+        Python 3.11 made ``get_type_hints`` wrap those strings itself (gh-85542), so on 3.11+ the
+        hints arrive fully resolved and this is a no-op that returns the annotation unchanged.
+
+        Args:
+            annotation: The annotation to resolve.
+            globalns: The namespace of the module owning the annotation.
+
+        Returns:
+            Any: The annotation, rebuilt with resolvable string arguments replaced by their types.
+        """
+        origin = get_origin(annotation)
+        args = get_args(annotation)
+        if origin is None or not args:
+            return annotation
+        if origin is Literal:
+            # Literal members are values, not names; a Literal["Inner"] must stay a string.
+            return annotation
+        if origin is Annotated:
+            # Only the annotated type is a type; the metadata that follows it is left alone.
+            return Annotated[(cls._resolve_forward_ref_arg(args[0], globalns), *args[1:])]
+
+        resolved_args = tuple(cls._resolve_forward_ref_arg(arg, globalns) for arg in args)
+        if resolved_args == args:
+            return annotation
+
+        try:
+            return origin[resolved_args]
+        except TypeError:
+            # Not every alias can be rebuilt from its origin and args; keep the original.
+            return annotation
+
+    @classmethod
+    def _resolve_forward_ref_arg(cls, arg: Any, globalns: dict[str, Any]) -> Any:
+        """Resolves a single generic argument, recursing into nested generic aliases.
+
+        Args:
+            arg: The generic argument to resolve.
+            globalns: The namespace to resolve the reference against.
+
+        Returns:
+            Any: The resolved type, or the argument unchanged when it cannot be resolved.
+        """
+        if isinstance(arg, str):
+            return cls._lookup_type(arg, globalns, default=arg)
+        if isinstance(arg, ForwardRef):
+            return cls._lookup_type(arg.__forward_arg__, globalns, default=arg)
+        return cls.resolve_forward_refs(arg, globalns)
+
+    @staticmethod
+    def _lookup_type(name: str, globalns: dict[str, Any], default: Any) -> Any:
+        """Looks a name up in the namespace, substituting it only when it names a type.
+
+        A forward reference that happens to share its name with a module, a function or a plain
+        value must not be rebuilt around that object: a module reaches ``build_model_schema`` and
+        fails on ``__module__``, and a function or value would produce a misleading schema. Keeping
+        the original reference lets the existing placeholder fallback apply instead.
+
+        Args:
+            name: The referenced name.
+            globalns: The namespace to look it up in.
+            default: What to return when the name is missing or does not name a type.
+
+        Returns:
+            Any: The class or typing construct the name refers to, or ``default``.
+        """
+        value = globalns.get(name, default)
+        if isinstance(value, type) or get_origin(value) is not None:
+            return value
+        return default
+
     @classmethod
     def _is_optional(cls, field_type: Any) -> bool:
         return get_origin(field_type) in {types.UnionType, Union} and type(None) in get_args(field_type)

python/tests/unit/functions/test_kernel_function_decorators.py

@@ -38,6 +38,14 @@ def func_docstring_as_description(self, input):
         """Description."""
         return input
 
+    @kernel_function
+    def func_forward_ref_list(self, items: list["InputObject"]):
+        return items
+
+    @kernel_function
+    def func_forward_ref_colliding_with_module(self, items: list["pytest"]):
+        return items
+
     @kernel_function
     def func_input_annotated(self, input: Annotated[str, "input description"]):
         return input
@@ -306,3 +314,61 @@ def test_annotation_parsing(name, annotation, description, type_, is_required):
     assert description == annotation_dict.get("description")
     assert type_ == annotation_dict["type_"]
     assert is_required == annotation_dict["is_required"]
+
+
+def test_kernel_function_resolves_forward_ref_in_list_parameter():
+    """list["InputObject"] on a kernel function parameter resolves to list[InputObject]."""
+    my_func = getattr(MiscClass(), "func_forward_ref_list")
+    param = my_func.__kernel_function_parameters__[0]
+
+    assert param["name"] == "items"
+    assert param["type_"] == "list[InputObject]"
+    assert param["type_object"] == list[InputObject]
+
+
+@pytest.mark.parametrize(
+    ("annotation", "expected_type_object", "expected_description"),
+    [
+        (list["InputObject"], list[InputObject], None),
+        (dict[str, "InputObject"], dict[str, InputObject], None),
+        (Annotated[list["InputObject"], "description"], list[InputObject], "description"),
+        (list["InputObject"] | None, list[InputObject], None),
+    ],
+)
+def test_process_signature_resolves_forward_refs_with_globalns(annotation, expected_type_object, expected_description):
+    param = Parameter(
+        name="items",
+        annotation=annotation,
+        default=Parameter.empty,
+        kind=Parameter.POSITIONAL_OR_KEYWORD,
+    )
+    func_sig = Signature(parameters=[param])
+
+    annotations = _process_signature(func_sig, globals())
+
+    assert annotations[0]["type_object"] == expected_type_object
+    assert annotations[0].get("description") == expected_description
+
+
+def test_process_signature_without_globalns_leaves_forward_refs_unchanged():
+    """Callers that pass no namespace get the previous behaviour."""
+    param = Parameter(
+        name="items",
+        annotation=list["InputObject"],
+        default=Parameter.empty,
+        kind=Parameter.POSITIONAL_OR_KEYWORD,
+    )
+    func_sig = Signature(parameters=[param])
+
+    annotations = _process_signature(func_sig)
+
+    assert annotations[0]["type_object"] == list["InputObject"]
+
+
+def test_kernel_function_forward_ref_that_names_a_module_is_left_unresolved():
+    """A reference colliding with a non type global must not be substituted or raise at decoration."""
+    my_func = getattr(MiscClass(), "func_forward_ref_colliding_with_module")
+    param = my_func.__kernel_function_parameters__[0]
+
+    assert param["type_"] == "list[pytest]"
+    assert param["type_object"] == list["pytest"]

python/tests/unit/schema/test_schema_builder.py

@@ -1,6 +1,7 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import json
+import sys
 from enum import Enum
 from typing import Annotated, Any, Literal, Optional, Union
 from unittest.mock import Mock
@@ -499,3 +500,123 @@ def test_build_model_with_literal_fields():
         },
         "required": ["level"],
     }
+
+
+class ForwardRefTarget(KernelBaseModel):
+    value: int
+    label: str
+
+
+class DirectListHolder(KernelBaseModel):
+    items: list[ForwardRefTarget] = []
+
+
+class ForwardRefListHolder(KernelBaseModel):
+    items: list["ForwardRefTarget"] = []
+
+
+class ForwardRefDictHolder(KernelBaseModel):
+    mapping: dict[str, "ForwardRefTarget"] = {}
+
+
+class ForwardRefNestedHolder(KernelBaseModel):
+    items: list[list["ForwardRefTarget"]] = []
+
+
+class ForwardRefOptionalHolder(KernelBaseModel):
+    items: list[Optional["ForwardRefTarget"]] = []
+
+
+class UnresolvableForwardRefHolder(KernelBaseModel):
+    items: list["NotDefinedAnywhere"] = []  # noqa: F821
+
+
+def test_build_model_schema_resolves_forward_ref_in_list():
+    """list["Target"] must produce the same schema as list[Target]."""
+    forward_ref = KernelJsonSchemaBuilder.build(ForwardRefListHolder)["properties"]["items"]
+    direct = KernelJsonSchemaBuilder.build(DirectListHolder)["properties"]["items"]
+
+    assert forward_ref == direct
+    assert forward_ref["items"]["properties"] == {"value": {"type": "integer"}, "label": {"type": "string"}}
+    assert forward_ref["items"]["required"] == ["value", "label"]
+
+
+def test_build_model_schema_resolves_forward_ref_in_dict():
+    """dict[str, "Target"] must carry the element schema through additionalProperties."""
+    schema = KernelJsonSchemaBuilder.build(ForwardRefDictHolder)["properties"]["mapping"]
+
+    assert schema["additionalProperties"]["properties"] == {
+        "value": {"type": "integer"},
+        "label": {"type": "string"},
+    }
+
+
+def test_build_model_schema_resolves_nested_forward_ref():
+    """Resolution recurses into nested generic aliases."""
+    schema = KernelJsonSchemaBuilder.build(ForwardRefNestedHolder)["properties"]["items"]
+
+    assert schema["items"]["items"]["properties"] == {
+        "value": {"type": "integer"},
+        "label": {"type": "string"},
+    }
+
+
+def test_build_model_schema_resolves_forward_ref_inside_optional():
+    """A ForwardRef nested under Optional inside a list still resolves."""
+    schema = KernelJsonSchemaBuilder.build(ForwardRefOptionalHolder)["properties"]["items"]
+
+    assert schema["items"]["type"] == ["object", "null"]
+    assert schema["items"]["properties"] == {"value": {"type": "integer"}, "label": {"type": "string"}}
+
+
+@pytest.mark.skipif(
+    sys.version_info >= (3, 11),
+    reason="On 3.11+ get_type_hints raises NameError for the unresolvable name before the builder sees it",
+)
+def test_build_model_schema_unresolvable_forward_ref_falls_back():
+    """An unresolvable name keeps the previous placeholder rather than raising."""
+    schema = KernelJsonSchemaBuilder.build(UnresolvableForwardRefHolder)["properties"]["items"]
+
+    assert schema == {"type": "array", "items": {"type": "object"}}
+
+
+def test_resolve_forward_refs_leaves_literal_values_and_annotated_metadata_alone():
+    """Strings that are values, not names, must not be looked up even when a global matches."""
+    namespace = {"ForwardRefTarget": ForwardRefTarget}
+
+    literal = KernelJsonSchemaBuilder.resolve_forward_refs(list[Literal["ForwardRefTarget"]], namespace)
+    annotated = KernelJsonSchemaBuilder.resolve_forward_refs(
+        list[Annotated["ForwardRefTarget", "ForwardRefTarget"]], namespace
+    )
+
+    assert literal == list[Literal["ForwardRefTarget"]]
+    assert annotated == list[Annotated[ForwardRefTarget, "ForwardRefTarget"]]
+
+
+def _not_a_type():
+    return None
+
+
+NOT_A_TYPE_VALUE = 3
+
+
+def test_resolve_forward_refs_ignores_names_that_do_not_refer_to_a_type():
+    """A reference that collides with a module, function or value keeps the placeholder fallback."""
+    namespace = {
+        "json": json,
+        "_not_a_type": _not_a_type,
+        "NOT_A_TYPE_VALUE": NOT_A_TYPE_VALUE,
+        "ForwardRefTarget": ForwardRefTarget,
+    }
+
+    assert KernelJsonSchemaBuilder.resolve_forward_refs(list["json"], namespace) == list["json"]
+    assert KernelJsonSchemaBuilder.resolve_forward_refs(list["_not_a_type"], namespace) == list["_not_a_type"]
+    assert (
+        KernelJsonSchemaBuilder.resolve_forward_refs(dict[str, "NOT_A_TYPE_VALUE"], namespace)
+        == (dict[str, "NOT_A_TYPE_VALUE"])
+    )
+    # a sibling argument that does name a type is still resolved
+    assert (
+        KernelJsonSchemaBuilder.resolve_forward_refs(dict["json", "ForwardRefTarget"], namespace)
+        == (dict["json", ForwardRefTarget])
+    )