From f33a6edda8b475891f0a35bb735ecf83c11b376f Mon Sep 17 00:00:00 2001 From: Daniel Peng Date: Fri, 23 Oct 2026 08:45:23 +0800 Subject: [PATCH] Python: preserve already-typed native function arguments --- .../functions/kernel_function_from_method.py | 7 ++ .../test_kernel_function_from_method.py | 110 +++++++++++++++++- 2 files changed, 116 insertions(+), 1 deletion(-) diff --git a/python/semantic_kernel/functions/kernel_function_from_method.py b/python/semantic_kernel/functions/kernel_function_from_method.py index d9dcf869b95c..6460335dae93 100644 --- a/python/semantic_kernel/functions/kernel_function_from_method.py +++ b/python/semantic_kernel/functions/kernel_function_from_method.py @@ -141,6 +141,13 @@ def _parse_parameter(self, value: Any, param_type: Any) -> Any: raise FunctionExecutionException(f"Expected a list for {param_type}, but got {type(value)}") else: try: + # Keep custom instances intact while retaining built-in constructor coercion. + if ( + isinstance(param_type, type) + and param_type.__module__ != "builtins" + and isinstance(value, param_type) + ): + return value if isinstance(value, dict) and hasattr(param_type, "__init__"): return param_type(**value) return param_type(value) 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 377fae16066f..c591a9aba836 100644 --- a/python/tests/unit/functions/test_kernel_function_from_method.py +++ b/python/tests/unit/functions/test_kernel_function_from_method.py @@ -2,6 +2,7 @@ import logging from collections.abc import AsyncGenerator, Iterable from contextlib import nullcontext +from dataclasses import dataclass from importlib import import_module from typing import Annotated, Any from unittest.mock import Mock, patch @@ -10,7 +11,7 @@ from opentelemetry.trace import StatusCode from semantic_kernel.connectors.ai.open_ai.services.open_ai_chat_completion import OpenAIChatCompletion -from semantic_kernel.exceptions import FunctionExecutionException, FunctionInitializationError +from semantic_kernel.exceptions import FunctionExecutionException, FunctionInitializationError, KernelInvokeException from semantic_kernel.filters.functions.function_invocation_context import FunctionInvocationContext from semantic_kernel.filters.kernel_filters_extension import _rebuild_function_invocation_context from semantic_kernel.functions.function_result import FunctionResult @@ -39,6 +40,30 @@ def __init__(self, id: str, name: str): self.name = name +@dataclass +class CustomTypeDataclass: + id: str + name: str + + +class DerivedCustomType(CustomTypeNonPydantic): + pass + + +@dataclass +class DerivedDataclass(CustomTypeDataclass): + pass + + +@dataclass +class SingleFieldDataclass: + name: str + + +class RevalidatedCustomType(CustomType): + model_config = {"revalidate_instances": "always", "validate_assignment": False} + + @pytest.fixture def get_custom_type_function_pydantic(): @kernel_function @@ -585,6 +610,89 @@ def func_default(base: str, input: str = "test"): assert str(res) == "test" +@pytest.mark.parametrize( + ("param_type", "value"), + [ + (CustomTypeNonPydantic, CustomTypeNonPydantic("1", "Ada")), + (CustomTypeDataclass, CustomTypeDataclass("1", "Ada")), + (SingleFieldDataclass, SingleFieldDataclass("Ada")), + (CustomTypeNonPydantic, DerivedCustomType("1", "Ada")), + (CustomTypeDataclass, DerivedDataclass("1", "Ada")), + ], +) +async def test_invoke_preserves_typed_instance(kernel, param_type, value): + @kernel_function + def identity(param: param_type) -> Any: + return param + + func = kernel.add_function(plugin_name="test", function=identity) + for _ in range(2): + result = await kernel.invoke(func, arguments=KernelArguments(param=value)) + assert result.value is value + + +@pytest.mark.parametrize("input_kind", ["dicts", "instances", "mixed"]) +async def test_invoke_list_of_typed_instances(kernel, input_kind): + @kernel_function + def identity(param: list[CustomTypeDataclass]) -> Any: + return param + + values = [CustomTypeDataclass("1", "Ada"), CustomTypeDataclass("2", "Grace")] + arguments = [ + {"id": value.id, "name": value.name} + if input_kind == "dicts" or (input_kind == "mixed" and index == 1) + else value + for index, value in enumerate(values) + ] + func = kernel.add_function(plugin_name="test", function=identity) + result = await kernel.invoke(func, arguments=KernelArguments(param=arguments)) + assert result.value == values + if input_kind != "dicts": + assert result.value[0] is values[0] + + +async def test_invoke_typed_instance_returned_by_function(kernel): + value = CustomTypeDataclass("1", "Ada") + + @kernel_function + def produce() -> CustomTypeDataclass: + return value + + @kernel_function + def consume(param: CustomTypeDataclass) -> str: + assert param is value + return param.name + + producer = kernel.add_function(plugin_name="test", function=produce) + consumer = kernel.add_function(plugin_name="test", function=consume) + produced = await kernel.invoke(producer) + result = await kernel.invoke(consumer, arguments=KernelArguments(param=produced.value)) + assert result.value == "Ada" + + +async def test_invoke_pydantic_instance_still_revalidated(kernel): + @kernel_function + def identity(param: RevalidatedCustomType) -> Any: + return param + + func = kernel.add_function(plugin_name="test", function=identity) + value = RevalidatedCustomType(id="1", name="Ada") + result = await kernel.invoke(func, arguments=KernelArguments(param=value)) + assert result.value == value + assert result.value is not value + + value.id = 123 + with pytest.raises(KernelInvokeException): + await kernel.invoke(func, arguments=KernelArguments(param=value)) + + +@pytest.mark.parametrize("value", [True, "7", 7]) +def test_parse_integer_still_coerces_values(get_custom_type_function_nonpydantic, value): + result = get_custom_type_function_nonpydantic._parse_parameter(value, int) + assert type(result) is int + assert result == int(value) + + def test_parse_list_of_objects(get_custom_type_function_pydantic): func = get_custom_type_function_pydantic