Python: fix: resolve string annotations in FunctionExecutor

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

@@ -17,6 +17,7 @@
 
 import asyncio
 import inspect
+import typing
 from collections.abc import Awaitable, Callable
 from typing import Any, overload
 
@@ -218,15 +219,16 @@ def _validate_function_signature(func: Callable[..., Any]) -> tuple[type, Any, l
     if message_param.annotation == inspect.Parameter.empty:
         raise ValueError(f"Function instance {func.__name__} must have a type annotation for the message parameter")
 
-    message_type = message_param.annotation
+    type_hints = typing.get_type_hints(func)
+    message_type = type_hints.get(message_param.name, message_param.annotation)
 
     # Check if there's a context parameter
     if len(params) == 2:
         ctx_param = params[1]
+        ctx_annotation = type_hints.get(ctx_param.name, ctx_param.annotation)
         output_types, workflow_output_types = validate_workflow_context_annotation(
-            ctx_param.annotation, f"parameter '{ctx_param.name}'", "Function instance"
+            ctx_annotation, f"parameter '{ctx_param.name}'", "Function instance"
         )
-        ctx_annotation = ctx_param.annotation
     else:
         # No context parameter (only valid for function executors)
         output_types, workflow_output_types = [], []

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

@@ -0,0 +1,39 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+from typing import Any
+
+from agent_framework import FunctionExecutor, WorkflowContext, executor
+
+
+class TestFunctionExecutorFutureAnnotations:
+    """Test suite for FunctionExecutor with from __future__ import annotations."""
+
+    def test_executor_decorator_future_annotations(self):
+        """Test @executor decorator works with stringified annotations."""
+
+        @executor(id="future_test")
+        async def process_future(value: int, ctx: WorkflowContext[int]) -> None:
+            await ctx.send_message(value * 2)
+
+        assert isinstance(process_future, FunctionExecutor)
+        assert process_future.id == "future_test"
+        assert int in process_future._handlers
+
+        # Check spec
+        spec = process_future._handler_specs[0]
+        assert spec["message_type"] is int
+        assert spec["output_types"] == [int]
+
+    def test_executor_decorator_future_annotations_complex(self):
+        """Test @executor decorator works with complex stringified annotations."""
+
+        @executor
+        async def process_complex(data: dict[str, Any], ctx: WorkflowContext[list[str]]) -> None:
+            await ctx.send_message(["done"])
+
+        assert isinstance(process_complex, FunctionExecutor)
+        spec = process_complex._handler_specs[0]
+        assert spec["message_type"] == dict[str, Any]
+        assert spec["output_types"] == [list[str]]