Python: fix: resolve string annotations in FunctionExecutor (#2308)

* fix: resolve string annotations in FunctionExecutor

Enhance type hint validation in FunctionExecutor by importing `typing` and
using `get_type_hints` to correctly resolve annotations.

This fixes validation failures when `from __future__ import annotations`
is enabled, which stores annotations as strings.

Fixes #1808

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

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

* ran pre commit

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Gwyneth Peña-Siguenza
2025-11-19 02:26:44 -05:00
committed by GitHub
Unverified
parent 037349ff90
commit f83c39f924
2 changed files with 44 additions and 3 deletions
@@ -17,6 +17,7 @@ Design Pattern:
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 = [], []
@@ -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]]