Python: resolve postponed @response_handler annotations

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

@@ -7,7 +7,6 @@
 import inspect
 import logging
 import types
-import typing
 from collections.abc import Awaitable, Callable
 from typing import Any, TypeVar, overload
 
@@ -21,7 +20,13 @@
 from ._request_info_mixin import RequestInfoMixin
 from ._runner_context import MessageType, RunnerContext, WorkflowMessage
 from ._state import State
-from ._typing_utils import contains_typevar, is_instance_of, normalize_type_to_list, resolve_type_annotation
+from ._typing_utils import (
+    _resolve_function_annotations,
+    contains_typevar,
+    is_instance_of,
+    normalize_type_to_list,
+    resolve_type_annotation,
+)
 from ._workflow_context import WorkflowContext, validate_workflow_context_annotation
 
 logger = logging.getLogger(__name__)
@@ -778,13 +783,7 @@ def _validate_handler_signature(
     if not skip_message_annotation and message_param.annotation == inspect.Parameter.empty:
         raise ValueError(f"Handler {func.__name__} must have a type annotation for the message parameter")
 
-    # 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}
+    type_hints = _resolve_function_annotations(func, params)
 
     message_type = type_hints.get(message_param.name, message_param.annotation)
     if message_type == inspect.Parameter.empty:

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

@@ -19,12 +19,16 @@
 import inspect
 import sys
 import types
-import typing
 from collections.abc import Awaitable, Callable
 from typing import Any
 
 from ._executor import Executor
-from ._typing_utils import contains_typevar, normalize_type_to_list, resolve_type_annotation
+from ._typing_utils import (
+    _resolve_function_annotations,
+    contains_typevar,
+    normalize_type_to_list,
+    resolve_type_annotation,
+)
 from ._workflow_context import WorkflowContext, validate_workflow_context_annotation
 
 if sys.version_info >= (3, 11):
@@ -359,13 +363,7 @@ 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")
 
-    # 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}
+    type_hints = _resolve_function_annotations(func, params)
     message_type = type_hints.get(message_param.name, message_param.annotation)
     if message_type == inspect.Parameter.empty:
         message_type = None

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

@@ -11,7 +11,14 @@
 from types import UnionType
 from typing import TYPE_CHECKING, Any, TypeVar, cast
 
-from ._typing_utils import is_instance_of, is_type_compatible, normalize_type_to_list, resolve_type_annotation
+from ._typing_utils import (
+    _resolve_function_annotations,
+    contains_typevar,
+    is_instance_of,
+    is_type_compatible,
+    normalize_type_to_list,
+    resolve_type_annotation,
+)
 from ._workflow_context import WorkflowContext, validate_workflow_context_annotation
 
 if sys.version_info >= (3, 11):
@@ -241,6 +248,19 @@ def decorator(
                     f"Response handler {func.__name__} with explicit type parameters must specify 'response' type"
                 )
 
+            for param_name, param_type in [
+                ("request", resolved_request_type),
+                ("response", resolved_response_type),
+            ]:
+                if contains_typevar(param_type):
+                    raise ValueError(
+                        f"Response handler {func.__name__} has an unresolved TypeVar '{param_type}' "
+                        f"as its {param_name} type. "
+                        "Generic TypeVar annotations are not supported for workflow type validation. "
+                        "Use @response_handler(request=<concrete_type>, response=<concrete_type>) "
+                        "to specify explicit types."
+                    )
+
             final_request_type = resolved_request_type
             final_response_type = resolved_response_type
             final_output_types = normalize_type_to_list(resolved_output_type) if resolved_output_type else []
@@ -348,20 +368,39 @@ def _validate_response_handler_signature(
     if not skip_annotations and response_param.annotation == inspect.Parameter.empty:
         raise ValueError(f"Response handler {func.__name__} must have a type annotation for the response parameter")
 
+    type_hints = _resolve_function_annotations(func, params)
+
     # Validate ctx parameter is WorkflowContext and extract type args (if annotated)
     ctx_param = params[3]
+    ctx_annotation = type_hints.get(ctx_param.name, ctx_param.annotation)
     if ctx_param.annotation != inspect.Parameter.empty:
         output_types, workflow_output_types = validate_workflow_context_annotation(
-            ctx_param.annotation, f"parameter '{ctx_param.name}'", "Response handler"
+            ctx_annotation, f"parameter '{ctx_param.name}'", "Response handler"
         )
     else:
         output_types, workflow_output_types = [], []
 
-    request_type = (
-        original_request_param.annotation if original_request_param.annotation != inspect.Parameter.empty else None
-    )
-    response_type = response_param.annotation if response_param.annotation != inspect.Parameter.empty else None
-    ctx_annotation = ctx_param.annotation if ctx_param.annotation != inspect.Parameter.empty else None
+    request_type = type_hints.get(original_request_param.name, original_request_param.annotation)
+    if request_type == inspect.Parameter.empty:
+        request_type = None
+    response_type = type_hints.get(response_param.name, response_param.annotation)
+    if response_type == inspect.Parameter.empty:
+        response_type = None
+    if ctx_annotation == inspect.Parameter.empty:
+        ctx_annotation = None
+
+    for param_name, param_type in [
+        ("original_request", request_type),
+        ("response", response_type),
+    ]:
+        if param_type is not None and contains_typevar(param_type):
+            raise ValueError(
+                f"Response handler {func.__name__} has an unresolved TypeVar '{param_type}' "
+                f"as its {param_name} type annotation. "
+                "Generic TypeVar annotations are not supported for workflow type validation. "
+                "Use @response_handler(request=<concrete_type>, response=<concrete_type>) "
+                "to specify explicit types."
+            )
 
     return request_type, response_type, ctx_annotation, output_types, workflow_output_types
 

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

@@ -2,7 +2,7 @@
 
 import sys
 import typing
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping, Sequence
 from dataclasses import InitVar, is_dataclass
 from types import ModuleType, UnionType
 from typing import Any, Literal, TypeGuard, Union, cast, get_args, get_origin, get_type_hints
@@ -55,6 +55,19 @@ def contains_typevar(annotation: Any) -> bool:
     return any(contains_typevar(arg) for arg in get_args(annotation))
 
 
+def _resolve_function_annotations(func: Callable[..., Any], params: Sequence[Any]) -> dict[str, Any]:
+    """Resolve function annotations and fall back to raw parameter annotations on failure.
+
+    ``typing.get_type_hints`` resolves postponed annotations, but a single unresolved
+    forward reference prevents it from returning any hints. Keeping the raw annotations
+    in that case lets the caller produce its normal, parameter-specific validation error.
+    """
+    try:
+        return get_type_hints(func)
+    except (NameError, AttributeError, RecursionError):
+        return {param.name: param.annotation for param in params}
+
+
 def is_chat_agent(agent: Any) -> TypeGuard[Agent]:
     """Check if the given agent is a Agent.
 

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

@@ -2,12 +2,13 @@
 
 from __future__ import annotations
 
-from typing import Any
+import inspect
+from typing import Any, TypeVar
 
 import pytest
 from pydantic import BaseModel
 
-from agent_framework import Executor, WorkflowContext, handler
+from agent_framework import Executor, WorkflowContext, handler, response_handler
 
 
 class MyTypeA(BaseModel):
@@ -22,6 +23,9 @@ class MyTypeC(BaseModel):
     pass
 
 
+_T = TypeVar("_T")
+
+
 class TestExecutorFutureAnnotations:
     """Test suite for Executor with from __future__ import annotations."""
 
@@ -109,6 +113,71 @@ async def example(self, input: str, ctx: WorkflowContext[MyTypeA | MyTypeB, MyTy
         assert spec["output_types"] == [MyTypeA, MyTypeB]
         assert spec["workflow_output_types"] == [MyTypeC]
 
+    def test_response_handler_decorator_future_annotations(self):
+        """Test @response_handler with stringified annotations and future annotations."""
+
+        class MyExecutor(Executor):
+            @handler
+            async def example(self, input: str, ctx: WorkflowContext) -> None:
+                pass
+
+            @response_handler
+            async def handle_response(
+                self, original_request: str, response: int, ctx: WorkflowContext[str, bool]
+            ) -> None:
+                pass
+
+        exec_instance = MyExecutor(id="test")
+        assert (str, int) in exec_instance._response_handlers  # pyright: ignore[reportPrivateUsage]
+        spec = exec_instance._response_handler_specs[0]  # pyright: ignore[reportPrivateUsage]
+        assert spec["request_type"] is str
+        assert spec["response_type"] is int
+        assert spec["output_types"] == [str]
+        assert spec["workflow_output_types"] == [bool]
+
+    def test_response_handler_unresolvable_annotation_raises(self):
+        """Test that an unresolvable response-handler annotation raises ValueError."""
+        with pytest.raises(ValueError, match="Response handler parameter 'ctx' must be annotated as"):
+
+            class BadResponseHandler(Executor):  # pyright: ignore[reportUnusedClass]
+                @response_handler  # pyright: ignore[reportUnknownArgumentType]
+                async def handle_response(
+                    self,
+                    original_request: NonExistentType,  # type: ignore[name-defined]  # noqa: F821
+                    response: int,
+                    ctx: WorkflowContext[MyTypeA, MyTypeB],
+                ) -> None:
+                    pass
+
+    def test_response_handler_rejects_unresolved_typevar_in_request_annotation(self):
+        """Test that response handlers reject an unresolved request TypeVar during registration."""
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+            class GenericRequestResponseExecutor(Executor):  # pyright: ignore[reportUnusedClass]
+                @response_handler  # pyright: ignore[reportUnknownArgumentType]
+                async def handle_response(self, original_request: _T, response: int, ctx: WorkflowContext) -> None:
+                    pass
+
+    def test_response_handler_rejects_unresolved_typevar_in_response_annotation(self):
+        """Test that response handlers reject an unresolved response TypeVar during registration."""
+        with pytest.raises(ValueError, match="unresolved TypeVar"):
+
+            class GenericResponseExecutor(Executor):  # pyright: ignore[reportUnusedClass]
+                @response_handler  # pyright: ignore[reportUnknownArgumentType]
+                async def handle_response(self, original_request: str, response: _T, ctx: WorkflowContext) -> None:
+                    pass
+
+    def test_annotation_resolver_falls_back_to_raw_annotations(self):
+        """Test that annotation resolution preserves raw annotations when a hint is unresolved."""
+        from agent_framework._workflows._typing_utils import _resolve_function_annotations
+
+        def sample(value: MissingType) -> None:  # type: ignore[name-defined]  # noqa: F821
+            pass
+
+        params = list(inspect.signature(sample).parameters.values())
+
+        assert _resolve_function_annotations(sample, params)["value"] == "MissingType"
+
     def test_handler_unresolvable_annotation_raises(self):
         """Test that an unresolvable forward-reference annotation raises ValueError.