diff --git a/python/packages/core/agent_framework/_workflows/_executor.py b/python/packages/core/agent_framework/_workflows/_executor.py index 39027955a7..6a748d2878 100644 --- a/python/packages/core/agent_framework/_workflows/_executor.py +++ b/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: diff --git a/python/packages/core/agent_framework/_workflows/_function_executor.py b/python/packages/core/agent_framework/_workflows/_function_executor.py index dc641697c1..5bf4ca6bed 100644 --- a/python/packages/core/agent_framework/_workflows/_function_executor.py +++ b/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 diff --git a/python/packages/core/agent_framework/_workflows/_request_info_mixin.py b/python/packages/core/agent_framework/_workflows/_request_info_mixin.py index ac7132e2fe..aee18efdf6 100644 --- a/python/packages/core/agent_framework/_workflows/_request_info_mixin.py +++ b/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=, response=) " + "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=, response=) " + "to specify explicit types." + ) return request_type, response_type, ctx_annotation, output_types, workflow_output_types diff --git a/python/packages/core/agent_framework/_workflows/_typing_utils.py b/python/packages/core/agent_framework/_workflows/_typing_utils.py index f7f1caf23f..3d0a204fcb 100644 --- a/python/packages/core/agent_framework/_workflows/_typing_utils.py +++ b/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. diff --git a/python/packages/core/tests/workflow/test_executor_future.py b/python/packages/core/tests/workflow/test_executor_future.py index 81818caed3..f65157fa22 100644 --- a/python/packages/core/tests/workflow/test_executor_future.py +++ b/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.