Python: Improve error message when TypeVar is used in handler registration

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

@@ -20,7 +20,7 @@
 from ._request_info_mixin import RequestInfoMixin
 from ._runner_context import MessageType, RunnerContext, WorkflowMessage
 from ._state import State
-from ._typing_utils import is_instance_of, normalize_type_to_list, resolve_type_annotation
+from ._typing_utils import is_instance_of, is_typevar, normalize_type_to_list, resolve_type_annotation
 from ._workflow_context import WorkflowContext, validate_workflow_context_annotation
 
 logger = logging.getLogger(__name__)
@@ -622,6 +622,20 @@ def decorator(
                 resolve_type_annotation(workflow_output, func.__globals__) if workflow_output is not None else None
             )
 
+            # Check for unresolved TypeVars in explicit type parameters
+            for param_name, param_type in [
+                ("input", resolved_input_type),
+                ("output", resolved_output_type),
+                ("workflow_output", resolved_workflow_output_type),
+            ]:
+                if param_type is not None and is_typevar(param_type):
+                    raise ValueError(
+                        f"Handler '{func.__name__}' has an unresolved TypeVar '{param_type}' "
+                        f"as its {param_name} type. "
+                        f"Use @handler(input=ConcreteType, output=ConcreteType) with concrete types "
+                        f"for parameterized executors."
+                    )
+
             # Validate signature structure (correct number of params, ctx is WorkflowContext)
             # but skip type extraction since we're using explicit types
             _validate_handler_signature(func, skip_message_annotation=True)
@@ -652,6 +666,15 @@ def decorator(
                     "or explicit type parameters (input, output, workflow_output)"
                 )
 
+            # Check for unresolved TypeVar in introspected message type
+            if is_typevar(message_type):
+                raise ValueError(
+                    f"Handler '{func.__name__}' has an unresolved TypeVar '{message_type}' "
+                    f"as its message type. "
+                    f"Use @handler(input=ConcreteType, output=ConcreteType) with concrete types "
+                    f"for parameterized executors."
+                )
+
             final_output_types = inferred_output_types
             final_workflow_output_types = inferred_workflow_output_types
 

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

@@ -24,7 +24,7 @@
 from typing import Any
 
 from ._executor import Executor
-from ._typing_utils import normalize_type_to_list, resolve_type_annotation
+from ._typing_utils import is_typevar, normalize_type_to_list, resolve_type_annotation
 from ._workflow_context import WorkflowContext, validate_workflow_context_annotation
 
 if sys.version_info >= (3, 11):
@@ -94,6 +94,19 @@ def __init__(
             _validate_function_signature(func, skip_message_annotation=resolved_input_type is not None)
         )
 
