Python: Define Workflow and Executor APIs (#272)

* Workflow init commit

* Add samples and clean up

* ExecutionContext -> WorkflowContext

* Address comments 1

* Fix mypy

* flatting folder structure, and rename contexts

* Remove add_loop

* Add map reduce sample, remove Activation conditions

* Add AgentExecutor and allow multiple handlers per executor

* Minor improvement

* Add RequestInfoExecutor

* Add unit tests part 1

* Address comments 2

* Pre-commit update

* Add run method and more unit tests

* Add xml docs

* run_stream -> run_streaming

* message_handler -> handler

---------

Co-authored-by: Chris <66376200+crickman@users.noreply.github.com>
Co-authored-by: Evan Mattson <evan.mattson@microsoft.com>

python/packages/main/agent_framework/workflow/__init__.py

@@ -0,0 +1,47 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import importlib
+from typing import Any
+
+PACKAGE_NAME = "agent_framework_workflow"
+PACKAGE_EXTRA = "workflow"
+_IMPORTS = [
+    "Executor",
+    "WorkflowContext",
+    "__version__",
+    "events",
+    "WorkflowBuilder",
+    "ExecutorCompletedEvent",
+    "ExecutorEvent",
+    "ExecutorInvokeEvent",
+    "RequestInfoEvent",
+    "WorkflowCompletedEvent",
+    "WorkflowEvent",
+    "WorkflowStartedEvent",
+    "AgentRunEvent",
+    "AgentRunStreamingEvent",
+    "handler",
+    "AgentExecutor",
+    "AgentExecutorRequest",
+    "AgentExecutorResponse",
+    "RequestInfoExecutor",
+    "RequestInfoMessage",
+    "WorkflowRunResult",
+    "Workflow",
+]
+
+
+def __getattr__(name: str) -> Any:
+    if name in _IMPORTS:
+        try:
+            return getattr(importlib.import_module(PACKAGE_NAME), name)
+        except ModuleNotFoundError as exc:
+            raise ModuleNotFoundError(
+                f"The '{PACKAGE_EXTRA}' extra is not installed, "
+                f"please do `pip install agent-framework[{PACKAGE_EXTRA}]`"
+            ) from exc
+    raise AttributeError(f"Module {PACKAGE_NAME} has no attribute {name}.")
+
+
+def __dir__() -> list[str]:
+    return _IMPORTS

python/packages/main/agent_framework/workflow/__init__.pyi

@@ -0,0 +1,49 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from agent_framework_workflow import (
+    AgentExecutor,
+    AgentExecutorRequest,
+    AgentExecutorResponse,
+    AgentRunEvent,
+    AgentRunStreamingEvent,
+    Executor,
+    ExecutorCompletedEvent,
+    ExecutorEvent,
+    ExecutorInvokeEvent,
+    RequestInfoEvent,
+    RequestInfoExecutor,
+    RequestInfoMessage,
+    Workflow,
+    WorkflowBuilder,
+    WorkflowCompletedEvent,
+    WorkflowContext,
+    WorkflowEvent,
+    WorkflowRunResult,
+    WorkflowStartedEvent,
+    __version__,
+    handler,
+)
+
+__all__ = [
+    "AgentExecutor",
+    "AgentExecutorRequest",
+    "AgentExecutorResponse",
+    "AgentRunEvent",
+    "AgentRunStreamingEvent",
+    "Executor",
+    "ExecutorCompletedEvent",
+    "ExecutorEvent",
+    "ExecutorInvokeEvent",
+    "RequestInfoEvent",
+    "RequestInfoExecutor",
+    "RequestInfoMessage",
+    "Workflow",
+    "WorkflowBuilder",
+    "WorkflowCompletedEvent",
+    "WorkflowContext",
+    "WorkflowEvent",
+    "WorkflowRunResult",
+    "WorkflowStartedEvent",
+    "__version__",
+    "handler",
+]

python/packages/main/pyproject.toml

@@ -38,6 +38,9 @@ azure = [
 foundry = [
     "agent-framework-foundry"
 ]
+workflow = [
+    "agent-framework-workflow"
+]
 
 [tool.uv]
 prerelease = "if-necessary-or-explicit"

python/packages/workflow/LICENSE

@@ -0,0 +1,21 @@
+    MIT License
+
+    Copyright (c) Microsoft Corporation.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"), to deal
+    in the Software without restriction, including without limitation the rights
+    to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+    copies of the Software, and to permit persons to whom the Software is
+    furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included in all
+    copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+    AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+    OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+    SOFTWARE

python/packages/workflow/README.md

@@ -0,0 +1,9 @@
+# Get Started with Microsoft Agent Framework Workflow
+
+Please install this package as the extra for `agent-framework`:
+
+```bash
+pip install agent-framework[workflow]
+```
+
+and see the [README](https://github.com/microsoft/agent-framework/tree/main/python/README.md) for more information.

python/packages/workflow/agent_framework_workflow/__init__.py

@@ -0,0 +1,72 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import importlib.metadata
+
+from ._events import (
+    AgentRunEvent,
+    AgentRunStreamingEvent,
+    ExecutorCompletedEvent,
+    ExecutorEvent,
+    ExecutorInvokeEvent,
+    RequestInfoEvent,
+    WorkflowCompletedEvent,
+    WorkflowEvent,
+    WorkflowStartedEvent,
+)
+from ._executor import (
+    AgentExecutor,
+    AgentExecutorRequest,
+    AgentExecutorResponse,
+    Executor,
+    RequestInfoExecutor,
+    RequestInfoMessage,
+    handler,
+)
+from ._validation import (
+    EdgeDuplicationError,
+    GraphConnectivityError,
+    TypeCompatibilityError,
+    ValidationTypeEnum,
+    WorkflowValidationError,
+    validate_workflow_graph,
+)
+from ._workflow import Workflow, WorkflowBuilder, WorkflowRunResult
+from ._workflow_context import WorkflowContext
+
+try:
+    __version__ = importlib.metadata.version(__name__)
+except importlib.metadata.PackageNotFoundError:
+    __version__ = "0.0.0"  # Fallback for development mode
+
+
+__all__ = [
+    "AgentExecutor",
+    "AgentExecutorRequest",
+    "AgentExecutorResponse",
+    "AgentRunEvent",
+    "AgentRunStreamingEvent",
+    "EdgeDuplicationError",
+    "Executor",
+    "ExecutorCompletedEvent",
+    "ExecutorEvent",
+    "ExecutorInvokeEvent",
+    "GraphConnectivityError",
+    "RequestInfoEvent",
+    "RequestInfoEvent",
+    "RequestInfoExecutor",
+    "RequestInfoExecutor",
+    "RequestInfoMessage",
+    "TypeCompatibilityError",
+    "ValidationTypeEnum",
+    "Workflow",
+    "WorkflowBuilder",
+    "WorkflowCompletedEvent",
+    "WorkflowContext",
+    "WorkflowEvent",
+    "WorkflowRunResult",
+    "WorkflowStartedEvent",
+    "WorkflowValidationError",
+    "__version__",
+    "handler",
+    "validate_workflow_graph",
+]

python/packages/workflow/agent_framework_workflow/_edge.py

@@ -0,0 +1,154 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+from collections.abc import Callable
+from typing import Any, ClassVar
+
+from ._executor import Executor
+from ._runner_context import Message, RunnerContext
+from ._shared_state import SharedState
+from ._workflow_context import WorkflowContext
+
+
+class Edge:
+    """Represents a directed edge in a graph."""
+
+    ID_SEPARATOR: ClassVar[str] = "->"
+
+    def __init__(
+        self,
+        source: Executor,
+        target: Executor,
+        condition: Callable[[Any], bool] | None = None,
+    ) -> None:
+        """Initialize the edge with a source and target node.
+
+        Args:
+            source (Executor): The source executor of the edge.
+            target (Executor): The target executor of the edge.
+            condition (Callable[[Any], bool], optional): A condition function that determines
+                if the edge can handle the data. If None, the edge can handle any data type.
+                Defaults to None.
+        """
+        self.source = source
+        self.target = target
+        self._condition = condition
+
+        # Edge group is used to group edges that share the same target executor.
+        # It allows for sending messages to the target executor only when all edges in the group have data.
+        self._edge_group_ids: list[str] = []
+
+    @property
+    def source_id(self) -> str:
+        """Get the source executor ID."""
+        return self.source.id
+
+    @property
+    def target_id(self) -> str:
+        """Get the target executor ID."""
+        return self.target.id
+
+    @property
+    def id(self) -> str:
+        """Get the unique ID of the edge."""
+        return f"{self.source_id}{self.ID_SEPARATOR}{self.target_id}"
+
+    def has_edge_group(self) -> bool:
+        """Check if the edge is part of an edge group."""
+        return bool(self._edge_group_ids)
+
+    @classmethod
+    def source_and_target_from_id(cls, edge_id: str) -> tuple[str, str]:
+        """Extract the source and target IDs from the edge ID.
+
+        Args:
+            edge_id (str): The edge ID in the format "source_id->target_id".
+
+        Returns:
+            tuple[str, str]: A tuple containing the source ID and target ID.
+        """
+        if cls.ID_SEPARATOR not in edge_id:
+            raise ValueError(f"Invalid edge ID format: {edge_id}")
+        ids = edge_id.split(cls.ID_SEPARATOR)
+        if len(ids) != 2:
+            raise ValueError(f"Invalid edge ID format: {edge_id}")
+        return ids[0], ids[1]
+
+    def can_handle(self, message_data: Any) -> bool:
+        """Check if the edge can handle the given data.
+
+        Args:
+            message_data (Any): The data to check.
+
+        Returns:
+            bool: True if the edge can handle the data, False otherwise.
+        """
+        if not self._edge_group_ids:
+            return self.target.can_handle(message_data)
+
+        # If the edge is part of an edge group, the target should expect a list of the data type.
+        return self.target.can_handle([message_data])
+
+    async def send_message(self, message: Message, shared_state: SharedState, ctx: RunnerContext) -> None:
+        """Send a message along this edge.
+
+        Args:
+            message (Message): The message to send.
+            shared_state (SharedState): The shared state to use for holding data.
+            ctx (RunnerContext): The context for the runner.
+        """
+        if not self.can_handle(message.data):
+            raise RuntimeError(f"Edge {self.id} cannot handle data of type {type(message.data)}.")
+
+        if not self._edge_group_ids and self._should_route(message.data):
+            await self.target.execute(
+                message.data, WorkflowContext(self.target.id, [self.source.id], shared_state, ctx)
+            )
+        elif self._edge_group_ids:
+            # Logic:
+            # 1. If not all edges in the edge group have data in the shared state,
+            #    add the data to the shared state.
+            # 2. If all edges in the edge group have data in the shared state,
+            #    copy the data to a list and send it to the target executor.
+            message_list: list[Message] = []
+            async with shared_state.hold() as held_shared_state:
+                has_data = await asyncio.gather(
+                    *(held_shared_state.has_within_hold(edge_id) for edge_id in self._edge_group_ids)
+                )
+                if not all(has_data):
+                    await held_shared_state.set_within_hold(self.id, message)
+                else:
+                    message_list = [
+                        await held_shared_state.get_within_hold(edge_id) for edge_id in self._edge_group_ids
+                    ] + [message]
+                    # Remove the data from the shared state after retrieving it
+                    await asyncio.gather(
+                        *(held_shared_state.delete_within_hold(edge_id) for edge_id in self._edge_group_ids)
+                    )
+
+            if message_list:
+                data_list = [msg.data for msg in message_list]
+                source_ids = [msg.source_id for msg in message_list]
+                await self.target.execute(data_list, WorkflowContext(self.target.id, source_ids, shared_state, ctx))
+
+    def _should_route(self, data: Any) -> bool:
+        """Determine if message should be routed through this edge."""
+        if self._condition is None:
+            return True
+
+        return self._condition(data)
+
+    def set_edge_group(self, edge_group_ids: list[str]) -> None:
+        """Set the edge group IDs for this edge.
+
+        Args:
+            edge_group_ids (list[str]): A list of edge IDs that belong to the same edge group.
+        """
+        # Validate that the edges in the edge group contain the same target executor as this edge
+        # TODO(@taochen): An edge cannot be part of multiple edge groups.
+        # TODO(@taochen): Can an edge have both a condition and an edge group?
+        if edge_group_ids:
+            for edge_id in edge_group_ids:
+                if Edge.source_and_target_from_id(edge_id)[1] != self.target.id:
+                    raise ValueError("All edges in the group must have the same target executor.")
+        self._edge_group_ids = edge_group_ids

python/packages/workflow/agent_framework_workflow/_events.py

@@ -0,0 +1,140 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from typing import Any
+
+from agent_framework import AgentRunResponse, AgentRunResponseUpdate
+
+
+class WorkflowEvent:
+    """Base class for workflow events."""
+
+    def __init__(self, data: Any | None = None):
+        """Initialize the workflow event with optional data."""
+        self.data = data
+
+    def __repr__(self):
+        """Return a string representation of the workflow event."""
+        return f"{self.__class__.__name__}(data={self.data if self.data is not None else 'None'})"
+
+
+class WorkflowStartedEvent(WorkflowEvent):
+    """Event triggered when a workflow starts."""
+
+    ...
+
+
+class WorkflowCompletedEvent(WorkflowEvent):
+    """Event triggered when a workflow completes."""
+
+    ...
+
+
+class WorkflowWarningEvent(WorkflowEvent):
+    """Event triggered when a warning occurs in the workflow."""
+
+    def __init__(self, data: str):
+        """Initialize the workflow warning event with optional data and warning message."""
+        super().__init__(data)
+
+    def __repr__(self):
+        """Return a string representation of the workflow warning event."""
+        return f"{self.__class__.__name__}(message={self.data})"
+
+
+class WorkflowErrorEvent(WorkflowEvent):
+    """Event triggered when an error occurs in the workflow."""
+
+    def __init__(self, data: Exception):
+        """Initialize the workflow error event with optional data and error message."""
+        super().__init__(data)
+
+    def __repr__(self):
+        """Return a string representation of the workflow error event."""
+        return f"{self.__class__.__name__}(exception={self.data})"
+
+
+class RequestInfoEvent(WorkflowEvent):
+    """Event triggered when a workflow executor requests external information."""
+
+    def __init__(
+        self,
+        request_id: str,
+        source_executor_id: str,
+        request_type: type,
+        request_data: Any,
+    ):
+        """Initialize the request info event.
+
+        Args:
+            request_id: Unique identifier for the request.
+            source_executor_id: ID of the executor that made the request.
+            request_type: Type of the request (e.g., a specific data type).
+            request_data: The data associated with the request.
+        """
+        super().__init__(request_data)
+        self.request_id = request_id
+        self.source_executor_id = source_executor_id
+        self.request_type = request_type
+
+    def __repr__(self):
+        """Return a string representation of the request info event."""
+        return (
+            f"{self.__class__.__name__}("
+            f"request_id={self.request_id}, "
+            f"source_executor_id={self.source_executor_id}, "
+            f"request_type={self.request_type.__name__}, "
+            f"data={self.data})"
+        )
+
+
+class ExecutorEvent(WorkflowEvent):
+    """Base class for executor events."""
+
+    def __init__(self, executor_id: str, data: Any | None = None):
+        """Initialize the executor event with an executor ID and optional data."""
+        super().__init__(data)
+        self.executor_id = executor_id
+
+    def __repr__(self):
+        """Return a string representation of the executor event."""
+        return f"{self.__class__.__name__}(executor_id={self.executor_id}, data={self.data})"
+
+
+class ExecutorInvokeEvent(ExecutorEvent):
+    """Event triggered when an executor handler is invoked."""
+
+    def __repr__(self):
+        """Return a string representation of the executor handler invoke event."""
+        return f"{self.__class__.__name__}(executor_id={self.executor_id})"
+
+
+class ExecutorCompletedEvent(ExecutorEvent):
+    """Event triggered when an executor handler is completed."""
+
+    def __repr__(self):
+        """Return a string representation of the executor handler complete event."""
+        return f"{self.__class__.__name__}(executor_id={self.executor_id})"
+
+
+class AgentRunStreamingEvent(ExecutorEvent):
+    """Event triggered when an agent is streaming messages."""
+
+    def __init__(self, executor_id: str, data: AgentRunResponseUpdate | None = None):
+        """Initialize the agent streaming event."""
+        super().__init__(executor_id, data)
+
+    def __repr__(self):
+        """Return a string representation of the agent streaming event."""
+        return f"{self.__class__.__name__}(executor_id={self.executor_id}, messages={self.data})"
+
+
+class AgentRunEvent(ExecutorEvent):
+    """Event triggered when an agent run is completed."""
+
+    def __init__(self, executor_id: str, data: AgentRunResponse | None = None):
+        """Initialize the agent run event."""
+        super().__init__(executor_id, data)
+
+    def __repr__(self):
+        """Return a string representation of the agent run event."""
+        return f"{self.__class__.__name__}(executor_id={self.executor_id}, data={self.data})"

python/packages/workflow/agent_framework_workflow/_executor.py

@@ -0,0 +1,336 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import functools
+import inspect
+import uuid
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from typing import Any, ClassVar, TypeVar, overload
+
+from agent_framework import AgentRunResponse, AgentRunResponseUpdate, AgentThread, AIAgent, ChatMessage
+
+from ._events import (
+    AgentRunEvent,
+    AgentRunStreamingEvent,
+    ExecutorCompletedEvent,
+    ExecutorInvokeEvent,
+    RequestInfoEvent,
+)
+from ._typing_utils import is_instance_of
+from ._workflow_context import WorkflowContext
+
+# region: Executor
+
+
+class Executor:
+    """An executor is a component that processes messages in a workflow."""
+
+    def __init__(self, id: str | None = None) -> None:
+        """Initialize the executor with a unique identifier.
+
+        Args:
+            id: A unique identifier for the executor. If None, a new UUID will be generated.
+        """
+        self._id = id or str(uuid.uuid4())
+
+        self._handlers: dict[type, Callable[[Any, WorkflowContext], Any]] = {}
+        self._discover_handlers()
+
+        if not self._handlers:
+            raise ValueError(
+                f"Executor {self.__class__.__name__} has no handlers defined. "
+                "Please define at least one handler using the @handler decorator."
+            )
+
+    async def execute(self, message: Any, context: WorkflowContext) -> None:
+        """Execute the executor with a given message and context.
+
+        Args:
+            message: The message to be processed by the executor.
+            context: The workflow context in which the executor operates.
+
+        Returns:
+            An awaitable that resolves to the result of the execution.
+        """
+        handler: Callable[[Any, WorkflowContext], Any] | None = None
+        for message_type in self._handlers:
+            if is_instance_of(message, message_type):
+                handler = self._handlers[message_type]
+                break
+
+        if handler is None:
+            raise RuntimeError(f"Executor {self.__class__.__name__} cannot handle message of type {type(message)}.")
+
+        await context.add_event(ExecutorInvokeEvent(self.id))
+        await handler(message, context)
+        await context.add_event(ExecutorCompletedEvent(self.id))
+
+    @property
+    def id(self) -> str:
+        """Get the unique identifier of the executor."""
+        return self._id
+
+    def _discover_handlers(self) -> None:
+        """Discover message handlers in the executor class."""
+        for attr_name in dir(self):
+            attr = getattr(self, attr_name)
+            if callable(attr) and hasattr(attr, "_handler_spec"):
+                handler_spec = attr._handler_spec  # type: ignore
+                if self._handlers.get(handler_spec["message_type"]) is not None:
+                    raise ValueError(
+                        f"Duplicate handler for type {handler_spec['message_type']} in {self.__class__.__name__}"
+                    )
+                self._handlers[handler_spec["message_type"]] = attr
+
+    def can_handle(self, message: Any) -> bool:
+        """Check if the executor can handle a given message type.
+
+        Args:
+            message: The message to check.
+
+        Returns:
+            True if the executor can handle the message type, False otherwise.
+        """
+        return any(is_instance_of(message, message_type) for message_type in self._handlers)
+
+
+# endregion: Executor
+
+# region: Handler Decorator
+
+
+ExecutorT = TypeVar("ExecutorT", bound="Executor")
+
+
+@overload
+def handler(
+    func: Callable[[ExecutorT, Any, WorkflowContext], Awaitable[Any]],
+) -> Callable[[ExecutorT, Any, WorkflowContext], Awaitable[Any]]: ...
+
+
+@overload
+def handler(
+    func: None = None,
+    *,
+    output_types: list[type] | None = None,
+) -> Callable[
+    [Callable[[ExecutorT, Any, WorkflowContext], Awaitable[Any]]],
+    Callable[[ExecutorT, Any, WorkflowContext], Awaitable[Any]],
+]: ...
+
+
+def handler(
+    func: Callable[[ExecutorT, Any, WorkflowContext], Awaitable[Any]] | None = None,
+    *,
+    output_types: list[type] | None = None,
+) -> (
+    Callable[[ExecutorT, Any, WorkflowContext], Awaitable[Any]]
+    | Callable[
+        [Callable[[ExecutorT, Any, WorkflowContext], Awaitable[Any]]],
+        Callable[[ExecutorT, Any, WorkflowContext], Awaitable[Any]],
+    ]
+):
+    """Decorator to register a handler for an executor.
+
+    Args:
+        func: The function to decorate. Can be None when using with parameters.
+        output_types: Optional list of message types this handler can emit.
+
+    Returns:
+        The decorated function with handler metadata.
+
+    Example:
+        @handler
+        async def handle_string(self, message: str, ctx: WorkflowContext) -> None:
+            ...
+
+        @handler(output_types=[str, int])
+        async def handle_data(self, message: dict, ctx: WorkflowContext) -> None:
+            ...
+    """
+
+    def decorator(
+        func: Callable[[ExecutorT, Any, WorkflowContext], Awaitable[Any]],
+    ) -> Callable[[ExecutorT, Any, WorkflowContext], Awaitable[Any]]:
+        # Extract the message type from a handler function.
+        sig = inspect.signature(func)
+        params = list(sig.parameters.values())
+
+        if len(params) != 3:  # self, message, ctx
+            raise ValueError(f"Handler must have exactly 3 parameters, got {len(params)}")
+
+        message_type = params[1].annotation
+        if message_type is inspect.Parameter.empty:
+            raise ValueError("Handler's second parameter must have a type annotation")
+
+        @functools.wraps(func)
+        async def wrapper(self: ExecutorT, message: Any, ctx: WorkflowContext) -> Any:
+            """Wrapper function to call the handler."""
+            return await func(self, message, ctx)
+
+        wrapper._handler_spec = {  # type: ignore
+            "name": func.__name__,
+            "message_type": message_type,
+            "output_types": output_types or [],
+        }
+
+        return wrapper
+
+    if func is None:
+        return decorator
+    return decorator(func)
+
+
+# endregion: Handler Decorator
+
+# region: Agent Executor
+
+
+@dataclass
+class AgentExecutorRequest:
+    """A request to an agent executor.
+
+    Attributes:
+        messages: A list of chat messages to be processed by the agent.
+        should_respond: A flag indicating whether the agent should respond to the messages.
+            If False, the messages will be saved to the executor's cache but not sent to the agent.
+    """
+
+    messages: list[ChatMessage]
+    should_respond: bool = True
+
+
+@dataclass
+class AgentExecutorResponse:
+    """A response from an agent executor.
+
+    Attributes:
+        executor_id: The ID of the executor that generated the response.
+        response: The agent run response containing the messages generated by the agent.
+    """
+
+    executor_id: str
+    agent_run_response: AgentRunResponse
+
+
+class AgentExecutor(Executor):
+    """built-in executor that wraps an agent for handling messages."""
+
+    def __init__(
+        self,
+        agent: AIAgent,
+        *,
+        agent_thread: AgentThread | None = None,
+        streaming: bool = False,
+        id: str | None = None,
+    ):
+        """Initialize the executor with a unique identifier.
+
+        Args:
+            agent: The agent to be wrapped by this executor.
+            agent_thread: The thread to use for running the agent. If None, a new thread will be created.
+            streaming: Whether to enable streaming for the agent. If enabled, the executor will emit
+                AgentRunStreamingEvent updates instead of a single AgentRunEvent.
+            id: A unique identifier for the executor. If None, a new UUID will be generated.
+        """
+        super().__init__(id or agent.id)
+        self._agent = agent
+        self._agent_thread = agent_thread or self._agent.get_new_thread()
+        self._streaming = streaming
+        self._cache: list[ChatMessage] = []
+
+    @handler(output_types=[AgentExecutorResponse])
+    async def run(self, request: AgentExecutorRequest, ctx: WorkflowContext) -> None:
+        """Run the agent executor with the given request."""
+        self._cache.extend(request.messages)
+
+        if request.should_respond:
+            if self._streaming:
+                updates: list[AgentRunResponseUpdate] = []
+                async for update in self._agent.run_streaming(
+                    self._cache,
+                    thread=self._agent_thread,
+                ):
+                    updates.append(update)
+                    await ctx.add_event(AgentRunStreamingEvent(self.id, update))
+                response = AgentRunResponse.from_agent_run_response_updates(updates)
+            else:
+                response = await self._agent.run(
+                    self._cache,
+                    thread=self._agent_thread,
+                )
+                await ctx.add_event(AgentRunEvent(self.id, response))
+
+            await ctx.send_message(AgentExecutorResponse(self.id, response))
+            self._cache.clear()
+
+
+# endregion: Agent Executor
+
+
+# region: Request Info Executor
+
+
+@dataclass
+class RequestInfoMessage:
+    """Base class for all request messages in workflows.
+
+    Any message that should be routed to the RequestInfoExecutor for external
+    handling must inherit from this class. This ensures type safety and makes
+    the request/response pattern explicit.
+    """
+
+    request_id: str = str(uuid.uuid4())
+
+
+class RequestInfoExecutor(Executor):
+    """Built-in executor that handles request/response patterns in workflows.
+
+    This executor acts as a gateway for external information requests. When it receives
+    a request message, it saves the request details and emits a RequestInfoEvent. When
+    a response is provided externally, it emits the response as a message.
+    """
+
+    # Well-known ID for the request info executor
+    EXECUTOR_ID: ClassVar[str] = "request_info"
+
+    def __init__(self):
+        """Initialize the RequestInfoExecutor with its well-known ID."""
+        super().__init__(id=self.EXECUTOR_ID)
+        self._request_events: dict[str, RequestInfoEvent] = {}
+
+    @handler
+    async def run(self, message: RequestInfoMessage, ctx: WorkflowContext) -> None:
+        """Run the RequestInfoExecutor with the given message."""
+        source_executor_id = ctx.get_source_executor_id()
+
+        event = RequestInfoEvent(
+            request_id=message.request_id,
+            source_executor_id=source_executor_id,
+            request_type=type(message),
+            request_data=message,
+        )
+        self._request_events[message.request_id] = event
+        await ctx.add_event(event)
+
+    async def handle_response(
+        self,
+        response_data: Any,
+        request_id: str,
+        ctx: WorkflowContext,
+    ) -> None:
+        """Handle a response to a request.
+
+        Args:
+            request_id: The ID of the request to which this response corresponds.
+            response_data: The data returned in the response.
+            ctx: The workflow context for sending the response.
+        """
+        if request_id not in self._request_events:
+            raise ValueError(f"No request found with ID: {request_id}")
+
+        event = self._request_events.pop(request_id)
+        await ctx.send_message(response_data, target_id=event.source_executor_id)
+
+
+# endregion: Request Info Executor

python/packages/workflow/agent_framework_workflow/_runner.py

@@ -0,0 +1,120 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import logging
+from collections import defaultdict
+from collections.abc import AsyncIterable
+
+from ._edge import Edge
+from ._events import WorkflowEvent
+from ._runner_context import Message, RunnerContext
+from ._shared_state import SharedState
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_MAX_ITERATIONS = 100
+
+
+class Runner:
+    """A class to run a workflow in Pregel supersteps."""
+
+    def __init__(
+        self,
+        edges: list[Edge],
+        shared_state: SharedState,
+        ctx: RunnerContext,
+        max_iterations: int = DEFAULT_MAX_ITERATIONS,
+    ) -> None:
+        """Initialize the runner with edges, shared state, and context.
+
+        Args:
+            edges: The edges of the workflow.
+            shared_state: The shared state for the workflow.
+            ctx: The runner context for the workflow.
+            max_iterations: The maximum number of iterations to run.
+        """
+        self._edge_map = self._parse_edges(edges)
+        self._ctx = ctx
+        self._iteration = 0
+        self._max_iterations = max_iterations
+        self._shared_state = shared_state
+        self._is_running = False
+
+    @property
+    def context(self) -> RunnerContext:
+        """Get the workflow context."""
+        return self._ctx
+
+    async def run_until_convergence(self) -> AsyncIterable[WorkflowEvent]:
+        """Run the workflow until no more messages are sent."""
+        try:
+            if self._is_running:
+                raise RuntimeError("Runner is already running.")
+            self._is_running = True
+            while self._iteration < self._max_iterations:
+                await self._run_iteration()
+                self._iteration += 1
+
+                if await self._ctx.has_events():
+                    events = await self._ctx.drain_events()
+                    for event in events:
+                        yield event
+
+                if not await self._ctx.has_messages():
+                    break
+            else:
+                raise RuntimeError(f"Runner did not converge after {self._max_iterations} iterations.")
+        finally:
+            self._is_running = False
+            self._iteration = 0
+
+    async def _run_iteration(self):
+        """Run a superstep of the workflow execution."""
+
+        async def _deliver_messages(source_executor_id: str, messages: list[Message]) -> None:
+            """Deliver messages to the executors.
+
+            Outer loop to concurrently deliver messages from all sources to their targets.
+            """
+
+            async def _deliver_messages_inner(
+                edge: Edge,
+                messages: list[Message],
+            ) -> None:
+                """Deliver messages to a specific target executor.
+
+                Inner loop to deliver messages to a specific target executor.
+                """
+                for message in messages:
+                    if message.target_id is not None and message.target_id != edge.target_id:
+                        continue
+
+                    if not edge.can_handle(message.data):
+                        continue
+
+                    await edge.send_message(message, self._shared_state, self._ctx)
+
+            associated_edges = self._edge_map.get(source_executor_id, [])
+            tasks = [asyncio.create_task(_deliver_messages_inner(edge, messages)) for edge in associated_edges]
+            await asyncio.gather(*tasks)
+
+        messages = await self._ctx.drain_messages()
+        tasks = [
+            asyncio.create_task(_deliver_messages(source_executor_id, messages))
+            for source_executor_id, messages in messages.items()
+        ]
+        await asyncio.gather(*tasks)
+
+    def _parse_edges(self, edges: list[Edge]) -> dict[str, list[Edge]]:
+        """Parse the edges of the workflow into a more convenient format.
+
+        Args:
+            edges: A list of edges in the workflow.
+
+        Returns:
+            A dictionary mapping each source executor ID to a list of target executor IDs.
+        """
+        parsed: defaultdict[str, list[Edge]] = defaultdict(list)
+        for edge in edges:
+            parsed[edge.source_id].append(edge)
+        return parsed

python/packages/workflow/agent_framework_workflow/_runner_context.py

@@ -0,0 +1,115 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import logging
+from collections import defaultdict
+from dataclasses import dataclass
+from typing import Any, Protocol, TypeVar, runtime_checkable
+
+from ._events import WorkflowEvent
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+@dataclass
+class Message:
+    """A class representing a message in the workflow."""
+
+    data: Any
+    source_id: str
+    target_id: str | None = None
+
+
+@runtime_checkable
+class RunnerContext(Protocol):
+    """Protocol for the execution context used by the runner."""
+
+    async def send_message(self, message: Message) -> None:
+        """Send a message from the executor to the context.
+
+        Args:
+            message: The message to be sent.
+        """
+        ...
+
+    async def drain_messages(self) -> dict[str, list[Message]]:
+        """Drain all messages from the context.
+
+        Returns:
+            A dictionary mapping executor IDs to lists of messages.
+        """
+        ...
+
+    async def has_messages(self) -> bool:
+        """Check if there are any messages in the context.
+
+        Returns:
+            True if there are messages, False otherwise.
+        """
+        ...
+
+    async def add_event(self, event: WorkflowEvent) -> None:
+        """Add an event to the execution context.
+
+        Args:
+            event: The event to be added.
+        """
+        ...
+
+    async def drain_events(self) -> list[WorkflowEvent]:
+        """Drain all events from the context.
+
+        Returns:
+            A list of events that were added to the context.
+        """
+        ...
+
+    async def has_events(self) -> bool:
+        """Check if there are any events in the context.
+
+        Returns:
+            True if there are events, False otherwise.
+        """
+        ...
+
+
+class InProcRunnerContext(RunnerContext):
+    """In-process execution context for local execution of workflows."""
+
+    def __init__(self):
+        """Initialize the in-process execution context."""
+        self._messages: defaultdict[str, list[Message]] = defaultdict(list)
+        self._events: list[WorkflowEvent] = []
+
+    async def send_message(self, message: Message) -> None:
+        """Send a message from the executor to the context."""
+        self._messages[message.source_id].append(message)
+
+    async def drain_messages(self) -> dict[str, list[Message]]:
+        """Drain all messages from the context."""
+        messages = dict(self._messages)
+        self._messages.clear()
+        return messages
+
+    async def has_messages(self) -> bool:
+        """Check if there are any messages in the context."""
+        return bool(self._messages)
+
+    async def add_event(self, event: WorkflowEvent) -> None:
+        """Add an event to the execution context.
+
+        Args:
+            event: The event to be added.
+        """
+        self._events.append(event)
+
+    async def drain_events(self) -> list[WorkflowEvent]:
+        """Drain all events from the context."""
+        events = self._events.copy()
+        self._events.clear()
+        return events
+
+    async def has_events(self) -> bool:
+        """Check if there are any events in the context."""
+        return bool(self._events)

python/packages/workflow/agent_framework_workflow/_shared_state.py

@@ -0,0 +1,68 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+from contextlib import asynccontextmanager
+from typing import Any
+
+
+class SharedState:
+    """A class to manage shared state in a workflow."""
+
+    def __init__(self) -> None:
+        """Initialize the shared state."""
+        self._state: dict[str, Any] = {}
+        self._shared_state_lock = asyncio.Lock()
+
+    async def set(self, key: str, value: Any) -> None:
+        """Set a value in the shared state."""
+        async with self._shared_state_lock:
+            await self.set_within_hold(key, value)
+
+    async def get(self, key: str) -> Any:
+        """Get a value from the shared state."""
+        async with self._shared_state_lock:
+            return await self.get_within_hold(key)
+
+    async def has(self, key: str) -> bool:
+        """Check if a key exists in the shared state."""
+        async with self._shared_state_lock:
+            return await self.has_within_hold(key)
+
+    async def delete(self, key: str) -> None:
+        """Delete a key from the shared state."""
+        async with self._shared_state_lock:
+            await self.delete_within_hold(key)
+
+    @asynccontextmanager
+    async def hold(self):
+        """Context manager to hold the shared state lock for multiple operations.
+
+        Usage:
+            async with shared_state.hold():
+                await shared_state.set_within_hold("key", value)
+                value = await shared_state.get_within_hold("key")
+        """
+        async with self._shared_state_lock:
+            yield self
+
+    # Unsafe methods that don't acquire locks (for use within hold() context)
+    async def set_within_hold(self, key: str, value: Any) -> None:
+        """Set a value without acquiring the lock (unsafe - use within hold() context)."""
+        self._state[key] = value
+
+    async def get_within_hold(self, key: str) -> Any:
+        """Get a value without acquiring the lock (unsafe - use within hold() context)."""
+        if key not in self._state:
+            raise KeyError(f"Key '{key}' not found in shared state.")
+        return self._state[key]
+
+    async def has_within_hold(self, key: str) -> bool:
+        """Check if a key exists without acquiring the lock (unsafe - use within hold() context)."""
+        return key in self._state
+
+    async def delete_within_hold(self, key: str) -> None:
+        """Delete a key without acquiring the lock (unsafe - use within hold() context)."""
+        if key in self._state:
+            del self._state[key]
+        else:
+            raise KeyError(f"Key '{key}' not found in shared state.")

python/packages/workflow/agent_framework_workflow/_typing_utils.py

@@ -0,0 +1,50 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from typing import Any, Union, get_args, get_origin
+
+
+def is_instance_of(data: Any, target_type: type) -> bool:
+    """Check if the data is an instance of the target type.
+
+    Args:
+        data (Any): The data to check.
+        target_type (type): The type to check against.
+
+    Returns:
+        bool: True if data is an instance of target_type, False otherwise.
+    """
+    origin = get_origin(target_type)
+    args = get_args(target_type)
+
+    # Case 1: origin is None, meaning target_type is not a generic type
+    if origin is None:
+        return isinstance(data, target_type)
+
+    # Case 2: target_type is Optional[T] or Union[T1, T2, ...]
+    # Optional[T] is really just as Union[T, None]
+    if origin is Union:
+        return any(is_instance_of(data, arg) for arg in args)
+
+    # Case 3: target_type is a generic type
+    if origin in [list, set]:
+        return isinstance(data, origin) and all(is_instance_of(item, args[0]) for item in data)  # type: ignore
+
+    # Case 4: target_type is a tuple
+    if origin is tuple:
+        if len(args) == 1 and args[0] is Ellipsis:  # Tuple[...] case
+            return isinstance(data, tuple)
+        return (
+            isinstance(data, tuple)
+            and len(data) == len(args)  # type: ignore
+            and all(is_instance_of(item, arg) for item, arg in zip(data, args, strict=False))  # type: ignore
+        )
+
+    # Case 5: target_type is a dict
+    if origin is dict:
+        return isinstance(data, dict) and all(
+            is_instance_of(key, args[0]) and is_instance_of(value, args[1])
+            for key, value in data.items()  # type: ignore
+        )
+
+    # Fallback: if we reach here, we assume data is an instance of the target_type
+    return isinstance(data, target_type)

python/packages/workflow/agent_framework_workflow/_validation.py

@@ -0,0 +1,493 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import inspect
+import logging
+from collections import defaultdict
+from enum import Enum
+from typing import Any, Union, get_args, get_origin
+
+from ._edge import Edge
+from ._executor import Executor
+
+logger = logging.getLogger(__name__)
+
+
+# region Enums and Base Classes
+class ValidationTypeEnum(Enum):
+    """Enumeration of workflow validation types."""
+
+    EDGE_DUPLICATION = "EDGE_DUPLICATION"
+    TYPE_COMPATIBILITY = "TYPE_COMPATIBILITY"
+    GRAPH_CONNECTIVITY = "GRAPH_CONNECTIVITY"
+
+
+class WorkflowValidationError(Exception):
+    """Base exception for workflow validation errors."""
+
+    def __init__(self, message: str, validation_type: ValidationTypeEnum):
+        super().__init__(message)
+        self.message = message
+        self.validation_type = validation_type
+
+    def __str__(self) -> str:
+        return f"[{self.validation_type.value}] {self.message}"
+
+
+class EdgeDuplicationError(WorkflowValidationError):
+    """Exception raised when duplicate edges are detected in the workflow."""
+
+    def __init__(self, edge_id: str):
+        super().__init__(
+            message=f"Duplicate edge detected: {edge_id}. Each edge in the workflow must be unique.",
+            validation_type=ValidationTypeEnum.EDGE_DUPLICATION,
+        )
+        self.edge_id = edge_id
+
+
+class TypeCompatibilityError(WorkflowValidationError):
+    """Exception raised when type incompatibility is detected between connected executors."""
+
+    def __init__(
+        self,
+        source_executor_id: str,
+        target_executor_id: str,
+        source_types: list[type[Any]],
+        target_types: list[type[Any]],
+    ):
+        # Use a placeholder for incompatible types - will be computed in WorkflowGraphValidator
+        super().__init__(
+            message=f"Type incompatibility between executors '{source_executor_id}' -> '{target_executor_id}'. "
+            f"Source executor outputs types {[str(t) for t in source_types]} but target executor "
+            f"can only handle types {[str(t) for t in target_types]}.",
+            validation_type=ValidationTypeEnum.TYPE_COMPATIBILITY,
+        )
+        self.source_executor_id = source_executor_id
+        self.target_executor_id = target_executor_id
+        self.source_types = source_types
+        self.target_types = target_types
+
+
+class GraphConnectivityError(WorkflowValidationError):
+    """Exception raised when graph connectivity issues are detected."""
+
+    def __init__(self, message: str):
+        super().__init__(message, validation_type=ValidationTypeEnum.GRAPH_CONNECTIVITY)
+
+
+# endregion
+
+
+# region Workflow Graph Validator
+class WorkflowGraphValidator:
+    """Validator for workflow graphs.
+
+    This validator performs multiple validation checks:
+    1. Edge duplication validation
+    2. Type compatibility validation between connected executors
+    3. Graph connectivity validation
+    """
+
+    def __init__(self):
+        self._edges: list[Edge] = []
+        self._executors: dict[str, Executor] = {}
+
+    # region Core Validation Methods
+    def validate_workflow(self, edges: list[Edge], start_executor: Executor | str) -> None:
+        """Validate the entire workflow graph.
+
+        Args:
+            edges: list of edges in the workflow
+            start_executor: The starting executor (can be instance or ID)
+
+        Raises:
+            WorkflowValidationError: If any validation fails
+        """
+        self._edges = edges
+        self._executors = self._build_executor_map(edges)
+
+        # Validate that start_executor exists in the graph
+        # It should because we check for it in the WorkflowBuilder
+        # but we do it here for completeness.
+        start_executor_id = start_executor.id if isinstance(start_executor, Executor) else start_executor
+        if start_executor_id not in self._executors:
+            raise GraphConnectivityError(f"Start executor '{start_executor_id}' is not present in the workflow graph")
+
+        # Run all checks
+        self._validate_edge_duplication()
+        self._validate_type_compatibility()
+        self._validate_graph_connectivity(start_executor_id)
+        self._validate_self_loops()
+        self._validate_handler_ambiguity()
+        self._validate_dead_ends()
+        self._validate_cycles()
+
+    def _build_executor_map(self, edges: list[Edge]) -> dict[str, Executor]:
+        """Build a map of executor IDs to executor instances."""
+        executors: dict[str, Executor] = {}
+        for edge in edges:
+            executors[edge.source_id] = edge.source
+            executors[edge.target_id] = edge.target
+        return executors
+
+    # endregion
+
+    # region Edge and Type Validation
+    def _validate_edge_duplication(self) -> None:
+        """Validate that there are no duplicate edges in the workflow.
+
+        Raises:
+            EdgeDuplicationError: If duplicate edges are found
+        """
+        seen_edge_ids: set[str] = set()
+
+        for edge in self._edges:
+            edge_id = edge.id
+            if edge_id in seen_edge_ids:
+                raise EdgeDuplicationError(edge_id)
+            seen_edge_ids.add(edge_id)
+
+    def _validate_type_compatibility(self) -> None:
+        """Validate type compatibility between connected executors.
+
+        This checks that the output types of source executors are compatible
+        with the input types expected by target executors.
+
+        Raises:
+            TypeCompatibilityError: If type incompatibility is detected
+        """
+        for edge in self._edges:
+            source_executor = edge.source
+            target_executor = edge.target
+
+            # Get output types from source executor
+            source_output_types = self._get_executor_output_types(source_executor)
+
+            # Get input types from target executor
+            target_input_types = self._get_executor_input_types(target_executor)
+
+            # If either executor has no type information, log warning and skip validation
+            # This allows for dynamic typing scenarios but warns about reduced validation coverage
+            if not source_output_types or not target_input_types:
+                if not source_output_types:
+                    logger.warning(
+                        f"Executor '{source_executor.id}' has no output type annotations. "
+                        f"Type compatibility validation will be skipped for edges from this executor. "
+                        f"Consider adding output_types to @handler decorators for better validation."
+                    )
+                if not target_input_types:
+                    logger.warning(
+                        f"Executor '{target_executor.id}' has no input type annotations. "
+                        f"Type compatibility validation will be skipped for edges to this executor. "
+                        f"Consider adding type annotations to message handler parameters for better validation."
+                    )
+                continue
+
+            # Check if any source output type is compatible with any target input type
+            compatible = False
+            compatible_pairs: list[tuple[type[Any], type[Any]]] = []
+
+            for source_type in source_output_types:
+                for target_type in target_input_types:
+                    if edge.has_edge_group():
+                        # If the edge is part of an edge group, the target expects a list of data types
+                        if self._is_type_compatible(list[source_type], target_type):
+                            compatible = True
+                            compatible_pairs.append((list[source_type], target_type))
+                    else:
+                        if self._is_type_compatible(source_type, target_type):
+                            compatible = True
+                            compatible_pairs.append((source_type, target_type))
+
+            # Log successful type compatibility for debugging
+            if compatible:
+                logger.debug(
+                    f"Type compatibility validated for edge '{source_executor.id}' -> '{target_executor.id}'. "
+                    f"Compatible type pairs: {[(str(s), str(t)) for s, t in compatible_pairs]}"
+                )
+
+            if not compatible:
+                # Enhanced error with more detailed information
+                raise TypeCompatibilityError(
+                    source_executor.id,
+                    target_executor.id,
+                    source_output_types,
+                    target_input_types,
+                )
+
+    def _get_executor_output_types(self, executor: Executor) -> list[type[Any]]:
+        """Extract output types from an executor's message handlers.
+
+        Args:
+            executor: The executor to analyze
+
+        Returns:
+            list of types that this executor can output
+        """
+        output_types: list[type[Any]] = []
+
+        for attr_name in dir(executor):
+            attr = getattr(executor, attr_name)
+            if callable(attr) and hasattr(attr, "_handler_spec"):
+                handler_spec = attr._handler_spec  # type: ignore
+                handler_output_types = handler_spec.get("output_types", [])
+                output_types.extend(handler_output_types)
+
+        return output_types
+
+    def _get_executor_input_types(self, executor: Executor) -> list[type[Any]]:
+        """Extract input types from an executor's message handlers.
+
+        Args:
+            executor: The executor to analyze
+
+        Returns:
+            list of types that this executor can handle as input
+        """
+        input_types: list[type[Any]] = []
+
+        # Access the private _handlers attribute to get input types
+        if hasattr(executor, "_handlers"):
+            input_types.extend(executor._handlers.keys())  # type: ignore
+
+        return input_types
+
+    # endregion
+
+    # region Graph Connectivity Validation
+    def _validate_graph_connectivity(self, start_executor_id: str) -> None:
+        """Validate graph connectivity and detect potential issues.
+
+        This performs several checks:
+        - Detects unreachable executors from the start node
+        - Detects isolated executors (no incoming or outgoing edges)
+        - Warns about potential infinite loops
+
+        Args:
+            start_executor_id: The ID of the starting executor
+
+        Raises:
+            GraphConnectivityError: If connectivity issues are detected
+        """
+        # Build adjacency list for the graph
+        graph: dict[str, list[str]] = defaultdict(list)
+        all_executors = set(self._executors.keys())
+
+        for edge in self._edges:
+            graph[edge.source_id].append(edge.target_id)
+
+        # Find reachable nodes from start
+        reachable = self._find_reachable_nodes(graph, start_executor_id)
+
+        # Check for unreachable executors
+        unreachable = all_executors - reachable
+        if unreachable:
+            raise GraphConnectivityError(
+                f"The following executors are unreachable from the start executor '{start_executor_id}': "
+                f"{sorted(unreachable)}. This may indicate a disconnected workflow graph."
+            )
+
+        # Check for isolated executors (no edges)
+        isolated_executors: list[str] = []
+        for executor_id in all_executors:
+            has_incoming = any(edge.target_id == executor_id for edge in self._edges)
+            has_outgoing = any(edge.source_id == executor_id for edge in self._edges)
+
+            if not has_incoming and not has_outgoing and executor_id != start_executor_id:
+                isolated_executors.append(executor_id)
+
+        if isolated_executors:
+            raise GraphConnectivityError(
+                f"The following executors are isolated (no incoming or outgoing edges): "
+                f"{sorted(isolated_executors)}. Isolated executors will never be executed."
+            )
+
+    def _find_reachable_nodes(self, graph: dict[str, list[str]], start: str) -> set[str]:
+        """Find all nodes reachable from the start node using DFS.
+
+        Args:
+            graph: Adjacency list representation of the graph
+            start: Starting node ID
+
+        Returns:
+            Set of reachable node IDs
+        """
+        visited: set[str] = set()
+        stack = [start]
+
+        while stack:
+            node = stack.pop()
+            if node not in visited:
+                visited.add(node)
+                stack.extend(graph[node])
+
+        return visited
+
+    # endregion
+
+    # region Additional Validation Scenarios
+    def _validate_self_loops(self) -> None:
+        """Detect and log self-loops (edges from executor to itself).
+
+        Self-loops might indicate recursive processing which could be intentional
+        but should be highlighted for review.
+        """
+        self_loops = [edge for edge in self._edges if edge.source_id == edge.target_id]
+
+        for edge in self_loops:
+            logger.warning(
+                f"Self-loop detected: Executor '{edge.source_id}' connects to itself. "
+                f"This may cause infinite recursion if not properly handled with conditions."
+            )
+
+    def _validate_handler_ambiguity(self) -> None:
+        """Check for potential ambiguity in message handlers.
+
+        Warns when executors have multiple handlers that could handle the same type,
+        which might lead to unexpected behavior.
+        """
+        for executor_id, executor in self._executors.items():
+            input_types = self._get_executor_input_types(executor)
+
+            # Check for duplicate input types
+            seen_types: set[type[Any]] = set()
+            duplicate_types: set[type[Any]] = set()
+
+            for input_type in input_types:
+                if input_type in seen_types:
+                    duplicate_types.add(input_type)
+                seen_types.add(input_type)
+
+            if duplicate_types:
+                logger.warning(
+                    f"Executor '{executor_id}' has multiple handlers for the same input types: "
+                    f"{[str(t) for t in duplicate_types]}. This may lead to ambiguous message routing."
+                )
+
+    def _validate_dead_ends(self) -> None:
+        """Identify executors that have no outgoing edges (potential dead ends).
+
+        These might be intentional final nodes or could indicate missing connections.
+        """
+        executors_with_outgoing = {edge.source_id for edge in self._edges}
+        all_executor_ids = set(self._executors.keys())
+        dead_ends = all_executor_ids - executors_with_outgoing
+
+        if dead_ends:
+            logger.info(
+                f"Dead-end executors detected (no outgoing edges): {sorted(dead_ends)}. "
+                f"Verify these are intended as final nodes in the workflow."
+            )
+
+    def _validate_cycles(self) -> None:
+        """Detect cycles in the workflow graph.
+
+        Cycles might be intentional for iterative processing but should be flagged
+        for review to ensure proper termination conditions exist.
+        """
+        # Build adjacency list
+        graph: dict[str, list[str]] = defaultdict(list)
+        for edge in self._edges:
+            graph[edge.source_id].append(edge.target_id)
+
+        # Use DFS to detect cycles
+        white = set(self._executors.keys())  # Unvisited
+        gray: set[str] = set()  # Currently being processed
+        black: set[str] = set()  # Completely processed
+
+        def has_cycle(node: str) -> bool:
+            if node in gray:  # Back edge found - cycle detected
+                return True
+            if node in black:  # Already processed
+                return False
+
+            # Mark as being processed
+            white.discard(node)
+            gray.add(node)
+
+            # Visit neighbors
+            for neighbor in graph[node]:
+                if has_cycle(neighbor):
+                    return True
+
+            # Mark as completely processed
+            gray.discard(node)
+            black.add(node)
+            return False
+
+        # Check for cycles starting from any unvisited node
+        cycle_detected = False
+        while white and not cycle_detected:
+            start_node = next(iter(white))
+            if has_cycle(start_node):
+                cycle_detected = True
+
+        if cycle_detected:
+            logger.warning(
+                "Cycle detected in the workflow graph. "
+                "Ensure proper termination conditions exist to prevent infinite loops."
+            )
+
+    # endregion
+
+    # region Type Compatibility Utilities
+    @staticmethod
+    def _is_type_compatible(source_type: type[Any], target_type: type[Any]) -> bool:
+        """Check if source_type is compatible with target_type."""
+        # Handle Any type
+        if source_type is Any or target_type is Any:
+            return True
+
+        # Handle exact match
+        if source_type == target_type:
+            return True
+
+        # Handle inheritance
+        try:
+            if inspect.isclass(source_type) and inspect.isclass(target_type):
+                return issubclass(source_type, target_type)
+        except TypeError:
+            # Handle generic types that can't be used with issubclass
+            pass
+
+        # Handle Union types
+        source_origin = get_origin(source_type)
+        target_origin = get_origin(target_type)
+
+        if target_origin is Union:
+            target_args = get_args(target_type)
+            return any(WorkflowGraphValidator._is_type_compatible(source_type, arg) for arg in target_args)
+
+        if source_origin is Union:
+            source_args = get_args(source_type)
+            return all(WorkflowGraphValidator._is_type_compatible(arg, target_type) for arg in source_args)
+
+        # Handle generic types
+        if source_origin is not None and target_origin is not None and source_origin == target_origin:
+            source_args = get_args(source_type)
+            target_args = get_args(target_type)
+            if len(source_args) == len(target_args):
+                return all(
+                    WorkflowGraphValidator._is_type_compatible(s_arg, t_arg)
+                    for s_arg, t_arg in zip(source_args, target_args, strict=True)
+                )
+
+        return False
+
+    # endregion
+
+
+# endregion
+
+
+def validate_workflow_graph(edges: list[Edge], start_executor: Executor | str) -> None:
+    """Convenience function to validate a workflow graph.
+
+    Args:
+        edges: list of edges in the workflow
+        start_executor: The starting executor (can be instance or ID)
+
+    Raises:
+        WorkflowValidationError: If any validation fails
+    """
+    validator = WorkflowGraphValidator()
+    validator.validate_workflow(edges, start_executor)

python/packages/workflow/agent_framework_workflow/_workflow.py

@@ -0,0 +1,350 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import sys
+from collections.abc import AsyncIterable, Callable, Sequence
+from typing import Any
+
+from ._edge import Edge
+from ._events import RequestInfoEvent, WorkflowCompletedEvent, WorkflowEvent
+from ._executor import Executor, RequestInfoExecutor
+from ._runner import DEFAULT_MAX_ITERATIONS, Runner
+from ._runner_context import InProcRunnerContext, RunnerContext
+from ._shared_state import SharedState
+from ._validation import validate_workflow_graph
+from ._workflow_context import WorkflowContext
+
+if sys.version_info >= (3, 11):
+    from typing import Self  # pragma: no cover
+else:
+    from typing_extensions import Self  # pragma: no cover
+
+
+class WorkflowRunResult(list[WorkflowEvent]):
+    """A list of events generated during the workflow execution in non-streaming mode."""
+
+    def get_completed_event(self) -> WorkflowCompletedEvent | None:
+        """Get the completed event from the workflow run result.
+
+        Returns:
+            A completed WorkflowEvent instance if the workflow has a completed event, otherwise None.
+
+        Raises:
+            ValueError: If there are multiple completed events in the workflow run result.
+        """
+        completed_events = [event for event in self if isinstance(event, WorkflowCompletedEvent)]
+        if not completed_events:
+            return None
+        if len(completed_events) > 1:
+            raise ValueError("Multiple completed events found.")
+        return completed_events[0]
+
+    def get_request_info_events(self) -> list[RequestInfoEvent]:
+        """Get all request info events from the workflow run result.
+
+        Returns:
+            A list of RequestInfoEvent instances found in the workflow run result.
+        """
+        return [event for event in self if isinstance(event, RequestInfoEvent)]
+
+
+class Workflow:
+    """A class representing a workflow that can be executed.
+
+    This class is a placeholder for the workflow logic and does not implement any specific functionality.
+    It serves as a base class for more complex workflows that can be defined in subclasses.
+    """
+
+    def __init__(
+        self,
+        edges: list[Edge],
+        start_executor: Executor | str,
+        runner_context: RunnerContext,
+        max_iterations: int,
+    ):
+        """Initialize the workflow with a list of edges.
+
+        Args:
+            edges: A list of directed edges representing the connections between nodes in the workflow.
+            start_executor: The starting executor for the workflow, which can be an Executor instance or its ID.
+            runner_context: The RunnerContext instance to be used during workflow execution.
+            max_iterations: The maximum number of iterations the workflow will run for convergence.
+        """
+        self._edges = edges
+        self._start_executor = start_executor
+        self._executors = {edge.source_id: edge.source for edge in edges} | {
+            edge.target_id: edge.target for edge in edges
+        }
+
+        self._shared_state = SharedState()
+        self._runner = Runner(self._edges, self._shared_state, runner_context, max_iterations=max_iterations)
+
+    @property
+    def edges(self) -> list[Edge]:
+        """Get the list of edges in the workflow."""
+        return self._edges
+
+    @property
+    def start_executor(self) -> Executor:
+        """Get the starting executor of the workflow.
+
+        Returns:
+            The starting executor, which can be an Executor instance or its ID.
+        """
+        if isinstance(self._start_executor, str):
+            return self._get_executor_by_id(self._start_executor)
+        return self._start_executor
+
+    @property
+    def executors(self) -> list[Executor]:
+        """Get the list of executors in the workflow."""
+        return list(self._executors.values())
+
+    async def run_streaming(self, message: Any) -> AsyncIterable[WorkflowEvent]:
+        """Send a message to the starting executor of the workflow and stream the events generated by the workflow.
+
+        Args:
+            message: The message to be sent to the starting executor.
+
+        Yields:
+            WorkflowEvent: The events generated during the workflow execution.
+        """
+        executor = self._start_executor
+        if isinstance(executor, str):
+            executor = self._get_executor_by_id(executor)
+
+        await executor.execute(
+            message,
+            WorkflowContext(
+                executor.id,
+                [
+                    # Using the workflow class name as the source executor ID when
+                    # delivering the first message to the starting executor
+                    self.__class__.__name__
+                ],
+                self._shared_state,
+                self._runner.context,
+            ),
+        )
+        async for event in self._runner.run_until_convergence():
+            yield event
+
+    async def send_responses_streaming(self, responses: dict[str, Any]) -> AsyncIterable[WorkflowEvent]:
+        """Send responses back to the workflow and stream the events generated by the workflow.
+
+        Args:
+            responses: The responses to be sent back to the workflow, where keys are request IDs
+                       and values are the corresponding response data.
+
+        Yields:
+            WorkflowEvent: The events generated during the workflow execution after sending the responses.
+        """
+        request_info_executor = self._get_executor_by_id(RequestInfoExecutor.EXECUTOR_ID)
+        if not isinstance(request_info_executor, RequestInfoExecutor):
+            raise ValueError(f"Executor with ID {RequestInfoExecutor.EXECUTOR_ID} is not a RequestInfoExecutor.")
+
+        async def _handle_response(response: Any, request_id: str) -> None:
+            """Handle the response from the RequestInfoExecutor."""
+            await request_info_executor.handle_response(
+                response,
+                request_id,
+                WorkflowContext(
+                    request_info_executor.id,
+                    [
+                        # Using the workflow class name as the source executor ID when
+                        # delivering the first message to the starting executor
+                        self.__class__.__name__
+                    ],
+                    self._shared_state,
+                    self._runner.context,
+                ),
+            )
+
+        await asyncio.gather(*[_handle_response(response, request_id) for request_id, response in responses.items()])
+
+        async for event in self._runner.run_until_convergence():
+            yield event
+
+    async def run(self, message: Any) -> WorkflowRunResult:
+        """Run the workflow with the given message.
+
+        Args:
+            message: The message to be processed by the workflow.
+
+        Returns:
+            A WorkflowRunResult instance containing a list of events generated during the workflow execution.
+        """
+        events = [event async for event in self.run_streaming(message)]
+        return WorkflowRunResult(events)
+
+    async def send_responses(self, responses: dict[str, Any]) -> WorkflowRunResult:
+        """Send responses back to the workflow.
+
+        Args:
+            responses: A dictionary where keys are request IDs and values are the corresponding response data.
+
+        Returns:
+            A WorkflowRunResult instance containing a list of events generated during the workflow execution.
+        """
+        events = [event async for event in self.send_responses_streaming(responses)]
+        return WorkflowRunResult(events)
+
+    def _get_executor_by_id(self, executor_id: str) -> Executor:
+        """Get an executor by its ID.
+
+        Args:
+            executor_id: The ID of the executor to retrieve.
+
+        Returns:
+            The Executor instance corresponding to the given ID.
+        """
+        if executor_id not in self._executors:
+            raise ValueError(f"Executor with ID {executor_id} not found.")
+        return self._executors[executor_id]
+
+
+class WorkflowBuilder:
+    """A builder class for constructing workflows.
+
+    This class provides methods to add edges and set the starting executor for the workflow.
+    """
+
+    def __init__(self):
+        """Initialize the WorkflowBuilder with an empty list of edges and no starting executor."""
+        self._edges: list[Edge] = []
+        self._start_executor: Executor | str | None = None
+        self._max_iterations: int = DEFAULT_MAX_ITERATIONS
+
+    def add_edge(
+        self,
+        source: Executor,
+        target: Executor,
+        condition: Callable[[Any], bool] | None = None,
+    ) -> "Self":
+        """Add a directed edge between two executors.
+
+        The output types of the source and the input types of the target must be compatible.
+
+        Args:
+            source: The source executor of the edge.
+            target: The target executor of the edge.
+            condition: An optional condition function that determines whether the edge
+                       should be traversed based on the message type.
+        """
+        # TODO(@taochen): Support executor factories for lazy initialization
+        self._edges.append(Edge(source, target, condition))
+        return self
+
+    def add_fan_out_edges(self, source: Executor, targets: Sequence[Executor]) -> "Self":
+        """Add multiple edges to the workflow.
+
+        The output types of the source and the input types of the targets must be compatible.
+        Messages from the source executor will be sent to all target executors.
+
+        Args:
+            source: The source executor of the edges.
+            targets: A list of target executors for the edges.
+        """
+        for target in targets:
+            self._edges.append(Edge(source, target))
+        return self
+
+    def add_fan_in_edges(self, sources: Sequence[Executor], target: Executor) -> "Self":
+        """Add multiple edges from sources to a single target executor.
+
+        The edges will be grouped together for synchronized processing, meaning
+        the target executor will only be executed once all source executors have completed.
+
+        The target executor will receive a list of messages aggregated from all source executors.
+        Thus the input types of the target executor must be compatible with a list of the output
+        types of the source executors. For example:
+
+            class Target(Executor):
+                @handler
+                def handle_messages(self, messages: list[Message]) -> None:
+                    # Process the aggregated messages from all sources
+
+            class Source(Executor):
+                @handler(output_type=[Message])
+                def handle_message(self, message: Message) -> None:
+                    # Send a message to the target executor
+                    self.send_message(message)
+
+            workflow = (
+                WorkflowBuilder()
+                .add_fan_in_edges(
+                    [Source(id="source1"), Source(id="source2")],
+                    Target(id="target")
+                )
+                .build()
+            )
+
+        Args:
+            sources: A list of source executors for the edges.
+            target: The target executor for the edges.
+        """
+        edges = [Edge(source, target) for source in sources]
+
+        # Set the edge groups for the edges to ensure they are processed together.
+        for i, edge in enumerate(edges):
+            group_ids: list[str] = []
+            group_ids.extend([e.id for e in edges[0:i]])
+            group_ids.extend([e.id for e in edges[i + 1 :]])
+            edge.set_edge_group(group_ids)
+
+        self._edges.extend(edges)
+
+        return self
+
+    def add_chain(self, executors: Sequence[Executor]) -> "Self":
+        """Add a chain of executors to the workflow.
+
+        The output of each executor in the chain will be sent to the next executor in the chain.
+        The input types of each executor must be compatible with the output types of the previous executor.
+
+        Circles in the chain are not allowed, meaning the chain cannot have two executors with the same ID.
+
+        Args:
+            executors: A list of executors to be added to the chain.
+        """
+        for i in range(len(executors) - 1):
+            self.add_edge(executors[i], executors[i + 1])
+        return self
+
+    def set_start_executor(self, executor: Executor | str) -> "Self":
+        """Set the starting executor for the workflow.
+
+        Args:
+            executor: The starting executor, which can be an Executor instance or its ID.
+        """
+        self._start_executor = executor
+        return self
+
+    def set_max_iterations(self, max_iterations: int) -> "Self":
+        """Set the maximum number of iterations for the workflow.
+
+        Args:
+            max_iterations: The maximum number of iterations the workflow will run for convergence.
+        """
+        self._max_iterations = max_iterations
+        return self
+
+    def build(self) -> Workflow:
+        """Build and return the constructed workflow.
+
+        This method performs validation before building the workflow.
+
+        Returns:
+            A Workflow instance with the defined edges and starting executor.
+
+        Raises:
+            ValueError: If starting executor is not set.
+            WorkflowValidationError: If workflow validation fails (includes EdgeDuplicationError,
+                TypeCompatibilityError, and GraphConnectivityError subclasses).
+        """
+        if not self._start_executor:
+            raise ValueError("Starting executor must be set before building the workflow.")
+
+        validate_workflow_graph(self._edges, self._start_executor)
+
+        return Workflow(self._edges, self._start_executor, InProcRunnerContext(), self._max_iterations)

python/packages/workflow/agent_framework_workflow/_workflow_context.py

@@ -0,0 +1,91 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from typing import Any
+
+from ._events import WorkflowEvent
+from ._runner_context import Message, RunnerContext
+from ._shared_state import SharedState
+
+
+class WorkflowContext:
+    """Context for executors in a workflow.
+
+    This class is used to provide a way for executors to interact with the workflow
+    context and shared state, while preventing direct access to the runtime context.
+    """
+
+    def __init__(
+        self,
+        executor_id: str,
+        source_executor_ids: list[str],
+        shared_state: SharedState,
+        runner_context: RunnerContext,
+    ):
+        """Initialize the executor context with the given workflow context.
+
+        Args:
+            executor_id: The unique identifier of the executor that this context belongs to.
+            source_executor_ids: The IDs of the source executors that sent messages to this executor.
+                This is a list to support fan_in scenarios where multiple sources send aggregated
+                messages to the same executor.
+            shared_state: The shared state for the workflow.
+            runner_context: The runner context that provides methods to send messages and events.
+        """
+        self._executor_id = executor_id
+        self._source_executor_ids = source_executor_ids
+        self._runner_context = runner_context
+        self._shared_state = shared_state
+
+        if not self._source_executor_ids:
+            raise ValueError("source_executor_ids cannot be empty. At least one source executor ID is required.")
+
+    async def send_message(self, message: Any, target_id: str | None = None) -> None:
+        """Send a message to the workflow context.
+
+        Args:
+            message: The message to send. This can be any data type that the target executor can handle.
+            target_id: The ID of the target executor to send the message to.
+                       If None, the message will be sent to all target executors.
+        """
+        await self._runner_context.send_message(
+            Message(
+                data=message,
+                source_id=self._executor_id,
+                target_id=target_id,
+            )
+        )
+
+    async def add_event(self, event: WorkflowEvent) -> None:
+        """Add an event to the workflow context."""
+        await self._runner_context.add_event(event)
+
+    async def get_shared_state(self, key: str) -> Any:
+        """Get a value from the shared state."""
+        return await self._shared_state.get(key)
+
+    async def set_shared_state(self, key: str, value: Any) -> None:
+        """Set a value in the shared state."""
+        await self._shared_state.set(key, value)
+
+    def get_source_executor_id(self) -> str:
+        """Get the ID of the source executor that sent the message to this executor.
+
+        Raises:
+            RuntimeError: If there are multiple source executors, this method raises an error.
+        """
+        if len(self._source_executor_ids) > 1:
+            raise RuntimeError(
+                "Cannot get source executor ID when there are multiple source executors. "
+                "Access the full list via the source_executor_ids property instead."
+            )
+        return self._source_executor_ids[0]
+
+    @property
+    def source_executor_ids(self) -> list[str]:
+        """Get the IDs of the source executors that sent messages to this executor."""
+        return self._source_executor_ids
+
+    @property
+    def shared_state(self) -> SharedState:
+        """Get the shared state."""
+        return self._shared_state

python/packages/workflow/pyproject.toml

@@ -0,0 +1,83 @@
+[project]
+name = "agent-framework-workflow"
+description = "Workflow integration for Microsoft Agent Framework."
+authors = [{ name = "Microsoft", email = "SK-Support@microsoft.com"}]
+readme = "README.md"
+requires-python = ">=3.10"
+version = "0.1.0b1"
+license-files = ["LICENSE"]
+urls.homepage = "https://learn.microsoft.com/en-us/semantic-kernel/overview/"
+urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
+urls.release_notes = "https://github.com/microsoft/agent-framework/releases?q=tag%3Apython-1&expanded=true"
+urls.issues = "https://github.com/microsoft/agent-framework/issues"
+classifiers = [
+  "License :: OSI Approved :: MIT License",
+  "Development Status :: 5 - Production/Stable",
+  "Intended Audience :: Developers",
+  "Programming Language :: Python :: 3",
+  "Programming Language :: Python :: 3.10",
+  "Programming Language :: Python :: 3.11",
+  "Programming Language :: Python :: 3.12",
+  "Programming Language :: Python :: 3.13",
+  "Framework :: Pydantic :: 2",
+  "Typing :: Typed",
+]
+dependencies = [
+    "agent-framework",
+]
+
+[tool.uv]
+prerelease = "if-necessary-or-explicit"
+environments = [
+    "sys_platform == 'darwin'",
+    "sys_platform == 'linux'",
+    "sys_platform == 'win32'"
+]
+
+[tool.uv-dynamic-versioning]
+fallback-version = "0.0.0"
+[tool.pytest.ini_options]
+testpaths = 'tests'
+addopts = "-ra -q -r fEX"
+asyncio_mode = "auto"
+asyncio_default_fixture_loop_scope = "function"
+filterwarnings = []
+timeout = 120
+
+[tool.ruff]
+extend = "../../pyproject.toml"
+
+[tool.pyright]
+extend = "../../pyproject.toml"
+exclude = ['tests']
+
+[tool.mypy]
+plugins = ['pydantic.mypy']
+strict = true
+python_version = "3.10"
+ignore_missing_imports = true
+disallow_untyped_defs = true
+no_implicit_optional = true
+check_untyped_defs = true
+warn_return_any = true
+show_error_codes = true
+warn_unused_ignores = false
+disallow_incomplete_defs = true
+disallow_untyped_decorators = true
+disallow_any_unimported = true
+
+[tool.bandit]
+targets = ["agent_framework_workflow"]
+exclude_dirs = ["tests"]
+
+[tool.poe]
+executor.type = "uv"
+include = "../../shared_tasks.toml"
+
+[tool.uv.build-backend]
+module-name = "agent_framework_workflow"
+module-root = ""
+
+[build-system]
+requires = ["uv_build>=0.8.2,<0.9.0"]
+build-backend = "uv_build"

python/packages/workflow/tests/test_edge.py

@@ -0,0 +1,47 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from dataclasses import dataclass
+from typing import Any
+
+from agent_framework.workflow import Executor, WorkflowContext, handler
+
+from agent_framework_workflow._edge import Edge
+
+
+@dataclass
+class MockMessage:
+    """A mock message for testing purposes."""
+
+    data: Any
+
+
+class MockExecutor(Executor):
+    """A mock executor for testing purposes."""
+
+    @handler
+    async def mock_handler(self, message: MockMessage, ctx: WorkflowContext) -> None:
+        """A mock handler that does nothing."""
+        pass
+
+
+def test_create_edge():
+    """Test creating an edge with a source and target executor."""
+    source = MockExecutor(id="source_executor")
+    target = MockExecutor(id="target_executor")
+
+    edge = Edge(source=source, target=target)
+
+    assert edge.source_id == "source_executor"
+    assert edge.target_id == "target_executor"
+    assert edge.id == f"{edge.source_id}{Edge.ID_SEPARATOR}{edge.target_id}"
+    assert (edge.source_id, edge.target_id) == Edge.source_and_target_from_id(edge.id)
+
+
+def test_edge_can_handle():
+    """Test creating an edge with a source and target executor."""
+    source = MockExecutor(id="source_executor")
+    target = MockExecutor(id="target_executor")
+
+    edge = Edge(source=source, target=target)
+
+    assert edge.can_handle(MockMessage(data="test"))

python/packages/workflow/tests/test_executor.py

@@ -0,0 +1,102 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import pytest
+from agent_framework.workflow import Executor, WorkflowContext, handler
+
+
+def test_executor_without_handlers():
+    """Test that an executor without handlers raises an error when trying to run."""
+
+    class MockExecutorWithoutHandlers(Executor):
+        """A mock executor that does not implement any handlers."""
+
+        pass
+
+    with pytest.raises(ValueError):
+        MockExecutorWithoutHandlers()
+
+
+def test_executor_handler_without_annotations():
+    """Test that an executor with one handler without annotations raises an error when trying to run."""
+
+    with pytest.raises(ValueError):
+
+        class MockExecutorWithOneHandlerWithoutAnnotations(Executor):  # type: ignore
+            """A mock executor with one handler that does not implement any annotations."""
+
+            @handler
+            async def handle(self, message, ctx) -> None:  # type: ignore
+                """A mock handler that does not implement any annotations."""
+                pass
+
+
+def test_executor_invalid_handler_signature():
+    """Test that an executor with an invalid handler signature raises an error when trying to run."""
+
+    with pytest.raises(ValueError):
+
+        class MockExecutorWithInvalidHandlerSignature(Executor):  # type: ignore
+            """A mock executor with an invalid handler signature."""
+
+            @handler  # type: ignore
+            async def handle(self, message, other, ctx) -> None:  # type: ignore
+                """A mock handler with an invalid signature."""
+                pass
+
+
+def test_executor_with_valid_handlers():
+    """Test that an executor with valid handlers can be instantiated and run."""
+
+    class MockExecutorWithValidHandlers(Executor):  # type: ignore
+        """A mock executor with valid handlers."""
+
+        @handler
+        async def handle_text(self, text: str, ctx: WorkflowContext) -> None:  # type: ignore
+            """A mock handler with a valid signature."""
+            pass
+
+        @handler
+        async def handle_number(self, number: int, ctx: WorkflowContext) -> None:  # type: ignore
+            """Another mock handler with a valid signature."""
+            pass
+
+    executor = MockExecutorWithValidHandlers()
+    assert executor.id is not None
+    assert len(executor._handlers) == 2  # type: ignore
+    assert executor.can_handle("text") is True
+    assert executor.can_handle(42) is True
+    assert executor.can_handle(3.14) is False
+
+
+def test_executor_handlers_with_output_types():
+    """Test that an executor with handlers that specify output types can be instantiated and run."""
+
+    class MockExecutorWithOutputTypes(Executor):  # type: ignore
+        """A mock executor with handlers that specify output types."""
+
+        @handler(output_types=[str])
+        async def handle_string(self, text: str, ctx: WorkflowContext) -> None:  # type: ignore
+            """A mock handler that outputs a string."""
+            pass
+
+        @handler(output_types=[int])
+        async def handle_integer(self, number: int, ctx: WorkflowContext) -> None:  # type: ignore
+            """A mock handler that outputs an integer."""
+            pass
+
+    executor = MockExecutorWithOutputTypes()
+    assert len(executor._handlers) == 2  # type: ignore
+
+    string_handler = executor._handlers[str]  # type: ignore
+    assert string_handler is not None
+    assert string_handler._handler_spec is not None  # type: ignore
+    assert string_handler._handler_spec["name"] == "handle_string"  # type: ignore
+    assert string_handler._handler_spec["message_type"] is str  # type: ignore
+    assert string_handler._handler_spec["output_types"] == [str]  # type: ignore
+
+    int_handler = executor._handlers[int]  # type: ignore
+    assert int_handler is not None
+    assert int_handler._handler_spec is not None  # type: ignore
+    assert int_handler._handler_spec["name"] == "handle_integer"  # type: ignore
+    assert int_handler._handler_spec["message_type"] is int  # type: ignore
+    assert int_handler._handler_spec["output_types"] == [int]  # type: ignore

python/packages/workflow/tests/test_runner.py

@@ -0,0 +1,145 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+from dataclasses import dataclass
+
+import pytest
+from agent_framework.workflow import Executor, WorkflowCompletedEvent, WorkflowContext, WorkflowEvent, handler
+
+from agent_framework_workflow._edge import Edge
+from agent_framework_workflow._runner import Runner
+from agent_framework_workflow._runner_context import InProcRunnerContext, RunnerContext
+from agent_framework_workflow._shared_state import SharedState
+
+
+@dataclass
+class MockMessage:
+    """A mock message for testing purposes."""
+
+    data: int
+
+
+class MockExecutor(Executor):
+    """A mock executor for testing purposes."""
+
+    @handler(output_types=[MockMessage])
+    async def mock_handler(self, message: MockMessage, ctx: WorkflowContext) -> None:
+        if message.data < 10:
+            await ctx.send_message(MockMessage(data=message.data + 1))
+        else:
+            await ctx.add_event(WorkflowCompletedEvent(data=message.data))
+
+
+def test_create_runner():
+    """Test creating a runner with edges and shared state."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b")
+
+    # Create a loop
+    edges = [
+        Edge(source=executor_a, target=executor_b),
+        Edge(source=executor_b, target=executor_a),
+    ]
+
+    runner = Runner(edges, shared_state=SharedState(), ctx=InProcRunnerContext())
+
+    assert runner.context is not None and isinstance(runner.context, RunnerContext)
+
+
+async def test_runner_run_until_convergence():
+    """Test running the runner with a simple workflow."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b")
+
+    # Create a loop
+    edges = [
+        Edge(source=executor_a, target=executor_b),
+        Edge(source=executor_b, target=executor_a),
+    ]
+
+    shared_state = SharedState()
+    ctx = InProcRunnerContext()
+
+    runner = Runner(edges, shared_state, ctx)
+
+    result: int | None = None
+    await executor_a.execute(
+        MockMessage(data=0),
+        WorkflowContext(
+            executor_id=executor_a.id,
+            source_executor_ids=["START"],
+            shared_state=shared_state,
+            runner_context=ctx,
+        ),
+    )
+    async for event in runner.run_until_convergence():
+        assert isinstance(event, WorkflowEvent)
+        if isinstance(event, WorkflowCompletedEvent):
+            result = event.data
+
+    assert result is not None and result == 10
+
+
+async def test_runner_run_until_convergence_not_completed():
+    """Test running the runner with a simple workflow."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b")
+
+    # Create a loop
+    edges = [
+        Edge(source=executor_a, target=executor_b),
+        Edge(source=executor_b, target=executor_a),
+    ]
+
+    shared_state = SharedState()
+    ctx = InProcRunnerContext()
+
+    runner = Runner(edges, shared_state, ctx, max_iterations=5)
+
+    await executor_a.execute(
+        MockMessage(data=0),
+        WorkflowContext(
+            executor_id=executor_a.id,
+            source_executor_ids=["START"],
+            shared_state=shared_state,
+            runner_context=ctx,
+        ),
+    )
+    with pytest.raises(RuntimeError, match="Runner did not converge after 5 iterations."):
+        async for event in runner.run_until_convergence():
+            assert not isinstance(event, WorkflowCompletedEvent)
+
+
+async def test_runner_already_running():
+    """Test that running the runner while it is already running raises an error."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b")
+
+    # Create a loop
+    edges = [
+        Edge(source=executor_a, target=executor_b),
+        Edge(source=executor_b, target=executor_a),
+    ]
+
+    shared_state = SharedState()
+    ctx = InProcRunnerContext()
+
+    runner = Runner(edges, shared_state, ctx)
+
+    await executor_a.execute(
+        MockMessage(data=0),
+        WorkflowContext(
+            executor_id=executor_a.id,
+            source_executor_ids=["START"],
+            shared_state=shared_state,
+            runner_context=ctx,
+        ),
+    )
+
+    with pytest.raises(RuntimeError, match="Runner is already running."):
+
+        async def _run():
+            async for _ in runner.run_until_convergence():
+                pass
+
+        await asyncio.gather(_run(), _run())

python/packages/workflow/tests/test_validation.py

@@ -0,0 +1,555 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import logging
+from typing import Any
+
+import pytest
+
+from agent_framework_workflow import (
+    EdgeDuplicationError,
+    Executor,
+    GraphConnectivityError,
+    TypeCompatibilityError,
+    ValidationTypeEnum,
+    WorkflowBuilder,
+    WorkflowContext,
+    WorkflowValidationError,
+    handler,
+    validate_workflow_graph,
+)
+from agent_framework_workflow._edge import Edge
+
+
+class StringExecutor(Executor):
+    @handler(output_types=[str])
+    async def handle_string(self, message: str, ctx: WorkflowContext) -> None:
+        await ctx.send_message(message.upper())
+
+
+class StringAggregator(Executor):
+    """A mock executor that aggregates results from multiple executors."""
+
+    @handler(output_types=[str])
+    async def mock_handler(self, messages: list[str], ctx: WorkflowContext) -> None:
+        # This mock simply returns the data incremented by 1
+        await ctx.send_message("Aggregated: " + ", ".join(messages))
+
+
+class IntExecutor(Executor):
+    @handler(output_types=[int])
+    async def handle_int(self, message: int, ctx: WorkflowContext) -> None:
+        await ctx.send_message(message * 2)
+
+
+class AnyExecutor(Executor):
+    @handler
+    async def handle_any(self, message: Any, ctx: WorkflowContext) -> None:
+        await ctx.send_message(f"Processed: {message}")
+
+
+class NoOutputTypesExecutor(Executor):
+    @handler
+    async def handle_message(self, message: str, ctx: WorkflowContext) -> None:
+        await ctx.send_message("processed")
+
+
+class MultiTypeExecutor(Executor):
+    @handler(output_types=[str])
+    async def handle_string(self, message: str, ctx: WorkflowContext) -> None:
+        await ctx.send_message(f"String: {message}")
+
+    @handler(output_types=[int])
+    async def handle_int(self, message: int, ctx: WorkflowContext) -> None:
+        await ctx.send_message(f"Int: {message}")
+
+
+def test_valid_workflow_passes_validation():
+    executor1 = StringExecutor(id="string_executor")
+    executor2 = StringExecutor(id="string_executor_2")
+
+    # Create a valid workflow
+    workflow = (
+        WorkflowBuilder()
+        .add_edge(executor1, executor2)
+        .set_start_executor(executor1)
+        .build()  # This should not raise any exceptions
+    )
+
+    assert workflow is not None
+
+
+def test_edge_duplication_validation_fails():
+    executor1 = StringExecutor(id="executor1")
+    executor2 = StringExecutor(id="executor2")
+
+    with pytest.raises(EdgeDuplicationError) as exc_info:
+        WorkflowBuilder().add_edge(executor1, executor2).add_edge(executor1, executor2).set_start_executor(
+            executor1
+        ).build()
+
+    assert "executor1->executor2" in str(exc_info.value)
+    assert exc_info.value.validation_type == ValidationTypeEnum.EDGE_DUPLICATION
+
+
+def test_type_compatibility_validation_fails():
+    string_executor = StringExecutor(id="string_executor")
+    int_executor = IntExecutor(id="int_executor")
+
+    with pytest.raises(TypeCompatibilityError) as exc_info:
+        WorkflowBuilder().add_edge(string_executor, int_executor).set_start_executor(string_executor).build()
+
+    error = exc_info.value
+    assert error.source_executor_id == "string_executor"
+    assert error.target_executor_id == "int_executor"
+    assert error.validation_type == ValidationTypeEnum.TYPE_COMPATIBILITY
+
+
+def test_type_compatibility_with_any_type_passes():
+    string_executor = StringExecutor(id="string_executor")
+    any_executor = AnyExecutor(id="any_executor")
+
+    # This should not raise an exception
+    workflow = WorkflowBuilder().add_edge(string_executor, any_executor).set_start_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")
+
+    # This should pass validation since no output types are specified
+    workflow = (
+        WorkflowBuilder().add_edge(no_output_executor, string_executor).set_start_executor(no_output_executor).build()
+    )
+
+    assert workflow is not None
+
+
+def test_multi_type_executor_compatibility():
+    string_executor = StringExecutor(id="string_executor")
+    multi_type_executor = MultiTypeExecutor(id="multi_type")
+
+    # String executor outputs strings, multi-type can handle strings
+    workflow = (
+        WorkflowBuilder().add_edge(string_executor, multi_type_executor).set_start_executor(string_executor).build()
+    )
+
+    assert workflow is not None
+
+
+def test_graph_connectivity_unreachable_executors():
+    executor1 = StringExecutor(id="executor1")
+    executor2 = StringExecutor(id="executor2")
+    executor3 = StringExecutor(id="executor3")  # This will be unreachable
+
+    with pytest.raises(GraphConnectivityError) as exc_info:
+        WorkflowBuilder().add_edge(executor1, executor2).add_edge(executor3, executor2).set_start_executor(
+            executor1
+        ).build()
+
+    assert "unreachable" in str(exc_info.value).lower()
+    assert "executor3" in str(exc_info.value)
+    assert exc_info.value.validation_type == ValidationTypeEnum.GRAPH_CONNECTIVITY
+
+
+def test_graph_connectivity_isolated_executors():
+    executor1 = StringExecutor(id="executor1")
+    executor2 = StringExecutor(id="executor2")
+    executor3 = StringExecutor(id="executor3")  # This will be isolated
+
+    # Create edges that include an isolated executor (self-loop that's not connected to main graph)
+    edges = [Edge(executor1, executor2), Edge(executor3, executor3)]  # Self-loop to include in graph
+
+    with pytest.raises(GraphConnectivityError) as exc_info:
+        validate_workflow_graph(edges, executor1)
+
+    assert "unreachable" in str(exc_info.value).lower()
+    assert "executor3" in str(exc_info.value)
+
+
+def test_start_executor_not_in_graph():
+    executor1 = StringExecutor(id="executor1")
+    executor2 = StringExecutor(id="executor2")
+    executor3 = StringExecutor(id="executor3")  # Not in graph
+
+    with pytest.raises(GraphConnectivityError) as exc_info:
+        WorkflowBuilder().add_edge(executor1, executor2).set_start_executor(executor3).build()
+
+    assert "not present in the workflow graph" in str(exc_info.value)
+
+
+def test_missing_start_executor():
+    executor1 = StringExecutor(id="executor1")
+    executor2 = StringExecutor(id="executor2")
+
+    with pytest.raises(ValueError) as exc_info:
+        WorkflowBuilder().add_edge(executor1, executor2).build()
+
+    assert "Starting executor must be set" in str(exc_info.value)
+
+
+def test_workflow_validation_error_base_class():
+    error = WorkflowValidationError("Test message", ValidationTypeEnum.EDGE_DUPLICATION)
+    assert str(error) == "[EDGE_DUPLICATION] Test message"
+    assert error.message == "Test message"
+    assert error.validation_type == ValidationTypeEnum.EDGE_DUPLICATION
+
+
+def test_complex_workflow_validation():
+    # Create a workflow with multiple paths
+    executor1 = StringExecutor(id="executor1")
+    executor2 = MultiTypeExecutor(id="executor2")
+    executor3 = StringExecutor(id="executor3")
+    executor4 = AnyExecutor(id="executor4")
+
+    workflow = (
+        WorkflowBuilder()
+        .add_edge(executor1, executor2)  # str -> MultiType (compatible)
+        .add_edge(executor2, executor3)  # MultiType -> str (compatible)
+        .add_edge(executor2, executor4)  # MultiType -> Any (compatible)
+        .add_edge(executor3, executor4)  # str -> Any (compatible)
+        .set_start_executor(executor1)
+        .build()
+    )
+
+    assert workflow is not None
+
+
+def test_type_compatibility_inheritance():
+    class BaseExecutor(Executor):
+        @handler(output_types=[str])
+        async def handle_base(self, message: str, ctx: WorkflowContext) -> None:
+            await ctx.send_message("base")
+
+    class DerivedExecutor(Executor):
+        @handler(output_types=[str])
+        async def handle_derived(self, message: str, ctx: WorkflowContext) -> None:
+            await ctx.send_message("derived")
+
+    base_executor = BaseExecutor(id="base")
+    derived_executor = DerivedExecutor(id="derived")
+
+    # This should pass since both handle str
+    workflow = WorkflowBuilder().add_edge(base_executor, derived_executor).set_start_executor(base_executor).build()
+
+    assert workflow is not None
+
+
+def test_direct_validation_function():
+    executor1 = StringExecutor(id="executor1")
+    executor2 = StringExecutor(id="executor2")
+    edges = [Edge(executor1, executor2)]
+
+    # This should not raise any exceptions
+    validate_workflow_graph(edges, executor1)
+
+    # Test with invalid start executor
+    executor3 = StringExecutor(id="executor3")
+    with pytest.raises(GraphConnectivityError):
+        validate_workflow_graph(edges, executor3)
+
+
+def test_fan_out_validation():
+    source = StringExecutor(id="source")
+    target1 = StringExecutor(id="target1")
+    target2 = AnyExecutor(id="target2")
+
+    workflow = WorkflowBuilder().add_fan_out_edges(source, [target1, target2]).set_start_executor(source).build()
+
+    assert workflow is not None
+
+
+def test_fan_in_validation():
+    start_executor = StringExecutor(id="start")
+    source1 = StringExecutor(id="source1")
+    source2 = StringExecutor(id="source2")
+    target = StringAggregator(id="target")
+
+    # Create a proper fan-in by having a start executor that connects to both sources
+    workflow = (
+        WorkflowBuilder()
+        .add_edge(start_executor, source1)  # Start connects to source1
+        .add_edge(start_executor, source2)  # Start connects to source2
+        .add_fan_in_edges([source1, source2], target)  # Both sources fan-in to target
+        .set_start_executor(start_executor)
+        .build()
+    )
+
+    assert workflow is not None
+
+
+def test_chain_validation():
+    executor1 = StringExecutor(id="executor1")
+    executor2 = StringExecutor(id="executor2")
+    executor3 = AnyExecutor(id="executor3")
+
+    workflow = WorkflowBuilder().add_chain([executor1, executor2, executor3]).set_start_executor(executor1).build()
+
+    assert workflow is not None
+
+
+def test_logging_for_missing_output_types(caplog: Any) -> None:
+    caplog.set_level(logging.WARNING)
+
+    # Create executor without output types
+    no_output_executor = NoOutputTypesExecutor(id="no_output")
+    string_executor = StringExecutor(id="string_executor")
+
+    # This should trigger a warning log
+    workflow = (
+        WorkflowBuilder().add_edge(no_output_executor, string_executor).set_start_executor(no_output_executor).build()
+    )
+
+    assert workflow is not None
+    assert "has no output type annotations" in caplog.text
+    assert "Consider adding output_types to @handler decorators" in caplog.text
+
+
+def test_logging_for_missing_input_types(caplog: Any) -> None:
+    caplog.set_level(logging.WARNING)
+
+    class NoInputTypesExecutor(Executor):
+        # Handler without type annotation for input parameter
+        async def handle_message(self, message: Any, ctx: WorkflowContext) -> None:
+            await ctx.send_message("processed")
+
+        def _discover_handlers(self) -> None:
+            # Override to manually register handler without type info
+            self._handlers[str] = self.handle_message
+
+    string_executor = StringExecutor(id="string_executor")
+    no_input_executor = NoInputTypesExecutor(id="no_input")
+
+    # This should pass since NoInputTypesExecutor has no proper input types
+    workflow = (
+        WorkflowBuilder().add_edge(string_executor, no_input_executor).set_start_executor(string_executor).build()
+    )
+
+    assert workflow is not None
+
+
+def test_self_loop_detection_warning(caplog: Any) -> None:
+    caplog.set_level(logging.WARNING)
+
+    executor = StringExecutor(id="self_loop_executor")
+
+    # Create a self-loop
+    workflow = WorkflowBuilder().add_edge(executor, executor).set_start_executor(executor).build()
+
+    assert workflow is not None
+    assert "Self-loop detected" in caplog.text
+    assert "may cause infinite recursion" in caplog.text
+
+
+def test_handler_validation_basic(caplog: Any) -> None:
+    caplog.set_level(logging.WARNING)
+
+    # Test basic handler validation - ensure the validation code runs without errors
+    start_executor = StringExecutor(id="start")
+    target_executor = StringExecutor(id="target")
+
+    workflow = WorkflowBuilder().add_edge(start_executor, target_executor).set_start_executor(start_executor).build()
+
+    assert workflow is not None
+    # Just ensure the validation runs without errors
+
+
+def test_dead_end_detection(caplog: Any) -> None:
+    caplog.set_level(logging.INFO)
+
+    executor1 = StringExecutor(id="executor1")
+    executor2 = StringExecutor(id="executor2")  # This will be a dead end
+
+    workflow = WorkflowBuilder().add_edge(executor1, executor2).set_start_executor(executor1).build()
+
+    assert workflow is not None
+    assert "Dead-end executors detected" in caplog.text
+    assert "executor2" in caplog.text
+    assert "Verify these are intended as final nodes" in caplog.text
+
+
+def test_cycle_detection_warning(caplog: Any) -> None:
+    caplog.set_level(logging.WARNING)
+
+    executor1 = StringExecutor(id="executor1")
+    executor2 = StringExecutor(id="executor2")
+    executor3 = StringExecutor(id="executor3")
+
+    # Create a cycle: executor1 -> executor2 -> executor3 -> executor1
+    workflow = (
+        WorkflowBuilder()
+        .add_edge(executor1, executor2)
+        .add_edge(executor2, executor3)
+        .add_edge(executor3, executor1)
+        .set_start_executor(executor1)
+        .build()
+    )
+
+    assert workflow is not None
+    assert "Cycle detected in the workflow graph" in caplog.text
+    assert "Ensure proper termination conditions exist" in caplog.text
+
+
+def test_successful_type_compatibility_logging(caplog: Any) -> None:
+    caplog.set_level(logging.DEBUG)
+
+    executor1 = StringExecutor(id="executor1")
+    executor2 = StringExecutor(id="executor2")
+
+    workflow = WorkflowBuilder().add_edge(executor1, executor2).set_start_executor(executor1).build()
+
+    assert workflow is not None
+    assert "Type compatibility validated for edge" in caplog.text
+    assert "Compatible type pairs" in caplog.text
+
+
+def test_complex_cycle_detection(caplog: Any) -> None:
+    caplog.set_level(logging.WARNING)
+
+    # Create a more complex graph with multiple cycles
+    executor1 = StringExecutor(id="executor1")
+    executor2 = StringExecutor(id="executor2")
+    executor3 = StringExecutor(id="executor3")
+    executor4 = StringExecutor(id="executor4")
+
+    # Create multiple paths and cycles
+    workflow = (
+        WorkflowBuilder()
+        .add_edge(executor1, executor2)
+        .add_edge(executor2, executor3)
+        .add_edge(executor3, executor4)
+        .add_edge(executor4, executor2)  # Creates cycle: executor2 -> executor3 -> executor4 -> executor2
+        .set_start_executor(executor1)
+        .build()
+    )
+
+    assert workflow is not None
+    assert "Cycle detected in the workflow graph" in caplog.text
+
+
+def test_no_cycles_in_simple_chain(caplog: Any) -> None:
+    caplog.set_level(logging.WARNING)
+
+    executor1 = StringExecutor(id="executor1")
+    executor2 = StringExecutor(id="executor2")
+    executor3 = StringExecutor(id="executor3")
+
+    # Simple chain without cycles
+    workflow = (
+        WorkflowBuilder()
+        .add_edge(executor1, executor2)
+        .add_edge(executor2, executor3)
+        .set_start_executor(executor1)
+        .build()
+    )
+
+    assert workflow is not None
+    # Should not log cycle detection
+    assert "Cycle detected" not in caplog.text
+
+
+def test_multiple_dead_ends_detection(caplog: Any) -> None:
+    caplog.set_level(logging.INFO)
+
+    executor1 = StringExecutor(id="executor1")
+    executor2 = StringExecutor(id="executor2")  # Dead end
+    executor3 = StringExecutor(id="executor3")  # Dead end
+
+    workflow = (
+        WorkflowBuilder()
+        .add_edge(executor1, executor2)
+        .add_edge(executor1, executor3)
+        .set_start_executor(executor1)
+        .build()
+    )
+
+    assert workflow is not None
+    assert "Dead-end executors detected" in caplog.text
+    assert "executor2" in caplog.text and "executor3" in caplog.text
+
+
+def test_single_executor_workflow(caplog: Any) -> None:
+    caplog.set_level(logging.INFO)
+
+    # Test workflow with minimal structure
+    executor1 = StringExecutor(id="executor1")
+    executor2 = StringExecutor(id="executor2")
+
+    # Create a simple two-executor workflow to avoid graph validation issues
+    workflow = WorkflowBuilder().add_edge(executor1, executor2).set_start_executor(executor1).build()
+
+    assert workflow is not None
+    # Should detect executor2 as dead end
+    assert "Dead-end executors detected" in caplog.text
+
+
+def test_enhanced_type_compatibility_error_details():
+    string_executor = StringExecutor(id="string_executor")
+    int_executor = IntExecutor(id="int_executor")
+
+    with pytest.raises(TypeCompatibilityError) as exc_info:
+        WorkflowBuilder().add_edge(string_executor, int_executor).set_start_executor(string_executor).build()
+
+    error = exc_info.value
+    # Verify enhanced error contains detailed type information
+    assert "Source executor outputs types" in str(error)
+    assert "target executor can only handle types" in str(error)
+    assert error.source_types is not None
+    assert error.target_types is not None
+
+
+def test_union_type_compatibility_validation() -> None:
+    class UnionOutputExecutor(Executor):
+        @handler(output_types=[str, int])
+        async def handle_message(self, message: str, ctx: WorkflowContext) -> None:
+            await ctx.send_message("output")
+
+    class UnionInputExecutor(Executor):
+        @handler(output_types=[str])
+        async def handle_message(self, message: str, ctx: WorkflowContext) -> None:
+            await ctx.send_message("processed")
+
+    union_output = UnionOutputExecutor(id="union_output")
+    union_input = UnionInputExecutor(id="union_input")
+
+    # This should pass validation due to type compatibility (str)
+    workflow = WorkflowBuilder().add_edge(union_output, union_input).set_start_executor(union_output).build()
+
+    assert workflow is not None
+
+
+def test_generic_type_compatibility() -> None:
+    class ListOutputExecutor(Executor):
+        @handler(output_types=[list[str]])
+        async def handle_message(self, message: str, ctx: WorkflowContext) -> None:
+            await ctx.send_message(["output"])
+
+    class ListInputExecutor(Executor):
+        @handler(output_types=[str])
+        async def handle_message(self, message: list[str], ctx: WorkflowContext) -> None:
+            await ctx.send_message("processed")
+
+    list_output = ListOutputExecutor(id="list_output")
+    list_input = ListInputExecutor(id="list_input")
+
+    # This should pass validation for generic type compatibility
+    workflow = WorkflowBuilder().add_edge(list_output, list_input).set_start_executor(list_output).build()
+
+    assert workflow is not None
+
+
+def test_validation_enum_usage() -> None:
+    # Test that all validation types use the enum correctly
+    edge_error = EdgeDuplicationError("test->test")
+    assert edge_error.validation_type == ValidationTypeEnum.EDGE_DUPLICATION
+
+    type_error = TypeCompatibilityError("source", "target", [str], [int])
+    assert type_error.validation_type == ValidationTypeEnum.TYPE_COMPATIBILITY
+
+    graph_error = GraphConnectivityError("test message")
+    assert graph_error.validation_type == ValidationTypeEnum.GRAPH_CONNECTIVITY
+
+    # Test enum string representation
+    assert str(ValidationTypeEnum.EDGE_DUPLICATION) == "ValidationTypeEnum.EDGE_DUPLICATION"
+    assert ValidationTypeEnum.EDGE_DUPLICATION.value == "EDGE_DUPLICATION"

python/packages/workflow/tests/test_workflow.py

@@ -0,0 +1,277 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from dataclasses import dataclass
+
+import pytest
+from agent_framework.workflow import (
+    Executor,
+    RequestInfoEvent,
+    RequestInfoExecutor,
+    RequestInfoMessage,
+    WorkflowBuilder,
+    WorkflowCompletedEvent,
+    WorkflowContext,
+    WorkflowEvent,
+    handler,
+)
+
+
+@dataclass
+class MockMessage:
+    """A mock message for testing purposes."""
+
+    data: int
+
+
+class MockExecutor(Executor):
+    """A mock executor for testing purposes."""
+
+    def __init__(self, id: str, limit: int = 10):
+        """Initialize the mock executor with a limit."""
+        super().__init__(id=id)
+        self.limit = limit
+
+    @handler(output_types=[MockMessage])
+    async def mock_handler(self, message: MockMessage, ctx: WorkflowContext) -> None:
+        if message.data < self.limit:
+            await ctx.send_message(MockMessage(data=message.data + 1))
+        else:
+            await ctx.add_event(WorkflowCompletedEvent(data=message.data))
+
+
+class MockAggregator(Executor):
+    """A mock executor that aggregates results from multiple executors."""
+
+    @handler
+    async def mock_handler(self, messages: list[MockMessage], ctx: WorkflowContext) -> None:
+        # This mock simply returns the data incremented by 1
+        await ctx.add_event(WorkflowCompletedEvent(data=sum(msg.data for msg in messages)))
+
+
+@dataclass
+class ApprovalMessage:
+    """A mock message for approval requests."""
+
+    approved: bool
+
+
+class MockExecutorRequestApproval(Executor):
+    """A mock executor that simulates a request for approval."""
+
+    @handler(output_types=[RequestInfoMessage])
+    async def mock_handler_a(self, message: MockMessage, ctx: WorkflowContext) -> None:
+        """A mock handler that requests approval."""
+        await ctx.set_shared_state(self.id, message.data)
+        await ctx.send_message(RequestInfoMessage())
+
+    @handler(output_types=[MockMessage])
+    async def mock_handler_b(self, message: ApprovalMessage, ctx: WorkflowContext) -> None:
+        """A mock handler that processes the approval response."""
+        data = await ctx.get_shared_state(self.id)
+        if message.approved:
+            await ctx.add_event(WorkflowCompletedEvent(data=data))
+        else:
+            await ctx.send_message(MockMessage(data=data))
+
+
+async def test_workflow_run_streaming():
+    """Test the workflow run stream."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b")
+
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .add_edge(executor_a, executor_b)
+        .add_edge(executor_b, executor_a)
+        .build()
+    )
+
+    result: int | None = None
+    async for event in workflow.run_streaming(MockMessage(data=0)):
+        assert isinstance(event, WorkflowEvent)
+        if isinstance(event, WorkflowCompletedEvent):
+            result = event.data
+
+    assert result is not None and result == 10
+
+
+async def test_workflow_run_stream_not_completed():
+    """Test the workflow run stream."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b")
+
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .add_edge(executor_a, executor_b)
+        .add_edge(executor_b, executor_a)
+        .set_max_iterations(5)
+        .build()
+    )
+
+    with pytest.raises(RuntimeError):
+        async for _ in workflow.run_streaming(MockMessage(data=0)):
+            pass
+
+
+async def test_workflow_run():
+    """Test the workflow run."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b")
+
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .add_edge(executor_a, executor_b)
+        .add_edge(executor_b, executor_a)
+        .build()
+    )
+
+    events = await workflow.run(MockMessage(data=0))
+    completed_event = events.get_completed_event()
+    assert isinstance(completed_event, WorkflowCompletedEvent)
+    assert completed_event.data == 10
+
+
+async def test_workflow_run_not_completed():
+    """Test the workflow run."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b")
+
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .add_edge(executor_a, executor_b)
+        .add_edge(executor_b, executor_a)
+        .set_max_iterations(5)
+        .build()
+    )
+
+    with pytest.raises(RuntimeError):
+        await workflow.run(MockMessage(data=0))
+
+
+async def test_workflow_send_responses_streaming():
+    """Test the workflow run with approval."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutorRequestApproval(id="executor_b")
+    request_info_executor = RequestInfoExecutor()
+
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .add_edge(executor_a, executor_b)
+        .add_edge(executor_b, executor_a)
+        .add_edge(executor_b, request_info_executor)
+        .add_edge(request_info_executor, executor_b)
+        .build()
+    )
+
+    request_info_event: RequestInfoEvent | None = None
+    async for event in workflow.run_streaming(MockMessage(data=0)):
+        if isinstance(event, RequestInfoEvent):
+            request_info_event = event
+
+    assert request_info_event is not None
+    result: int | None = None
+    async for event in workflow.send_responses_streaming({
+        request_info_event.request_id: ApprovalMessage(approved=True)
+    }):
+        if isinstance(event, WorkflowCompletedEvent):
+            result = event.data
+
+    assert result is not None and result == 1  # The data should be incremented by 1 from the initial message
+
+
+async def test_workflow_send_responses():
+    """Test the workflow run with approval."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutorRequestApproval(id="executor_b")
+    request_info_executor = RequestInfoExecutor()
+
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .add_edge(executor_a, executor_b)
+        .add_edge(executor_b, executor_a)
+        .add_edge(executor_b, request_info_executor)
+        .add_edge(request_info_executor, executor_b)
+        .build()
+    )
+
+    events = await workflow.run(MockMessage(data=0))
+    request_info_events = events.get_request_info_events()
+
+    assert len(request_info_events) == 1
+
+    result = await workflow.send_responses({request_info_events[0].request_id: ApprovalMessage(approved=True)})
+
+    completed_event = result.get_completed_event()
+    assert isinstance(completed_event, WorkflowCompletedEvent)
+    assert completed_event.data == 1  # The data should be incremented by 1 from the initial message
+
+
+async def test_fan_out():
+    """Test a fan-out workflow."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b", limit=1)
+    executor_c = MockExecutor(id="executor_c", limit=2)  # This executor will not complete the workflow
+
+    workflow = (
+        WorkflowBuilder().set_start_executor(executor_a).add_fan_out_edges(executor_a, [executor_b, executor_c]).build()
+    )
+
+    events = await workflow.run(MockMessage(data=0))
+
+    # Each executor will emit two events: ExecutorInvokeEvent and ExecutorCompletedEvent
+    # executor_b will also emit a WorkflowCompletedEvent
+    assert len(events) == 7
+
+    completed_event = events.get_completed_event()
+    assert completed_event is not None and completed_event.data == 1
+
+
+async def test_fan_out_multiple_completed_events():
+    """Test a fan-out workflow with multiple completed events."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b", limit=1)
+    executor_c = MockExecutor(id="executor_c", limit=1)
+
+    workflow = (
+        WorkflowBuilder().set_start_executor(executor_a).add_fan_out_edges(executor_a, [executor_b, executor_c]).build()
+    )
+
+    events = await workflow.run(MockMessage(data=0))
+
+    # Each executor will emit two events: ExecutorInvokeEvent and ExecutorCompletedEvent
+    # executor_a and executor_b will also emit a WorkflowCompletedEvent
+    assert len(events) == 8
+
+    with pytest.raises(ValueError):
+        events.get_completed_event()
+
+
+async def test_fan_in():
+    """Test a fan-in workflow."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b")
+    executor_c = MockExecutor(id="executor_c")
+    aggregator = MockAggregator(id="aggregator")
+
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .add_fan_out_edges(executor_a, [executor_b, executor_c])
+        .add_fan_in_edges([executor_b, executor_c], aggregator)
+        .build()
+    )
+
+    events = await workflow.run(MockMessage(data=0))
+
+    # Each executor will emit two events: ExecutorInvokeEvent and ExecutorCompletedEvent
+    # aggregator will also emit a WorkflowCompletedEvent
+    assert len(events) == 9
+
+    completed_event = events.get_completed_event()
+    assert completed_event is not None and completed_event.data == 4

python/packages/workflow/tests/test_workflow_builder.py

@@ -0,0 +1,65 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from dataclasses import dataclass
+from typing import Any
+
+import pytest
+from agent_framework.workflow import Executor, WorkflowBuilder, WorkflowContext, handler
+
+
+@dataclass
+class MockMessage:
+    """A mock message for testing purposes."""
+
+    data: Any
+
+
+class MockExecutor(Executor):
+    """A mock executor for testing purposes."""
+
+    @handler(output_types=[MockMessage])
+    async def mock_handler(self, message: MockMessage, ctx: WorkflowContext) -> None:
+        """A mock handler that does nothing."""
+        pass
+
+
+class MockAggregator(Executor):
+    """A mock executor that aggregates results from multiple executors."""
+
+    @handler(output_types=[MockMessage])
+    async def mock_handler(self, messages: list[MockMessage], ctx: WorkflowContext) -> None:
+        # This mock simply returns the data incremented by 1
+        pass
+
+
+def test_workflow_builder_without_start_executor_throws():
+    """Test creating a workflow builder without a start executor."""
+
+    builder = WorkflowBuilder()
+    with pytest.raises(ValueError):
+        builder.build()
+
+
+def test_workflow_builder_fluent_api():
+    """Test the fluent API of the workflow builder."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b")
+    executor_c = MockExecutor(id="executor_c")
+    executor_d = MockExecutor(id="executor_d")
+    executor_e = MockAggregator(id="executor_e")
+    executor_f = MockExecutor(id="executor_f")
+
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .add_edge(executor_a, executor_b)
+        .add_fan_out_edges(executor_b, [executor_c, executor_d])
+        .add_fan_in_edges([executor_c, executor_d], executor_e)
+        .add_chain([executor_e, executor_f])
+        .set_max_iterations(5)
+        .build()
+    )
+
+    assert len(workflow.edges) == 6
+    assert workflow.start_executor.id == executor_a.id
+    assert len(workflow.executors) == 6