Python: Fix executor handler type resolution when using from __future__ import annotations (#4317)

* Python: Fix Executor handler type checking with __future__ annotations (#3898)

Use typing.get_type_hints() in _validate_handler_signature to resolve
string annotations from `from __future__ import annotations`. This
mirrors the fix applied to FunctionExecutor in #2308.

When __future__ annotations are enabled, type annotations are stored as
strings. The handler decorator was passing these strings directly to
validate_workflow_context_annotation, which uses typing.get_origin and
returns None for strings, causing a ValueError.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Address PR review feedback for #3898: improve error handling and test coverage

- Wrap typing.get_type_hints() in try/except to provide a descriptive
  ValueError mentioning the handler name when annotations cannot be resolved
- Strengthen bare context test to assert output_types and workflow_output_types
- Add test for @handler(input=..., output=...) with future annotations
  covering the skip_message_annotation branch
- Add test for union-type context annotations with future annotations

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Narrow exception catch and add test for unresolvable annotations (#3898)

- Narrow except clause from bare Exception to (NameError, AttributeError,
  TypeError) to avoid masking unexpected errors.
- Add test_handler_unresolvable_annotation_raises to verify that a handler
  with a forward-reference to a non-existent type raises ValueError with
  the expected message.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Fix #3898: fall back to raw annotations when get_type_hints fails

When typing.get_type_hints(func) raises NameError (unresolvable forward
ref), AttributeError, RecursionError, or any other exception, fall back
to the raw parameter annotations instead of raising a ValueError.
This matches the suggestion from @moonbox3 on PR #4317.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Fix test to match new fallback behavior when get_type_hints fails (#3898)

The code now falls back to raw string annotations instead of raising
'Failed to resolve type annotations'. A ValueError is still raised when
the raw string ctx annotation is not a valid WorkflowContext type, so
update the test to match on ValueError without checking the message.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Apply pyupgrade: remove unnecessary string annotation quote

* Add noqa for intentionally undefined name in annotation test

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

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

@@ -6,6 +6,7 @@ import functools
 import inspect
 import logging
 import types
+import typing
 from collections.abc import Awaitable, Callable
 from typing import Any, TypeVar, overload
 
@@ -722,20 +723,30 @@ 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 Exception:
+        type_hints = {p.name: p.annotation for p in params}
+
     # Validate ctx parameter is WorkflowContext and extract type args
     ctx_param = params[2]
-    if skip_message_annotation and ctx_param.annotation == inspect.Parameter.empty:
+    ctx_annotation = type_hints.get(ctx_param.name, ctx_param.annotation)
+    if skip_message_annotation and ctx_annotation == inspect.Parameter.empty:
         # When explicit types are provided via @handler(input=..., output=...),
         # the ctx parameter doesn't need a type annotation - types come from the decorator.
         output_types: list[type[Any] | types.UnionType] = []
         workflow_output_types: list[type[Any] | types.UnionType] = []
     else:
         output_types, workflow_output_types = validate_workflow_context_annotation(
-            ctx_param.annotation, f"parameter '{ctx_param.name}'", "Handler"
+            ctx_annotation, f"parameter '{ctx_param.name}'", "Handler"
         )
 
-    message_type = message_param.annotation if message_param.annotation != inspect.Parameter.empty else None
-    ctx_annotation = ctx_param.annotation
+    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/tests/workflow/test_executor_future.py

@@ -0,0 +1,124 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from pydantic import BaseModel
+
+from agent_framework import Executor, WorkflowContext, handler
+
+
+class MyTypeA(BaseModel):
+    pass
+
+
+class MyTypeB(BaseModel):
+    pass
+
+
+class MyTypeC(BaseModel):
+    pass
+
+
+class TestExecutorFutureAnnotations:
+    """Test suite for Executor with from __future__ import annotations."""
+
+    def test_handler_decorator_future_annotations(self):
+        """Test @handler decorator works with stringified annotations (issue #3898)."""
+
+        class MyExecutor(Executor):
+            @handler
+            async def example(self, input: str, ctx: WorkflowContext[MyTypeA, MyTypeB]) -> None:
+                pass
+
+        exec_instance = MyExecutor(id="test")
+        assert str in exec_instance._handlers
+        spec = exec_instance._handler_specs[0]
+        assert spec["message_type"] is str
+        assert spec["output_types"] == [MyTypeA]
+        assert spec["workflow_output_types"] == [MyTypeB]
+
+    def test_handler_decorator_future_annotations_single_type_arg(self):
+        """Test @handler with single type argument and future annotations."""
+
+        class MyExecutor(Executor):
+            @handler
+            async def example(self, input: int, ctx: WorkflowContext[MyTypeA]) -> None:
+                pass
+
+        exec_instance = MyExecutor(id="test")
+        assert int in exec_instance._handlers
+        spec = exec_instance._handler_specs[0]
+        assert spec["message_type"] is int
+        assert spec["output_types"] == [MyTypeA]
+
+    def test_handler_decorator_future_annotations_complex(self):
+        """Test @handler with complex type annotations and future annotations."""
+
+        class MyExecutor(Executor):
+            @handler
+            async def example(self, data: dict[str, Any], ctx: WorkflowContext[list[str]]) -> None:
+                pass
+
+        exec_instance = MyExecutor(id="test")
+        spec = exec_instance._handler_specs[0]
+        assert spec["message_type"] == dict[str, Any]
+        assert spec["output_types"] == [list[str]]
+
+    def test_handler_decorator_future_annotations_bare_context(self):
+        """Test @handler with bare WorkflowContext and future annotations."""
+
+        class MyExecutor(Executor):
+            @handler
+            async def example(self, input: str, ctx: WorkflowContext) -> None:
+                pass
+
+        exec_instance = MyExecutor(id="test")
+        assert str in exec_instance._handlers
+        spec = exec_instance._handler_specs[0]
+        assert spec["output_types"] == []
+        assert spec["workflow_output_types"] == []
+
+    def test_handler_decorator_future_annotations_explicit_types(self):
+        """Test @handler with explicit type parameters under future annotations."""
+
+        class MyExecutor(Executor):
+            @handler(input=str, output=MyTypeA)
+            async def example(self, input, ctx) -> None:
+                pass
+
+        exec_instance = MyExecutor(id="test")
+        assert str in exec_instance._handlers
+        spec = exec_instance._handler_specs[0]
+        assert spec["message_type"] is str
+        assert spec["output_types"] == [MyTypeA]
+
+    def test_handler_decorator_future_annotations_union_context(self):
+        """Test @handler with union type context annotations and future annotations."""
+
+        class MyExecutor(Executor):
+            @handler
+            async def example(self, input: str, ctx: WorkflowContext[MyTypeA | MyTypeB, MyTypeC]) -> None:
+                pass
+
+        exec_instance = MyExecutor(id="test")
+        assert str in exec_instance._handlers
+        spec = exec_instance._handler_specs[0]
+        assert spec["output_types"] == [MyTypeA, MyTypeB]
+        assert spec["workflow_output_types"] == [MyTypeC]
+
+    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):
+
+            class Bad(Executor):
+                @handler
+                async def example(self, input: NonExistentType, ctx: WorkflowContext[MyTypeA, MyTypeB]) -> None:  # noqa: F821
+                    pass