From 0d086d4730f3bf56fbe3d57bb8c7304ce60377ad Mon Sep 17 00:00:00 2001 From: M-Hietala <78813398+M-Hietala@users.noreply.github.com> Date: Tue, 27 Oct 2026 12:32:48 -0500 Subject: [PATCH] Python: fix(core): accept parameterized collection interfaces in workflows --- .../_workflows/_typing_utils.py | 17 +++++++- .../core/tests/workflow/test_typing_utils.py | 36 +++++++++++++++++ .../tests/workflow/test_workflow_builder.py | 40 ++++++++++++++++++- 3 files changed, 91 insertions(+), 2 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_typing_utils.py b/python/packages/core/agent_framework/_workflows/_typing_utils.py index f0509da9a16..f8bdf97ca3e 100644 --- a/python/packages/core/agent_framework/_workflows/_typing_utils.py +++ b/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 diff --git a/python/packages/core/tests/workflow/test_typing_utils.py b/python/packages/core/tests/workflow/test_typing_utils.py index ef86137bb92..9aacc1cdf37 100644 --- a/python/packages/core/tests/workflow/test_typing_utils.py +++ b/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 diff --git a/python/packages/core/tests/workflow/test_workflow_builder.py b/python/packages/core/tests/workflow/test_workflow_builder.py index 2f92aceb77e..027fdf7d2aa 100644 --- a/python/packages/core/tests/workflow/test_workflow_builder.py +++ b/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."""