Python: fix(core): accept parameterized collection interfaces in work…

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

@@ -457,6 +457,7 @@ def is_type_compatible(source_type: type | UnionType | Any, target_type: type |
     A type is compatible if values of source_type can be assigned to variables of target_type.
     For example:
     - list[Message] is compatible with list[str | Message]
+    - list[str] is compatible with Sequence[str]
     - str is compatible with str | int
     - int is compatible with Any
 
@@ -533,8 +534,22 @@ def is_type_compatible(source_type: type | UnionType | Any, target_type: type |
         except TypeError:
             return False
 
-    # Case 5: different container types are not compatible
+    # Case 5: built-in containers can implement parameterized collection interfaces
     if source_origin != target_origin:
+        # Tuple-to-interface compatibility is not implemented. Supporting it would require
+        # handling fixed-length, variable-length (tuple[T, ...]), and empty (tuple[()]) tuples.
+        if (
+            source_origin in (list, set, frozenset, dict)
+            and isinstance(target_origin, type)
+            and target_origin.__module__ == "collections.abc"
+            and issubclass(source_origin, target_origin)
+            and len(source_args) == (2 if source_origin is dict else 1)
+        ):
+            interface_args = source_args[:1] if len(target_args) == 1 else source_args
+            return len(interface_args) == len(target_args) and all(
+                is_type_compatible(source_arg, target_arg)
+                for source_arg, target_arg in zip(interface_args, target_args, strict=False)
+            )
         return False
 
     # Case 6: same container type - check generic arguments

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

@@ -2,6 +2,8 @@
 
 import importlib
 import sys
+from collections.abc import Collection, Iterable, Mapping, MutableSequence, Sequence
+from collections.abc import Set as AbstractSet
 from dataclasses import InitVar, dataclass, make_dataclass
 from dataclasses import field as dataclass_field
 from types import ModuleType
@@ -469,6 +471,40 @@ def test_type_compatibility_bare_container() -> None:
     assert not is_type_compatible(list[str], str)
 
 
+@pytest.mark.parametrize(
+    ("source", "target", "expected"),
+    [
+        (list[str], Sequence[str], True),
+        (list[int], Sequence[str], False),
+        (list[str], Iterable[str], True),
+        (list[str], Collection[str], True),
+        (list[str], MutableSequence[str], True),
+        (list[str], Sequence[str | int], True),
+        (list[str | int], Sequence[str], False),
+        (list[Any], Sequence[str], True),
+        (list[list[str]], Sequence[Sequence[str]], True),
+        (list[list[int]], Sequence[Sequence[str]], False),
+        (dict[str, int], Mapping[str, int], True),
+        (dict[str, int], Mapping[str, str], False),
+        (dict[str, int], Mapping[int, int], False),
+        (dict[str, int], Iterable[str], True),
+        (dict[str, int], Iterable[int], False),
+        (set[int], AbstractSet[int], True),
+        (set[int], AbstractSet[str], False),
+        (frozenset[int], AbstractSet[int], True),
+        (set[int], Iterable[int], True),
+        (set[int], Sequence[int], False),
+        (list[str], Mapping[str, int], False),
+        (Sequence[str], list[str], False),
+        (Mapping[str, int], dict[str, int], False),
+        (list[str], set[str], False),
+    ],
+)
+def test_type_compatibility_parameterized_container_interfaces(source: Any, target: Any, expected: bool) -> None:
+    """Container interfaces preserve direction and validate their corresponding element types."""
+    assert is_type_compatible(source, target) is expected
+
+
 def test_type_compatibility_bare_container_and_protocol() -> None:
     """A Protocol that is not runtime_checkable is reported as incompatible, not raised."""
     from typing import Generic, Protocol, TypeVar

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

@@ -1,6 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-from collections.abc import AsyncIterator, Awaitable
+from collections.abc import AsyncIterator, Awaitable, Sequence
 from dataclasses import dataclass
 from typing import Any, Literal, overload
 
@@ -84,6 +84,44 @@ def test_builder_accepts_agents_directly():
     assert any(isinstance(e, AgentExecutor) and e.id in {"writer", "reviewer"} for e in wf.executors.values())
 
 
+async def test_workflow_builds_and_runs_with_sequence_handler() -> None:
+    """A list output can be delivered to a parameterized sequence handler."""
+
+    class Splitter(Executor):
+        @handler
+        async def split(self, text: str, ctx: WorkflowContext[list[str]]) -> None:
+            await ctx.send_message(text.split())
+
+    class Counter(Executor):
+        @handler
+        async def count(self, words: Sequence[str], ctx: WorkflowContext[None, int]) -> None:
+            await ctx.yield_output(len(words))
+
+    splitter = Splitter(id="splitter")
+    counter = Counter(id="counter")
+    workflow = WorkflowBuilder(start_executor=splitter, output_from=[counter]).add_edge(splitter, counter).build()
+
+    result = await workflow.run("one two three")
+
+    assert result.get_outputs() == [3]
+
+
+async def test_workflow_as_agent_accepts_sequence_messages() -> None:
+    """Workflow agents can deliver their message list to a sequence handler."""
+
+    class Chat(Executor):
+        @handler
+        async def chat(self, messages: Sequence[Message], ctx: WorkflowContext[None, str]) -> None:
+            await ctx.yield_output(messages[-1].text)
+
+    chat = Chat(id="chat")
+    agent = WorkflowBuilder(start_executor=chat, output_from=[chat]).build().as_agent()
+
+    response = await agent.run("hello")
+
+    assert response.text == "hello"
+
+
 @dataclass
 class MockMessage:
     """A mock message for testing purposes."""