+        # Check for unresolved TypeVars in explicit type parameters
+        for param_name, param_type in [
+            ("input", resolved_input_type),
+            ("output", resolved_output_type),
+            ("workflow_output", resolved_workflow_output_type),
+        ]:
+            if param_type is not None and is_typevar(param_type):
+                raise ValueError(
+                    f"Executor '{func.__name__}' has an unresolved TypeVar '{param_type}' "
+                    f"as its {param_name} type. "
+                    f"Use @executor(input=ConcreteType, output=ConcreteType) with concrete types."
+                )
+
         # Use explicit types if provided, otherwise fall back to introspection
         message_type = resolved_input_type if resolved_input_type is not None else introspected_message_type
         output_types: list[type[Any] | types.UnionType] = (
@@ -114,6 +127,14 @@ def __init__(
                 "or an explicit input_type parameter"
             )
 
+        # Check for unresolved TypeVar in introspected message type
+        if is_typevar(message_type):
+            raise ValueError(
+                f"Executor '{func.__name__}' has an unresolved TypeVar '{message_type}' "
+                f"as its message type. "
+                f"Use @executor(input=ConcreteType, output=ConcreteType) with concrete types."
+            )
+
         # Store the original function
         self._original_func = func
         # Determine if function has WorkflowContext parameter

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

@@ -1,10 +1,30 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+import typing
 from types import UnionType
 from typing import Any, TypeGuard, Union, cast, get_args, get_origin
 
+import typing_extensions
+
 from .._agents import Agent
 
+# Pre-compute the TypeVar types for runtime-safe detection.
+# isinstance(x, TypeVar) can fail if TypeVar is a factory/callable
+# on some Python versions, so we compare against the actual runtime type.
+_TYPEVAR_TYPES: tuple[type, ...] = (type(typing.TypeVar("_T")), type(typing_extensions.TypeVar("_T")))  # pyright: ignore[reportUnknownVariableType]
+
+
+def is_typevar(x: Any) -> bool:
+    """Check if x is an unresolved TypeVar instance (from typing or typing_extensions).
+
+    Args:
+        x: The value to check.
+
+    Returns:
+        True if x is a TypeVar instance, False otherwise.
+    """
+    return isinstance(x, _TYPEVAR_TYPES)
+
 
 def is_chat_agent(agent: Any) -> TypeGuard[Agent]:
     """Check if the given agent is a Agent.

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

@@ -21,6 +21,7 @@
 )
 from ._runner_context import RunnerContext, WorkflowMessage
 from ._state import State
+from ._typing_utils import is_typevar
 
 if TYPE_CHECKING:
     from ._executor import Executor
@@ -176,10 +177,27 @@ def _is_type_like(x: Any) -> bool:
             if type_arg is Any:
                 continue
 
+            # Check for unresolved TypeVar early with an actionable error message
+            if is_typevar(type_arg):
+                raise ValueError(
+                    f"{context_description} {parameter_name} {param_description} "
+                    f"has an unresolved TypeVar '{type_arg}'. "
+                    f"Use @handler(input=ConcreteType, output=ConcreteType) with concrete types "
+                    f"for parameterized executors."
+                )
+
             # Check if it's a union type and validate each member
             union_origin = get_origin(type_arg)
             if union_origin in (Union, UnionType):
                 union_members = get_args(type_arg)
+                typevar_members = [m for m in union_members if is_typevar(m)]
+                if typevar_members:
+                    raise ValueError(
+                        f"{context_description} {parameter_name} {param_description} "
+                        f"contains unresolved TypeVar(s): {typevar_members}. "
+                        f"Use @handler(input=ConcreteType, output=ConcreteType) with concrete types "
+                        f"for parameterized executors."
+                    )
                 invalid_members = [m for m in union_members if not _is_type_like(m) and m is not Any]
                 if invalid_members:
                     raise ValueError(

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

@@ -0,0 +1,190 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for unresolved TypeVar detection during handler/executor registration."""
+
+from typing import TypeVar
+
+import pytest
+from typing_extensions import Never
+
+from agent_framework import (
+    Executor,
+    FunctionExecutor,
+    WorkflowContext,
+    executor,
+    handler,
+)
+from agent_framework._workflows._typing_utils import is_typevar
+
+T = TypeVar("T")
+U = TypeVar("U")
+
+
+class TestIsTypevarHelper:
+    """Tests for the runtime-safe is_typevar helper."""
+
+    def test_detects_typing_typevar(self):
+        """is_typevar should detect TypeVar from typing module."""
+        import typing
+
+        tv = typing.TypeVar("tv")
+        assert is_typevar(tv)
+
+    def test_detects_typing_extensions_typevar(self):
+        """is_typevar should detect TypeVar from typing_extensions module."""
+        import typing_extensions
+
+        tv = typing_extensions.TypeVar("tv")
+        assert is_typevar(tv)
+
+    def test_rejects_concrete_types(self):
+        """is_typevar should return False for concrete types."""
+        assert not is_typevar(str)
+        assert not is_typevar(int)
+        assert not is_typevar(None)
+        assert not is_typevar(Never)
+
+    def test_rejects_non_types(self):
+        """is_typevar should return False for non-type values."""
+        assert not is_typevar("hello")
+        assert not is_typevar(42)
+        assert not is_typevar([])
+
+
+class TestHandlerTypeVarValidation:
+    """Tests for @handler decorator rejecting unresolved TypeVars."""
+
+    def test_handler_explicit_input_typevar_raises(self):
+        """@handler(input=T) with a TypeVar should raise ValueError."""
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+            class _Bad(Executor):  # pyright: ignore[reportUnusedClass]
+                @handler(input=T)  # type: ignore[arg-type]
+                async def handle(self, message, ctx: WorkflowContext[str]) -> None:  # type: ignore[no-untyped-def]
+                    pass
+
+    def test_handler_explicit_output_typevar_raises(self):
+        """@handler(input=str, output=T) with a TypeVar should raise ValueError."""
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+            class _Bad(Executor):  # pyright: ignore[reportUnusedClass]
+                @handler(input=str, output=T)  # type: ignore[arg-type]
+                async def handle(self, message: str, ctx: WorkflowContext[str]) -> None:
+                    pass
+
+    def test_handler_explicit_workflow_output_typevar_raises(self):
+        """@handler(input=str, workflow_output=T) should raise ValueError."""
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+            class _Bad(Executor):  # pyright: ignore[reportUnusedClass]
+                @handler(input=str, workflow_output=T)  # type: ignore[arg-type]
+                async def handle(self, message: str, ctx: WorkflowContext[str]) -> None:
+                    pass
+
+    def test_handler_introspected_typevar_raises(self):
+        """@handler with TypeVar in message annotation should raise ValueError."""
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+            class _Bad(Executor):  # pyright: ignore[reportUnusedClass]
+                @handler  # type: ignore[arg-type]
+                async def handle(self, message: T, ctx: WorkflowContext[str]) -> None:  # type: ignore[valid-type]
+                    pass
+
+    def test_handler_concrete_types_work(self):
+        """@handler with concrete types should succeed."""
+
+        class Good(Executor):
+            @handler(input=str, output=str)
+            async def handle(self, message: str, ctx: WorkflowContext[str]) -> None:
+                pass
+
+        assert Good is not None
+
+
+class TestExecutorTypeVarValidation:
+    """Tests for @executor decorator rejecting unresolved TypeVars."""
+
+    def test_executor_explicit_input_typevar_raises(self):
+        """@executor(input=T) with a TypeVar should raise ValueError."""
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+            @executor(input=T)  # type: ignore[arg-type]
+            async def bad_func(message, ctx: WorkflowContext[str]) -> None:  # type: ignore[no-untyped-def]
+                pass
+
+    def test_executor_explicit_output_typevar_raises(self):
+        """@executor(input=str, output=T) with a TypeVar should raise ValueError."""
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+            @executor(input=str, output=T)  # type: ignore[arg-type]
+            async def bad_func(message: str, ctx: WorkflowContext[str]) -> None:
+                pass
+
+    def test_executor_introspected_typevar_raises(self):
+        """@executor with TypeVar in message annotation should raise ValueError."""
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+            FunctionExecutor(self._make_typevar_func())  # type: ignore[arg-type]
+
+    def test_executor_concrete_types_work(self):
+        """@executor with concrete types should succeed."""
+
+        @executor(input=str, output=str)
+        async def good_func(message: str, ctx: WorkflowContext[str]) -> None:
+            pass
+
+        assert good_func is not None
+
+    @staticmethod
+    def _make_typevar_func():
+        """Create a function with TypeVar annotation for testing."""
+
+        async def func(message: T, ctx: WorkflowContext[str]) -> None:  # type: ignore[valid-type]
+            pass
+
+        return func
+
+
+class TestWorkflowContextTypeVarValidation:
+    """Tests for WorkflowContext[T] rejecting unresolved TypeVars."""
+
+    def test_context_direct_typevar_raises(self):
+        """WorkflowContext[T] with a TypeVar should raise ValueError."""
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+            @executor(id="bad")
+            async def bad_func(message: str, ctx: WorkflowContext[T]) -> None:  # type: ignore[valid-type]
+                pass
+
+    def test_context_union_typevar_raises(self):
+        """WorkflowContext[T | str] with a TypeVar in union should raise ValueError."""
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+            @executor(id="bad")
+            async def bad_func(message: str, ctx: WorkflowContext[T | str]) -> None:  # type: ignore[valid-type]
+                pass
+
+    def test_context_workflow_output_typevar_raises(self):
+        """WorkflowContext[str, T] with a TypeVar should raise ValueError."""
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+            @executor(id="bad")
+            async def bad_func(message: str, ctx: WorkflowContext[str, T]) -> None:  # type: ignore[valid-type]
+                pass
+
+    def test_context_concrete_types_work(self):
+        """WorkflowContext[str] with concrete types should succeed."""
+
+        @executor(id="good")
+        async def good_func(message: str, ctx: WorkflowContext[str]) -> None:
+            pass
+
+        assert good_func is not None
+
+    def test_context_class_handler_typevar_raises(self):
+        """Class-based handler with WorkflowContext[T] should raise ValueError."""
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+            class _Bad(Executor):  # pyright: ignore[reportUnusedClass]
+                @handler  # pyright: ignore[reportUnknownArgumentType]
+                async def handle(self, message: str, ctx: WorkflowContext[T]) -> None:  # type: ignore[valid-type]
+                    pass