From b6d706055fd435ffbd2b95be8efac50e7bcf3533 Mon Sep 17 00:00:00 2001
From: Nithin <kumbam.nithingoud@gmail.com>
Date: Wed, 14 Oct 2026 14:46:45 -0400
Subject: [PATCH] fix(workflows): support Literal types in edge type
 compatibility

---
 .../_workflows/_typing_utils.py               | 13 ++++++++++++
 .../core/tests/workflow/test_typing_utils.py  | 19 ++++++++++++++++++
 .../tests/workflow/test_workflow_builder.py   | 20 +++++++++++++++++++
 3 files changed, 52 insertions(+)

diff --git a/python/packages/core/agent_framework/_workflows/_typing_utils.py b/python/packages/core/agent_framework/_workflows/_typing_utils.py
index f0509da9a16..d06a84ae8fd 100644
--- a/python/packages/core/agent_framework/_workflows/_typing_utils.py
+++ b/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
diff --git a/python/packages/core/tests/workflow/test_typing_utils.py b/python/packages/core/tests/workflow/test_typing_utils.py
index ef86137bb92..9d8cc334730 100644
--- a/python/packages/core/tests/workflow/test_typing_utils.py
+++ b/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
 
 
diff --git a/python/packages/core/tests/workflow/test_workflow_builder.py b/python/packages/core/tests/workflow/test_workflow_builder.py
index 2f92aceb77e..717cb7e0c74 100644
--- a/python/packages/core/tests/workflow/test_workflow_builder.py
+++ b/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