Python: fix(workflows): support Literal types in workflow edge type compatibility

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

@@ -492,6 +492,19 @@ 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: Literal type compatibility
+    if source_origin in (Literal, typing_extensions.Literal):
+        if target_origin in (Literal, typing_extensions.Literal):
+            return set(source_args).issubset(set(target_args))
+        if target_origin is Union or target_origin is UnionType:
+            return all(
+                any(is_type_compatible(Literal[val], target_arg) for target_arg in target_args) for val in source_args
+            )
+        return all(is_type_compatible(type(val), target_type) for val in source_args)
+
+    if target_origin in (Literal, typing_extensions.Literal):
+        return False
+
     # 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

@@ -547,6 +547,25 @@ class Message:
     assert not is_type_compatible(source, incompatible_target)
 
 
+def test_type_compatibility_literals() -> None:
+    """Test type compatibility with Literal types."""
+    # Literal source to general type target
+    assert is_type_compatible(Literal["yes", "no"], str)
+    assert is_type_compatible(Literal[1, 2], int)
+    assert not is_type_compatible(Literal["yes", 1], str)
+    assert is_type_compatible(Literal["yes", 1], Union[str, int])
+    assert is_type_compatible(Literal["yes", 1], str | int)
+
+    # Literal to Literal (subset semantics)
+    assert is_type_compatible(Literal["yes"], Literal["yes", "no"])
+    assert not is_type_compatible(Literal["yes", "no"], Literal["yes"])
+    assert is_type_compatible(Literal["yes", "no"], Literal["yes", "no"])
+
+    # General type source to Literal target (rejected)
+    assert not is_type_compatible(str, Literal["yes", "no"])
+    assert not is_type_compatible(int, Literal[1, 2])
+
+
 # region: try_coerce_to_type tests
 
 

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

@@ -371,4 +371,24 @@ def test_with_output_from_with_invalid_executor_raises_validation_error():
         builder.build()
 
 
+def test_workflow_builder_literal_to_str_edge() -> None:
+    """Test building a workflow where an executor outputs a Literal and the next accepts str."""
+
+    class LiteralOutputExecutor(Executor):
+        @handler
+        async def handle(self, message: str, ctx: WorkflowContext[Literal["yes", "no"]]) -> None:
+            pass
+
+    class StringInputExecutor(Executor):
+        @handler
+        async def handle(self, message: str, ctx: WorkflowContext[str]) -> None:
+            pass
+
+    sender = LiteralOutputExecutor(id="sender")
+    receiver = StringInputExecutor(id="receiver")
+
+    workflow = WorkflowBuilder(start_executor=sender).add_edge(sender, receiver).build()
+    assert workflow is not None
+
+
 # endregion