From b66cd3683a86f018174d60baa6a9dcb9adc1d3f4 Mon Sep 17 00:00:00 2001 From: Shanto Date: Tue, 13 Oct 2026 23:43:47 +0600 Subject: [PATCH] Accept Literal source types in the workflow edge type check --- .../_workflows/_typing_utils.py | 15 +++++++++ .../core/tests/workflow/test_typing_utils.py | 33 +++++++++++++++++++ .../core/tests/workflow/test_validation.py | 18 +++++++++- 3 files changed, 65 insertions(+), 1 deletion(-) diff --git a/python/packages/core/agent_framework/_workflows/_typing_utils.py b/python/packages/core/agent_framework/_workflows/_typing_utils.py index f0509da9a16..ad41201d04d 100644 --- a/python/packages/core/agent_framework/_workflows/_typing_utils.py +++ b/python/packages/core/agent_framework/_workflows/_typing_utils.py @@ -492,6 +492,21 @@ def is_type_compatible(source_type: type | UnionType | Any, target_type: type | if target_origin is not None and not target_args and isinstance(target_origin, type): target_type, target_origin = target_origin, None + # Case 1b: source is Literal[...] - it only holds its listed values, so it is compatible when every value fits + # the target or one of its union members: a Literal member by value (is_instance_of), any other member through + # the value's own type. + if source_origin is Literal: + targets = target_args if target_origin is Union or target_origin is UnionType else (target_type,) + return all( + any( + is_instance_of(value, target) + if get_origin(target) is Literal + else is_type_compatible(type(value), target) + for target in targets + ) + for value in cast(tuple[object, ...], source_args) + ) + # Case 2: target is Union/Optional - source is compatible if it matches any target member if target_origin is Union or target_origin is UnionType: # Special case: if source is also a Union, check that each source member diff --git a/python/packages/core/tests/workflow/test_typing_utils.py b/python/packages/core/tests/workflow/test_typing_utils.py index ef86137bb92..560020f2fc1 100644 --- a/python/packages/core/tests/workflow/test_typing_utils.py +++ b/python/packages/core/tests/workflow/test_typing_utils.py @@ -4,6 +4,7 @@ import sys from dataclasses import InitVar, dataclass, make_dataclass from dataclasses import field as dataclass_field +from enum import Enum from types import ModuleType from typing import Annotated, Any, Generic, Literal, NewType, Optional, TypeVar, Union from unittest.mock import Mock @@ -497,6 +498,38 @@ def test_type_compatibility_unions() -> None: assert not is_type_compatible(Union[str, int, bytes], Union[str, int]) +def test_type_compatibility_literals() -> None: + """Test that a Literal source is compatible when every literal value fits the target type.""" + + class Color(Enum): + RED = "red" + BLUE = "blue" + + user_id = NewType("user_id", str) + t = TypeVar("t") + + assert is_type_compatible(Literal["approve", "reject"], str) + assert is_type_compatible(Literal[1, 2], int) + assert is_type_compatible(Literal[b"x"], bytes) + assert is_type_compatible(Literal["a", 1], Union[str, int]) + assert is_type_compatible(Literal[None], Optional[str]) + assert is_type_compatible(Literal["a"], Literal["a", "b"]) + assert is_type_compatible(Literal["a"], Optional[Literal["a", "b"]]) + assert is_type_compatible(Literal["a"], Union[user_id, str]) + assert is_type_compatible(Literal[Color.RED], Color) + assert is_type_compatible(list[Literal["a", "b"]], list[str]) + + assert not is_type_compatible(Literal["a", 1], str) + assert not is_type_compatible(Literal["a", "c"], Literal["a", "b"]) + # bool literal values do not match int literal members, the same as is_instance_of + assert not is_type_compatible(Literal[True], Literal[1]) + # targets that isinstance() cannot check are handled like a plain str source, without raising + assert not is_type_compatible(Literal["a"], user_id) + assert not is_type_compatible(Literal["a"], t) + # a plain str can hold any string, so it still does not fit a Literal target + assert not is_type_compatible(str, Literal["a", "b"]) + + def test_type_compatibility_collections() -> None: """Test type compatibility with collection types.""" diff --git a/python/packages/core/tests/workflow/test_validation.py b/python/packages/core/tests/workflow/test_validation.py index 82ca127bcd0..8d96f92b994 100644 --- a/python/packages/core/tests/workflow/test_validation.py +++ b/python/packages/core/tests/workflow/test_validation.py @@ -1,7 +1,7 @@ # Copyright (c) Microsoft. All rights reserved. import logging -from typing import Any +from typing import Any, Literal import pytest @@ -150,6 +150,22 @@ def test_type_compatibility_from_any_source_passes(): assert workflow is not None +def test_type_compatibility_from_literal_source_passes(): + """An executor that sends a Literal type can connect to an executor that handles the literal values' type.""" + + class DecisionExecutor(Executor): + @handler + async def decide(self, message: str, ctx: WorkflowContext[Literal["approve", "reject"]]) -> None: + await ctx.send_message("approve") + + decision_executor = DecisionExecutor(id="decision_executor") + string_executor = StringExecutor(id="string_executor") + + workflow = WorkflowBuilder(start_executor=decision_executor).add_edge(decision_executor, string_executor).build() + + assert workflow is not None + + def test_type_compatibility_with_no_output_types(): no_output_executor = NoOutputTypesExecutor(id="no_output") string_executor = StringExecutor(id="string_executor")