Python: fix(functions): validate complex kernel function arguments

python/semantic_kernel/functions/kernel_function_decorator.py

@@ -102,14 +102,14 @@ def _get_non_none_type(args: tuple) -> Any:
 def _get_underlying_type(annotation: Any) -> Any:
     """Get the underlying type of the annotation."""
     if isinstance(annotation, types.UnionType):
-        return _get_non_none_type(annotation.__args__)
+        return _get_non_none_type(annotation.__args__) or annotation
 
     if hasattr(annotation, "__origin__"):
         if annotation.__origin__ is Union:
-            return _get_non_none_type(get_args(annotation))
+            return _get_non_none_type(get_args(annotation)) or annotation
 
         if isinstance(annotation.__origin__, types.UnionType):
-            return _get_non_none_type(annotation.__origin__.__args__)
+            return _get_non_none_type(annotation.__origin__.__args__) or annotation.__origin__
 
         return annotation.__origin__
 

python/semantic_kernel/functions/kernel_function_from_method.py

@@ -4,9 +4,9 @@
 import logging
 from collections.abc import Callable
 from inspect import isasyncgen, isasyncgenfunction, isawaitable, iscoroutinefunction, isgenerator, isgeneratorfunction
-from typing import Any
+from typing import Any, get_origin
 
-from pydantic import Field, ValidationError
+from pydantic import ConfigDict, Field, TypeAdapter, ValidationError
 
 from semantic_kernel.exceptions import FunctionExecutionException, FunctionInitializationError
 from semantic_kernel.filters.functions.function_invocation_context import FunctionInvocationContext
@@ -140,6 +140,14 @@ def _parse_parameter(self, value: Any, param_type: Any) -> Any:
                 return [self._parse_parameter(item, item_type) for item in value]
             raise FunctionExecutionException(f"Expected a list for {param_type}, but got {type(value)}")
         else:
+            if get_origin(param_type) is not None:
+                try:
+                    adapter = TypeAdapter(param_type, config=ConfigDict(arbitrary_types_allowed=True))
+                    return adapter.validate_python(value)
+                except Exception as exc:
+                    raise FunctionExecutionException(
+                        f"Parameter is expected to be parsed to {param_type} but is not."
+                    ) from exc
             try:
                 # Keep custom instances intact while retaining built-in constructor coercion.
                 if (
@@ -178,7 +186,6 @@ def gather_function_parameters(self, context: FunctionInvocationContext) -> dict
                 value: Any = context.arguments[param.name]
                 if (
                     param.type_
-                    and "," not in param.type_
                     and param.type_object
                     and param.type_object is not inspect._empty
                     and param.type_object is not Any

python/tests/unit/functions/test_kernel_function_from_method.py

@@ -883,3 +883,122 @@ def test_function_model_dump_json(get_custom_type_function_pydantic):
     model_dump = func.model_dump_json()
     assert isinstance(model_dump, str)
     assert "metadata" in model_dump
+
+
+async def test_function_invoke_accepts_valid_typed_dict_argument(kernel: Kernel):
+    @kernel_function
+    def typed_dict_function(value: dict[str, int]) -> dict[str, int]:
+        return value
+
+    function = KernelFunction.from_method(typed_dict_function, "test")
+    result = await function.invoke(kernel=kernel, arguments=KernelArguments(value={"first": 1, "second": 2}))
+
+    assert result.value == {"first": 1, "second": 2}
+
+
+async def test_function_invoke_rejects_invalid_typed_dict_argument(kernel: Kernel):
+    invoked = False
+
+    @kernel_function
+    def typed_dict_function(value: dict[str, int]) -> dict[str, int]:
+        nonlocal invoked
+        invoked = True
+        return value
+
+    function = KernelFunction.from_method(typed_dict_function, "test")
+
+    with pytest.raises(
+        FunctionExecutionException,
+        match=r"Parameter value is expected to be parsed to .* but is not\.",
+    ):
+        await function.invoke(
+            kernel=kernel,
+            arguments=KernelArguments(value={"first": "not-an-integer"}),
+        )
+
+    assert not invoked
+
+
+async def test_function_invoke_rejects_invalid_union_argument(kernel: Kernel):
+    invoked = False
+
+    @kernel_function
+    def union_function(value: int | str) -> int | str:
+        nonlocal invoked
+        invoked = True
+        return value
+
+    function = KernelFunction.from_method(union_function, "test")
+
+    with pytest.raises(
+        FunctionExecutionException,
+        match=r"Parameter value is expected to be parsed to .* but is not\.",
+    ):
+        await function.invoke(
+            kernel=kernel,
+            arguments=KernelArguments(value={"unexpected": "value"}),
+        )
+
+    assert not invoked
+
+
+@pytest.mark.parametrize("value", [42, "valid"])
+async def test_function_invoke_accepts_valid_union_argument(kernel: Kernel, value: int | str):
+    @kernel_function
+    def union_function(value: int | str) -> int | str:
+        return value
+
+    function = KernelFunction.from_method(union_function, "test")
+    result = await function.invoke(kernel=kernel, arguments=KernelArguments(value=value))
+
+    assert result.value == value
+
+
+class CustomArgument:
+    """Plain Python class used to verify arbitrary types survive union validation."""
+
+    def __init__(self, label: str) -> None:
+        self.label = label
+
+
+async def test_function_invoke_accepts_valid_primitive_in_custom_union(kernel: Kernel):
+    @kernel_function
+    def custom_union_function(value: CustomArgument | str) -> str:
+        return value if isinstance(value, str) else value.label
+
+    function = KernelFunction.from_method(custom_union_function, "test")
+    result = await function.invoke(kernel=kernel, arguments=KernelArguments(value="plain"))
+
+    assert result.value == "plain"
+
+
+async def test_function_invoke_accepts_custom_instance_in_custom_union(kernel: Kernel):
+    @kernel_function
+    def custom_union_function(value: CustomArgument | str) -> str:
+        return value if isinstance(value, str) else value.label
+
+    function = KernelFunction.from_method(custom_union_function, "test")
+    instance = CustomArgument("instance")
+    result = await function.invoke(kernel=kernel, arguments=KernelArguments(value=instance))
+
+    assert result.value == "instance"
+
+
+async def test_function_invoke_rejects_invalid_custom_union_argument(kernel: Kernel):
+    invoked = False
+
+    @kernel_function
+    def custom_union_function(value: CustomArgument | str) -> str:
+        nonlocal invoked
+        invoked = True
+        return value if isinstance(value, str) else value.label
+
+    function = KernelFunction.from_method(custom_union_function, "test")
+
+    with pytest.raises(
+        FunctionExecutionException,
+        match=r"Parameter value is expected to be parsed to .* but is not\.",
+    ):
+        await function.invoke(kernel=kernel, arguments=KernelArguments(value={"unexpected": "value"}))
+
+    assert not invoked