From daa7e3d4377662e175ff10937f8460e6a10e9943 Mon Sep 17 00:00:00 2001 From: ADIL ALPEREN CIFTCI <134228585+adilalperenciftci@users.noreply.github.com> Date: Mon, 21 Sep 2026 09:49:18 +0300 Subject: [PATCH 1/2] fix(functions): validate complex kernel function arguments Complex and Union annotations containing commas could bypass parameter parsing in gather_function_parameters, allowing raw argument values to reach the kernel function. Validate generic annotations with Pydantic TypeAdapter and remove the condition that skipped parsing for comma-containing type metadata. --- .../functions/kernel_function_from_method.py | 10 ++++++-- .../test_kernel_function_from_method.py | 24 +++++++++++++++++++ 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/python/semantic_kernel/functions/kernel_function_from_method.py b/python/semantic_kernel/functions/kernel_function_from_method.py index d9dcf869b95c..1ad4349d77f6 100644 --- a/python/semantic_kernel/functions/kernel_function_from_method.py +++ b/python/semantic_kernel/functions/kernel_function_from_method.py @@ -6,7 +6,7 @@ from inspect import isasyncgen, isasyncgenfunction, isawaitable, iscoroutinefunction, isgenerator, isgeneratorfunction from typing import Any -from pydantic import Field, ValidationError +from pydantic import Field, TypeAdapter, ValidationError from semantic_kernel.exceptions import FunctionExecutionException, FunctionInitializationError from semantic_kernel.filters.functions.function_invocation_context import FunctionInvocationContext @@ -140,6 +140,13 @@ 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 getattr(param_type, "__origin__", None) is not None: + try: + return TypeAdapter(param_type).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: if isinstance(value, dict) and hasattr(param_type, "__init__"): return param_type(**value) @@ -171,7 +178,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 diff --git a/python/tests/unit/functions/test_kernel_function_from_method.py b/python/tests/unit/functions/test_kernel_function_from_method.py index 3b868625d87c..ccb3abde3d99 100644 --- a/python/tests/unit/functions/test_kernel_function_from_method.py +++ b/python/tests/unit/functions/test_kernel_function_from_method.py @@ -566,3 +566,27 @@ 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_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 + From b7dbf3ca3355f6ad5104ba195753dc07bfbddd46 Mon Sep 17 00:00:00 2001 From: ADIL ALPEREN CIFTCI <134228585+adilalperenciftci@users.noreply.github.com> Date: Mon, 21 Sep 2026 10:00:18 +0300 Subject: [PATCH 2/2] fix(functions): validate PEP 604 union arguments --- .../functions/kernel_function_decorator.py | 2 ++ .../functions/kernel_function_from_method.py | 4 ++-- .../functions/test_kernel_function_from_method.py | 11 +++++++++++ 3 files changed, 15 insertions(+), 2 deletions(-) diff --git a/python/semantic_kernel/functions/kernel_function_decorator.py b/python/semantic_kernel/functions/kernel_function_decorator.py index 4d103a1a9093..c93c2309aed7 100644 --- a/python/semantic_kernel/functions/kernel_function_decorator.py +++ b/python/semantic_kernel/functions/kernel_function_decorator.py @@ -124,6 +124,8 @@ def _process_signature(func_sig: Signature) -> list[dict[str, Any]]: parsed_annotation = _parse_parameter(arg.name, annotation, default) if get_origin(annotation) is Annotated or get_origin(annotation) in {Union, types.UnionType}: underlying_type = _get_underlying_type(annotation) + if underlying_type is None: + underlying_type = annotation else: underlying_type = annotation parsed_annotation["type_object"] = underlying_type diff --git a/python/semantic_kernel/functions/kernel_function_from_method.py b/python/semantic_kernel/functions/kernel_function_from_method.py index 1ad4349d77f6..dc4125aefe43 100644 --- a/python/semantic_kernel/functions/kernel_function_from_method.py +++ b/python/semantic_kernel/functions/kernel_function_from_method.py @@ -4,7 +4,7 @@ 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, TypeAdapter, ValidationError @@ -140,7 +140,7 @@ 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 getattr(param_type, "__origin__", None) is not None: + if get_origin(param_type) is not None: try: return TypeAdapter(param_type).validate_python(value) except Exception as exc: diff --git a/python/tests/unit/functions/test_kernel_function_from_method.py b/python/tests/unit/functions/test_kernel_function_from_method.py index ccb3abde3d99..f3756c341339 100644 --- a/python/tests/unit/functions/test_kernel_function_from_method.py +++ b/python/tests/unit/functions/test_kernel_function_from_method.py @@ -590,3 +590,14 @@ def union_function(value: int | str) -> int | str: 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