Python: Raise clear handler registration error for unresolved TypeVar annotations

python/packages/core/agent_framework/_workflows/_executor.py

@@ -728,9 +728,22 @@ def _validate_handler_signature(
     # AttributeError, or RecursionError), so registration failures are easier to diagnose.
     try:
         type_hints = typing.get_type_hints(func)
-    except Exception:
+    except (NameError, AttributeError, RecursionError):
         type_hints = {p.name: p.annotation for p in params}
 
+    message_type = type_hints.get(message_param.name, message_param.annotation)
+    if message_type == inspect.Parameter.empty:
+        message_type = None
+
+    # Reject unresolved TypeVar in message annotation -- these are not supported
+    # for workflow type validation and must be replaced with concrete types.
+    if not skip_message_annotation and isinstance(message_type, TypeVar):
+        raise ValueError(
+            f"Handler {func.__name__} has an unresolved TypeVar '{message_type}' as its message type annotation. "
+            "Generic TypeVar annotations are not supported for workflow type validation. "
+            "Use @handler(input=<concrete_type>, output=<concrete_type>) to specify explicit types."
+        )
+
     # Validate ctx parameter is WorkflowContext and extract type args
     ctx_param = params[2]
     ctx_annotation = type_hints.get(ctx_param.name, ctx_param.annotation)
@@ -744,10 +757,6 @@ def _validate_handler_signature(
             ctx_annotation, f"parameter '{ctx_param.name}'", "Handler"
         )
 
-    message_type = type_hints.get(message_param.name, message_param.annotation)
-    if message_type == inspect.Parameter.empty:
-        message_type = None
-
     return message_type, ctx_annotation, output_types, workflow_output_types
 
 

python/packages/core/agent_framework/_workflows/_function_executor.py

@@ -325,11 +325,26 @@ def _validate_function_signature(
     if not skip_message_annotation and message_param.annotation == inspect.Parameter.empty:
         raise ValueError(f"Function instance {func.__name__} must have a type annotation for the message parameter")
 
-    type_hints = typing.get_type_hints(func)
+    # Resolve string annotations from `from __future__ import annotations`.
+    # Fall back to raw annotations if resolution fails (e.g. unresolvable forward refs,
+    # AttributeError, or RecursionError), so registration failures are easier to diagnose.
+    try:
+        type_hints = typing.get_type_hints(func)
+    except (NameError, AttributeError, RecursionError):
+        type_hints = {p.name: p.annotation for p in params}
     message_type = type_hints.get(message_param.name, message_param.annotation)
     if message_type == inspect.Parameter.empty:
         message_type = None
 
+    # Reject unresolved TypeVar in message annotation -- these are not supported
+    # for workflow type validation and must be replaced with concrete types.
+    if not skip_message_annotation and isinstance(message_type, typing.TypeVar):
+        raise ValueError(
+            f"Function instance {func.__name__} has an unresolved TypeVar '{message_type}' as its message type "
+            "annotation. Generic TypeVar annotations are not supported for workflow type validation. "
+            "Use @executor(input=<concrete_type>, output=<concrete_type>) to specify explicit types."
+        )
+
     # Check if there's a context parameter
     if len(params) == 2:
         ctx_param = params[1]

python/packages/core/tests/workflow/test_executor.py

@@ -1,6 +1,7 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 from dataclasses import dataclass
+from typing import Generic, TypeVar
 
 import pytest
 from typing_extensions import Never
@@ -919,3 +920,87 @@ async def handle(self, message: str, ctx: WorkflowContext[int, bool]) -> None:
 
 
 # endregion: Tests for @handler decorator with explicit input_type and output_type
+
+
+# region Tests for unresolved TypeVar rejection in handler registration
+
+_T = TypeVar("_T")
+
+
+def test_handler_rejects_unresolved_typevar_in_message_annotation():
+    """Test that @handler raises ValueError when the message parameter is an unresolved TypeVar."""
+
+    with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+        class GenericEcho(Executor, Generic[_T]):
+            @handler
+            async def echo(self, message: _T, ctx: WorkflowContext) -> None:
+                pass
+
+
+_BT = TypeVar("_BT", bound=str)
+
+
+def test_handler_rejects_bounded_typevar_in_message_annotation():
+    """Test that @handler raises ValueError for a bounded TypeVar in message annotation."""
+
+    with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+        class BoundedGenericExecutor(Executor, Generic[_BT]):
+            @handler
+            async def process(self, message: _BT, ctx: WorkflowContext) -> None:
+                await ctx.send_message(message)
+
+
+def test_handler_allows_concrete_types():
+    """Test that @handler works normally with concrete type annotations."""
+
+    class ConcreteExecutor(Executor):
+        @handler
+        async def handle(self, message: str, ctx: WorkflowContext[str]) -> None:
+            pass
+
+    exec_instance = ConcreteExecutor(id="concrete")
+    assert str in exec_instance.input_types
+
+
+def test_handler_explicit_input_bypasses_typevar_check():
+    """Test that @handler(input=...) bypasses TypeVar check since explicit types take precedence."""
+
+    class GenericWithExplicit(Executor, Generic[_T]):
+        @handler(input=str, output=str)
+        async def echo(self, message, ctx: WorkflowContext) -> None:
+            pass
+
+    exec_instance = GenericWithExplicit(id="explicit")
+    assert str in exec_instance.input_types
+
+
+def test_handler_error_message_recommends_explicit_types():
+    """Test that the TypeVar error message recommends @handler(input=..., output=...)."""
+
+    with pytest.raises(ValueError, match=r"@handler\(input=<concrete_type>, output=<concrete_type>\)"):
+
+        class GenericBad(Executor, Generic[_T]):
+            @handler
+            async def echo(self, message: _T, ctx: WorkflowContext) -> None:
+                pass
+
+
+# endregion: Tests for unresolved TypeVar rejection in handler registration
+
+
+def test_handler_typevar_error_takes_priority_over_context_error():
+    """Test that TypeVar message error is raised before WorkflowContext validation.
+
+    When a handler has both a TypeVar message annotation and an unannotated ctx
+    parameter, the TypeVar error should be reported first since it is the more
+    actionable issue.
+    """
+
+    with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+        class DualBad(Executor, Generic[_T]):
+            @handler
+            async def process(self, message: _T, ctx) -> None:  # type: ignore[no-untyped-def]
+                pass

python/packages/core/tests/workflow/test_function_executor.py

@@ -1,7 +1,7 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, TypeVar
 
 import pytest
 from typing_extensions import Never
@@ -895,3 +895,73 @@ async def my_func(message: str, ctx: WorkflowContext) -> None:
         assert str in exec_instance._handlers  # pyright: ignore[reportPrivateUsage]
         assert int in exec_instance.output_types
         assert bool in exec_instance.workflow_output_types
+
+
+# region Tests for unresolved TypeVar rejection in function executor registration
+
+_FT = TypeVar("_FT")
+
+
+class TestFunctionExecutorTypeVarRejection:
+    """Tests that FunctionExecutor rejects unresolved TypeVar in message annotations."""
+
+    def test_function_executor_rejects_unresolved_typevar(self):
+        """Test that FunctionExecutor raises ValueError for unresolved TypeVar message annotation."""
+
+        def echo(message: _FT) -> _FT:
+            return message
+
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+            FunctionExecutor(echo, id="echo")
+
+    def test_function_executor_rejects_typevar_with_context(self):
+        """Test that FunctionExecutor raises ValueError for TypeVar even with WorkflowContext."""
+
+        async def echo(message: _FT, ctx: WorkflowContext) -> None:
+            pass
+
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+            FunctionExecutor(echo, id="echo")
+
+    def test_function_executor_explicit_input_bypasses_typevar_check(self):
+        """Test that explicit input= parameter bypasses TypeVar detection."""
+
+        async def echo(message: _FT, ctx: WorkflowContext) -> None:
+            pass
+
+        exec_instance = FunctionExecutor(echo, id="echo", input=str, output=str)
+        assert str in exec_instance.input_types
+
+    def test_function_executor_allows_concrete_types(self):
+        """Test that FunctionExecutor works normally with concrete type annotations."""
+
+        async def handle(message: str, ctx: WorkflowContext[str]) -> None:
+            pass
+
+        exec_instance = FunctionExecutor(handle, id="concrete")
+        assert str in exec_instance.input_types
+
+    def test_function_executor_error_recommends_explicit_types(self):
+        """Test that error message recommends @executor(input=..., output=...)."""
+
+        def echo(message: _FT) -> _FT:
+            return message
+
+        with pytest.raises(ValueError, match=r"@executor\(input=<concrete_type>, output=<concrete_type>\)"):
+            FunctionExecutor(echo, id="echo")
+
+
+# endregion: Tests for unresolved TypeVar rejection in function executor registration
+
+
+_FBT = TypeVar("_FBT", bound=str)
+
+
+def test_function_executor_rejects_bounded_typevar_in_message_annotation():
+    """Test that FunctionExecutor raises ValueError for a bounded TypeVar in message annotation."""
+
+    async def process(message: _FBT, ctx: WorkflowContext) -> None:
+        await ctx.send_message(message)
+
+    with pytest.raises(ValueError, match="unresolved TypeVar"):
+        FunctionExecutor(process, id="bounded")

python/packages/core/tests/workflow/test_function_executor_future.py

@@ -4,6 +4,8 @@
 
 from typing import Any
 
+import pytest
+
 from agent_framework import FunctionExecutor, WorkflowContext, executor
 
 
@@ -37,3 +39,20 @@ async def process_complex(data: dict[str, Any], ctx: WorkflowContext[list[str]])
         spec = process_complex._handler_specs[0]  # pyright: ignore[reportPrivateUsage]
         assert spec["message_type"] == dict[str, Any]
         assert spec["output_types"] == [list[str]]
+
+    def test_handler_unresolvable_annotation_raises(self):
+        """Test that an unresolvable forward-reference annotation raises ValueError.
+
+        When get_type_hints fails (e.g. NameError for NonExistentType), the code falls back
+        to raw string annotations. The ctx parameter's raw string annotation is then not
+        recognised as a valid WorkflowContext type, so a ValueError is still raised.
+        """
+        with pytest.raises(ValueError):
+            FunctionExecutor(
+                _func_with_bad_annotation,  # pyright: ignore[reportUnknownArgumentType]
+                id="bad",
+            )
+
+
+async def _func_with_bad_annotation(message: NonExistentType, ctx: WorkflowContext[int]) -> None:  # noqa: F821  # type: ignore[name-defined]
+    pass