Python: fix(core): accept Literal source types in the workflow edge type check

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

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."""
 

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")