Python: Improve the workflow getting started samples (#570)

* Wip: samples

* wip - samples

* Updates to workflow getting started samples

* Checkpointing enhancements

* Cleanup

* PR feedback

* Updates

* Sample updates

* Updates

* Revamp samples, improve doc strings and code comments

* Cleanup unused comment

* Formatting cleanup

* wip

* Further work on samples. Allow agent to be specified as edge.

* Cleanup

* Typing cleanup

* Sample updates

---------

Co-authored-by: Chris <66376200+crickman@users.noreply.github.com>
Co-authored-by: Eric Zhu <ekzhu@users.noreply.github.com>

python/packages/workflow/agent_framework_workflow/_executor.py

@@ -777,11 +777,15 @@ class AgentExecutorResponse:
 
     Attributes:
         executor_id: The ID of the executor that generated the response.
-        response: The agent run response containing the messages generated by the agent.
+    agent_run_response: The underlying agent run response (unaltered from client).
+    full_conversation: The full conversation context (prior inputs + all assistant/tool outputs) that
+        should be used when chaining to another AgentExecutor. This prevents downstream agents losing
+        user prompts while keeping the emitted AgentRunEvent text faithful to the raw agent output.
     """
 
     executor_id: str
     agent_run_response: AgentRunResponse
+    full_conversation: list[ChatMessage] | None = None
 
 
 class AgentExecutor(Executor):
@@ -800,21 +804,75 @@ class AgentExecutor(Executor):
         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.
+            streaming: Enable streaming (emits incremental AgentRunUpdateEvent events) vs single response.
             id: A unique identifier for the executor. If None, a new UUID will be generated.
         """
-        super().__init__(id or agent.id)
+        # Prefer provided id; else use agent.name if present; else generate deterministic prefix
+        if id is not None:
+            exec_id = id
+        else:
+            agent_name = agent.name
+            exec_id = str(agent_name) if agent_name else f"executor_{uuid.uuid4()}"
+        super().__init__(exec_id)
         self._agent = agent
         self._agent_thread = agent_thread or self._agent.get_new_thread()
         self._streaming = streaming
         self._cache: list[ChatMessage] = []
 
+    async def _run_agent_and_emit(self, ctx: WorkflowContext[AgentExecutorResponse]) -> None:
+        """Execute the underlying agent, emit events, and enqueue response.
+
+        Terminal detection & WorkflowCompletedEvent emission are handled centrally in Runner.
+        This method only produces AgentRunEvent/AgentRunUpdateEvent plus enqueues an
+        AgentExecutorResponse message for routing.
+        """
+        if self._streaming:
+            updates: list[AgentRunResponseUpdate] = []
+            async for update in self._agent.run_stream(
+                self._cache,
+                thread=self._agent_thread,
+            ):
+                # Skip empty updates (no textual or structural content)
+                if not update:
+                    continue
+                contents = getattr(update, "contents", None)
+                text_val = getattr(update, "text", "")
+                has_text_content = False
+                if contents:
+                    for c in contents:
+                        if getattr(c, "text", None):
+                            has_text_content = True
+                            break
+                if not (text_val or has_text_content):
+                    continue
+                updates.append(update)
+                await ctx.add_event(AgentRunUpdateEvent(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))
+
+        full_conversation: list[ChatMessage] | None = None
+        if self._cache:
+            # Construct conversation snapshot = inputs (cache) + agent outputs (agent_run_response.messages).
+            # Do not mutate response.messages so AgentRunEvent remains clean.
+            full_conversation = list(self._cache) + list(response.messages)
+
+        agent_response = AgentExecutorResponse(self.id, response, full_conversation=full_conversation)
+        await ctx.send_message(agent_response)
+        self._cache.clear()
+
     @handler
     async def run(self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorResponse]) -> None:
-        """Run the agent executor with the given request."""
-        self._cache.extend(request.messages)
+        """Handle an AgentExecutorRequest (canonical input).
 
+        This is the standard path: extend cache with provided messages; if should_respond
+        run the agent and emit an AgentExecutorResponse downstream.
+        """
+        self._cache.extend(request.messages)
         if request.should_respond:
             if self._streaming:
                 updates: list[AgentRunResponseUpdate] = []
@@ -822,6 +880,18 @@ class AgentExecutor(Executor):
                     self._cache,
                     thread=self._agent_thread,
                 ):
+                    if not update:
+                        continue
+                    contents = getattr(update, "contents", None)
+                    text_val = getattr(update, "text", "")
+                    has_text_content = False
+                    if contents:
+                        for c in contents:
+                            if getattr(c, "text", None):
+                                has_text_content = True
+                                break
+                    if not (text_val or has_text_content):
+                        continue
                     updates.append(update)
                     await ctx.add_event(AgentRunUpdateEvent(self.id, update))
                 response = AgentRunResponse.from_agent_run_response_updates(updates)
@@ -832,8 +902,37 @@ class AgentExecutor(Executor):
                 )
                 await ctx.add_event(AgentRunEvent(self.id, response))
 
-            await ctx.send_message(AgentExecutorResponse(self.id, response))
-            self._cache.clear()
+    @handler
+    async def from_response(self, prior: AgentExecutorResponse, ctx: WorkflowContext[AgentExecutorResponse]) -> None:
+        """Enable seamless chaining: accept a prior AgentExecutorResponse as input.
+
+        Strategy: treat the prior response's messages as the conversation state and
+        immediately run the agent to produce a new response.
+        """
+        # Replace cache with full conversation if available, else fall back to agent_run_response messages.
+        if prior.full_conversation is not None:
+            self._cache = list(prior.full_conversation)
+        else:
+            self._cache = list(prior.agent_run_response.messages)
+        await self._run_agent_and_emit(ctx)
+
+    @handler
+    async def from_str(self, text: str, ctx: WorkflowContext[AgentExecutorResponse]) -> None:
+        """Accept a raw user prompt string and run the agent (one-shot)."""
+        self._cache = [ChatMessage(role="user", text=text)]  # type: ignore[arg-type]
+        await self._run_agent_and_emit(ctx)
+
+    @handler
+    async def from_message(self, message: ChatMessage, ctx: WorkflowContext[AgentExecutorResponse]) -> None:  # type: ignore[name-defined]
+        """Accept a single ChatMessage as input."""
+        self._cache = [message]
+        await self._run_agent_and_emit(ctx)
+
+    @handler
+    async def from_messages(self, messages: list[ChatMessage], ctx: WorkflowContext[AgentExecutorResponse]) -> None:  # type: ignore[name-defined]
+        """Accept a list of ChatMessage objects as conversation context."""
+        self._cache = list(messages)
+        await self._run_agent_and_emit(ctx)
 
 
 # endregion: Agent Executor

python/packages/workflow/agent_framework_workflow/_function_executor.py

@@ -10,8 +10,6 @@ This module provides:
   with proper type validation and handler registration.
 """
 
-from __future__ import annotations
-
 import asyncio
 import inspect
 from collections.abc import Awaitable, Callable

python/packages/workflow/agent_framework_workflow/_runner.py

@@ -3,7 +3,7 @@
 import asyncio
 import logging
 from collections import defaultdict
-from collections.abc import AsyncIterable, Sequence
+from collections.abc import AsyncGenerator, Sequence
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
@@ -11,7 +11,7 @@ if TYPE_CHECKING:
 
 from ._edge import EdgeGroup
 from ._edge_runner import EdgeRunner, create_edge_runner
-from ._events import WorkflowEvent
+from ._events import WorkflowCompletedEvent, WorkflowEvent
 from ._executor import Executor
 from ._runner_context import Message, RunnerContext
 from ._shared_state import SharedState
@@ -74,19 +74,17 @@ class Runner:
         if max_iterations is not None:
             self._max_iterations = max_iterations
 
-    async def run_until_convergence(self) -> AsyncIterable[WorkflowEvent]:
+    async def run_until_convergence(self) -> AsyncGenerator[WorkflowEvent, None]:
         """Run the workflow until no more messages are sent."""
         if self._running:
             raise RuntimeError("Runner is already running.")
 
         self._running = True
         try:
-            # Process any events from initial execution before checkpointing
+            # Emit any events already produced prior to entering loop
             if await self._ctx.has_events():
-                logger.info("Processing events from initial execution")
-                events = await self._ctx.drain_events()
-                for event in events:
-                    logger.info(f"Yielding initial event: {event}")
+                logger.info("Yielding pre-loop events")
+                for event in await self._ctx.drain_events():
                     yield event
 
             # Create first checkpoint if there are messages from initial execution
@@ -102,22 +100,33 @@ class Runner:
 
             while self._iteration < self._max_iterations:
                 logger.info(f"Starting superstep {self._iteration + 1}")
-                await self._run_iteration()
+
+                # Run iteration concurrently with live event streaming: we poll
+                # for new events while the iteration coroutine progresses.
+                iteration_task = asyncio.create_task(self._run_iteration())
+                while not iteration_task.done():
+                    try:
+                        # Wait briefly for any new event; timeout allows progress checks
+                        event = await asyncio.wait_for(self._ctx.next_event(), timeout=0.05)
+                        yield event
+                    except asyncio.TimeoutError:
+                        # Periodically continue to let iteration advance
+                        continue
+
+                # Propagate errors from iteration
+                await iteration_task
                 self._iteration += 1
 
+                # Drain any straggler events emitted at tail end
+                if await self._ctx.has_events():
+                    for event in await self._ctx.drain_events():
+                        yield event
+
                 # Update context with current iteration state immediately
                 await self._update_context_with_shared_state()
 
                 logger.info(f"Completed superstep {self._iteration}")
 
-                # Process events first before any checkpointing
-                if await self._ctx.has_events():
-                    logger.info("Processing events before checkpointing")
-                    events = await self._ctx.drain_events()
-                    for event in events:
-                        logger.debug(f"Yielding event: {event}")
-                        yield event
-
                 # Create checkpoint after each superstep iteration
                 await self._create_checkpoint_if_enabled(f"superstep_{self._iteration}")
 
@@ -142,7 +151,7 @@ class Runner:
                 from ._executor import SubWorkflowRequestInfo
 
                 # Handle SubWorkflowRequestInfo messages - only process those not already targeted
-                sub_workflow_messages = []
+                sub_workflow_messages: list[Message] = []
                 for msg in messages:
                     # Skip messages sent directly to RequestInfoExecutor - they are already forwarded
                     if self._is_message_to_request_info_executor(msg):
@@ -152,14 +161,15 @@ class Runner:
                         sub_workflow_messages.append(msg)
 
                 for message in sub_workflow_messages:
-                    sub_request = message.data
+                    # message.data is guaranteed to be SubWorkflowRequestInfo via filtering above
+                    sub_request = message.data  # type: ignore[assignment]
 
                     # Find executor that can intercept the wrapped type
                     interceptor_found = False
                     for executor in self._executors.values():
-                        if hasattr(executor, "_request_interceptors") and executor.id != message.source_id:
-                            # Check if any registered interceptor can handle this request type
-                            for registered_type in executor._request_interceptors:
+                        interceptors = getattr(executor, "_request_interceptors", [])
+                        if interceptors and executor.id != message.source_id:
+                            for registered_type in interceptors:  # type: ignore[assignment]
                                 # Check type matching - handle both type and string cases
                                 matched = False
                                 if (
@@ -234,7 +244,7 @@ class Runner:
             # since they were handled specially
             from ._executor import SubWorkflowRequestInfo
 
-            non_sub_workflow_messages = []
+            non_sub_workflow_messages: list[Message] = []
             for msg in messages:
                 # Keep messages sent directly to RequestInfoExecutor (forwarded messages)
                 if self._is_message_to_request_info_executor(msg):
@@ -251,8 +261,43 @@ class Runner:
             for message in non_sub_workflow_messages:
                 # Deliver a message through all edge runners associated with the source executor concurrently.
                 tasks = [_deliver_message_inner(edge_runner, message) for edge_runner in associated_edge_runners]
+                if not tasks:
+                    # No outgoing edges. If this is an AgentExecutorResponse, treat it as an
+                    # intentional terminal emission and emit a WorkflowCompletedEvent here.
+                    # (Previously this relied on the executor to emit, but AgentExecutor only
+                    # sends an AgentExecutorResponse message; centralized completion keeps the
+                    # contract consistent with other executors.)
+                    try:  # Local import to avoid circular dependencies at module import time.
+                        from ._executor import AgentExecutorResponse  # type: ignore
+
+                        if isinstance(message.data, AgentExecutorResponse):
+                            final_messages = message.data.agent_run_response.messages
+                            final_text = final_messages[-1].text if final_messages else "(no content)"
+                            await self._ctx.add_event(WorkflowCompletedEvent(final_text))
+                            continue  # Terminal handled
+                    except Exception as exc:  # pragma: no cover - defensive
+                        logger.debug("Suppressed exception during terminal message type check: %s", exc)
+                    # Otherwise keep prior behavior (emit warning for unexpected undelivered message).
+                    logger.warning(
+                        f"Message {message} could not be delivered (no outgoing edges). "
+                        "Add a downstream executor or remove the send if this is unexpected."
+                    )
+                    continue
                 results = await asyncio.gather(*tasks)
                 if not any(results):
+                    # Outgoing edges exist but none accepted the message. If this is an
+                    # AgentExecutorResponse, treat as natural terminal and emit completion.
+                    try:
+                        from ._executor import AgentExecutorResponse  # type: ignore
+
+                        if isinstance(message.data, AgentExecutorResponse):
+                            # Emit a single completion event with final text (best-effort extraction)
+                            final_messages = message.data.agent_run_response.messages
+                            final_text = final_messages[-1].text if final_messages else "(no content)"
+                            await self._ctx.add_event(WorkflowCompletedEvent(final_text))
+                            continue
+                    except Exception as exc:  # pragma: no cover
+                        logger.debug("Terminal completion emission failed: %s", exc)
                     logger.warning(
                         f"Message {message} could not be delivered. "
                         "This may be due to type incompatibility or no matching targets."
@@ -389,7 +434,8 @@ class Runner:
         """
         parsed: defaultdict[str, list[EdgeRunner]] = defaultdict(list)
         for runner in edge_runners:
-            for source_executor_id in runner._edge_group.source_executor_ids:
+            # Accessing protected attribute (_edge_group) intentionally for internal wiring.
+            for source_executor_id in runner._edge_group.source_executor_ids:  # type: ignore[attr-defined]
                 parsed[source_executor_id].append(runner)
 
         return parsed

python/packages/workflow/agent_framework_workflow/_runner_context.py

@@ -1,14 +1,16 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+import asyncio
+import importlib
 import logging
 import uuid
 from collections import defaultdict
-from dataclasses import dataclass
-from typing import Any, Protocol, TypedDict, TypeVar, runtime_checkable
+from dataclasses import dataclass, fields, is_dataclass
+from typing import Any, Protocol, TypedDict, TypeVar, cast, runtime_checkable
 
 from ._checkpoint import CheckpointStorage, WorkflowCheckpoint
 from ._const import DEFAULT_MAX_ITERATIONS
-from ._events import WorkflowEvent
+from ._events import AgentRunUpdateEvent, WorkflowEvent
 from ._shared_state import SharedState
 
 logger = logging.getLogger(__name__)
@@ -49,6 +51,176 @@ class CheckpointState(TypedDict):
     max_iterations: int
 
 
+# Checkpoint serialization helpers
+_PYDANTIC_MARKER = "__af_pydantic_model__"
+_DATACLASS_MARKER = "__af_dataclass__"
+
+# Guards to prevent runaway recursion while encoding arbitrary user data
+_MAX_ENCODE_DEPTH = 100
+_CYCLE_SENTINEL = "<cycle>"
+
+
+def _is_pydantic_model(obj: object) -> bool:
+    """Best-effort check for Pydantic models (e.g., AFBaseModel).
+
+    We avoid hard dependencies by duck-typing on model_dump/model_validate.
+    """
+    try:
+        obj_type: type[Any] = type(obj)
+        return hasattr(obj, "model_dump") and hasattr(obj_type, "model_validate")
+    except Exception:
+        return False
+
+
+def _encode_checkpoint_value(value: Any) -> Any:
+    """Recursively encode values into JSON-serializable structures.
+
+    - Pydantic models -> { _PYDANTIC_MARKER: "module:Class", value: model_dump(mode="json") }
+    - dataclass instances -> { _DATACLASS_MARKER: "module:Class", value: {field: encoded} }
+    - dict -> encode keys as str and values recursively
+    - list/tuple/set -> list of encoded items
+    - other -> returned as-is if already JSON-serializable
+
+    Includes cycle and depth protection to avoid infinite recursion.
+    """
+
+    def _enc(v: Any, stack: set[int], depth: int) -> Any:
+        # Depth guard
+        if depth > _MAX_ENCODE_DEPTH:
+            logger.debug(f"Max encode depth reached at depth={depth} for type={type(v)}")
+            return "<max_depth>"
+
+        # Pydantic (AFBaseModel) handling
+        if _is_pydantic_model(v):
+            cls = cast(type[Any], type(v))  # type: ignore
+            try:
+                return {
+                    _PYDANTIC_MARKER: f"{cls.__module__}:{cls.__name__}",
+                    "value": v.model_dump(mode="json"),
+                }
+            except Exception as exc:  # best-effort fallback
+                logger.debug("Pydantic model_dump failed for %s: %s", cls, exc)
+                return str(v)
+
+        # Dataclasses (instances only)
+        if is_dataclass(v) and not isinstance(v, type):
+            oid = id(v)
+            if oid in stack:
+                logger.debug("Cycle detected while encoding dataclass instance")
+                return _CYCLE_SENTINEL
+            stack.add(oid)
+            try:
+                # type(v) already narrows sufficiently; cast was redundant
+                dc_cls: type[Any] = type(v)
+                field_values: dict[str, Any] = {}
+                for f in fields(v):  # type: ignore[arg-type]
+                    field_values[f.name] = _enc(getattr(v, f.name), stack, depth + 1)
+                return {
+                    _DATACLASS_MARKER: f"{dc_cls.__module__}:{dc_cls.__name__}",
+                    "value": field_values,
+                }
+            finally:
+                stack.remove(oid)
+
+        # Collections
+        if isinstance(v, dict):
+            v_dict = cast("dict[object, object]", v)
+            oid = id(v_dict)
+            if oid in stack:
+                logger.debug("Cycle detected while encoding dict")
+                return _CYCLE_SENTINEL
+            stack.add(oid)
+            try:
+                json_dict: dict[str, Any] = {}
+                for k_any, val_any in v_dict.items():  # type: ignore[assignment]
+                    k_str: str = str(k_any)
+                    json_dict[k_str] = _enc(val_any, stack, depth + 1)
+                return json_dict
+            finally:
+                stack.remove(oid)
+
+        if isinstance(v, (list, tuple, set)):
+            iterable_v = cast("list[object] | tuple[object, ...] | set[object]", v)
+            oid = id(iterable_v)
+            if oid in stack:
+                logger.debug("Cycle detected while encoding iterable")
+                return _CYCLE_SENTINEL
+            stack.add(oid)
+            try:
+                seq: list[object] = list(iterable_v)
+                encoded_list: list[Any] = []
+                for item in seq:
+                    encoded_list.append(_enc(item, stack, depth + 1))
+                return encoded_list
+            finally:
+                stack.remove(oid)
+
+        # Primitives (or unknown objects): ensure JSON-serializable
+        if isinstance(v, (str, int, float, bool)) or v is None:
+            return v
+        # Fallback: stringify unknown objects to avoid JSON serialization errors
+        try:
+            return str(v)
+        except Exception:
+            return f"<{type(v).__name__}>"
+
+    return _enc(value, set(), 0)
+
+
+def _decode_checkpoint_value(value: Any) -> Any:
+    """Recursively decode values previously encoded by _encode_checkpoint_value."""
+    if isinstance(value, dict):
+        value_dict = cast(dict[str, Any], value)  # encoded form always uses string keys
+        # Pydantic marker handling
+        if _PYDANTIC_MARKER in value_dict and "value" in value_dict:
+            type_key: str | None = value_dict.get(_PYDANTIC_MARKER)  # type: ignore[assignment]
+            raw: Any = value_dict.get("value")
+            if isinstance(type_key, str):
+                try:
+                    module_name, class_name = type_key.split(":", 1)
+                    module = importlib.import_module(module_name)
+                    cls: Any = getattr(module, class_name)
+                    if hasattr(cls, "model_validate"):
+                        return cls.model_validate(raw)
+                except Exception as exc:
+                    logger.debug(
+                        "Failed to decode pydantic model %s: %s; returning raw value",
+                        type_key,
+                        exc,
+                    )
+        # Dataclass marker handling
+        if _DATACLASS_MARKER in value_dict and "value" in value_dict:
+            type_key_dc: str | None = value_dict.get(_DATACLASS_MARKER)  # type: ignore[assignment]
+            raw_dc: Any = value_dict.get("value")
+            if isinstance(type_key_dc, str):
+                try:
+                    module_name, class_name = type_key_dc.split(":", 1)
+                    module = importlib.import_module(module_name)
+                    cls_dc: Any = getattr(module, class_name)
+                    decoded_raw = _decode_checkpoint_value(raw_dc)
+                    if isinstance(decoded_raw, dict):
+                        return cls_dc(**decoded_raw)
+                except Exception as exc:
+                    logger.debug(
+                        "Failed to decode dataclass %s: %s; returning raw value",
+                        type_key_dc,
+                        exc,
+                    )
+            # Fallback to decoded raw value
+            return _decode_checkpoint_value(raw_dc)
+
+        # Regular dict: decode recursively
+        decoded: dict[str, Any] = {}
+        for k_any, v_any in value_dict.items():
+            decoded[k_any] = _decode_checkpoint_value(v_any)
+        return decoded
+    if isinstance(value, list):
+        # After isinstance check, treat value as list[Any] for decoding
+        value_list: list[Any] = value  # type: ignore[assignment]
+        return [_decode_checkpoint_value(v_any) for v_any in value_list]
+    return value
+
+
 @runtime_checkable
 class RunnerContext(Protocol):
     """Protocol for the execution context used by the runner.
@@ -105,6 +277,10 @@ class RunnerContext(Protocol):
         """
         ...
 
+    async def next_event(self) -> WorkflowEvent:  # pragma: no cover - interface only
+        """Wait for and return the next event emitted by the workflow run."""
+        ...
+
     async def set_state(self, executor_id: str, state: dict[str, Any]) -> None:
         """Set the state for a specific executor.
 
@@ -185,7 +361,8 @@ class InProcRunnerContext:
             checkpoint_storage: Optional storage to enable checkpointing.
         """
         self._messages: defaultdict[str, list[Message]] = defaultdict(list)
-        self._events: list[WorkflowEvent] = []
+        # Event queue for immediate streaming of events (e.g., AgentRunUpdateEvent)
+        self._event_queue: asyncio.Queue[WorkflowEvent] = asyncio.Queue()
 
         # Checkpointing configuration/state
         self._checkpoint_storage = checkpoint_storage
@@ -207,15 +384,54 @@ class InProcRunnerContext:
         return bool(self._messages)
 
     async def add_event(self, event: WorkflowEvent) -> None:
-        self._events.append(event)
+        """Add an event to the context immediately.
+
+        Events are enqueued so runners can stream them in real time instead of
+        waiting for superstep boundaries.
+        """
+        # Filter out empty AgentRunUpdateEvent updates to avoid emitting None/empty chunks
+        try:
+            if isinstance(event, AgentRunUpdateEvent):
+                update = getattr(event, "data", None)
+                # Skip if no update payload
+                if not update:
+                    return
+                # Robust emptiness check: allow either top-level text or any text-bearing content
+                text_val = getattr(update, "text", None)
+                contents = getattr(update, "contents", None)
+                has_text_content = False
+                if contents:
+                    for c in contents:
+                        if getattr(c, "text", None):
+                            has_text_content = True
+                            break
+                if not (text_val or has_text_content):
+                    return
+        except Exception as exc:  # pragma: no cover - defensive logging path
+            # Best-effort filtering only; never block event delivery on filtering errors
+            logger.debug("Error while filtering event %r: %s", event, exc, exc_info=True)
+
+        await self._event_queue.put(event)
 
     async def drain_events(self) -> list[WorkflowEvent]:
-        events = self._events.copy()
-        self._events.clear()
+        """Drain all currently queued events without blocking for new ones."""
+        events: list[WorkflowEvent] = []
+        while True:
+            try:
+                events.append(self._event_queue.get_nowait())
+            except asyncio.QueueEmpty:  # type: ignore[attr-defined]
+                break
         return events
 
     async def has_events(self) -> bool:
-        return bool(self._events)
+        return not self._event_queue.empty()
+
+    async def next_event(self) -> WorkflowEvent:
+        """Wait for and return the next event.
+
+        Used by the runner to interleave event emission with ongoing iteration work.
+        """
+        return await self._event_queue.get()
 
     async def set_state(self, executor_id: str, state: dict[str, Any]) -> None:
         self._executor_states[executor_id] = state
@@ -229,9 +445,10 @@ class InProcRunnerContext:
     def set_workflow_id(self, workflow_id: str) -> None:
         self._workflow_id = workflow_id
 
-    def reset_for_new_run(self, workflow_shared_state: "SharedState | None" = None) -> None:
+    def reset_for_new_run(self, workflow_shared_state: SharedState | None = None) -> None:
         self._messages.clear()
-        self._events.clear()
+        # Clear any pending events (best-effort) by recreating the queue
+        self._event_queue = asyncio.Queue()
         self._shared_state.clear()
         self._executor_states.clear()
         self._iteration_count = 0
@@ -285,7 +502,7 @@ class InProcRunnerContext:
         for source_id, message_list in self._messages.items():
             serializable_messages[source_id] = [
                 {
-                    "data": msg.data,
+                    "data": _encode_checkpoint_value(msg.data),
                     "source_id": msg.source_id,
                     "target_id": msg.target_id,
                     "trace_contexts": msg.trace_contexts,
@@ -295,8 +512,8 @@ class InProcRunnerContext:
             ]
         return {
             "messages": serializable_messages,
-            "shared_state": self._shared_state,
-            "executor_states": self._executor_states,
+            "shared_state": _encode_checkpoint_value(self._shared_state),
+            "executor_states": _encode_checkpoint_value(self._executor_states),
             "iteration_count": self._iteration_count,
             "max_iterations": self._max_iterations,
         }
@@ -307,7 +524,7 @@ class InProcRunnerContext:
         for source_id, message_list in messages_data.items():
             self._messages[source_id] = [
                 Message(
-                    data=msg.get("data"),
+                    data=_decode_checkpoint_value(msg.get("data")),
                     source_id=msg.get("source_id", ""),
                     target_id=msg.get("target_id"),
                     trace_contexts=msg.get("trace_contexts"),
@@ -315,7 +532,28 @@ class InProcRunnerContext:
                 )
                 for msg in message_list
             ]
-        self._shared_state = state.get("shared_state", {})
-        self._executor_states = state.get("executor_states", {})
+        # Restore shared_state
+        decoded_shared_raw = _decode_checkpoint_value(state.get("shared_state", {}))
+        if isinstance(decoded_shared_raw, dict):
+            self._shared_state = cast(dict[str, Any], decoded_shared_raw)
+        else:  # fallback to empty dict if corrupted
+            self._shared_state = {}
+
+        # Restore executor_states ensuring value types are dicts
+        decoded_exec_raw = _decode_checkpoint_value(state.get("executor_states", {}))
+        if isinstance(decoded_exec_raw, dict):
+            typed_exec: dict[str, dict[str, Any]] = {}
+            for k_raw, v_raw in decoded_exec_raw.items():  # type: ignore[assignment]
+                if isinstance(k_raw, str) and isinstance(v_raw, dict):
+                    # Filter inner dict to string keys only (best-effort)
+                    inner: dict[str, Any] = {}
+                    for inner_k, inner_v in v_raw.items():  # type: ignore[assignment]
+                        if isinstance(inner_k, str):
+                            inner[inner_k] = inner_v
+                    typed_exec[k_raw] = inner
+            self._executor_states = typed_exec
+        else:
+            self._executor_states = {}
+
         self._iteration_count = state.get("iteration_count", 0)
         self._max_iterations = state.get("max_iterations", 100)

python/packages/workflow/agent_framework_workflow/_validation.py

@@ -167,6 +167,23 @@ class WorkflowGraphValidator:
         if start_executor_id not in self._executors:
             raise GraphConnectivityError(f"Start executor '{start_executor_id}' is not present in the workflow graph")
 
+        # Additional presence verification:
+        # A start executor that is only injected via the builder (present in the executors map)
+        # but not referenced by any edge while other executors ARE referenced indicates a
+        # configuration error: the chosen start node is effectively disconnected / unknown to the
+        # defined graph topology. For single-node workflows (no edges) we allow the start executor
+        # to stand alone (handled above when we inject it into the map). We perform this refined
+        # check only when there is at least one edge group defined.
+        if self._edges:  # Only evaluate when the workflow defines edges
+            edge_executor_ids: set[str] = set()
+            for _e in self._edges:
+                edge_executor_ids.add(_e.source_id)
+                edge_executor_ids.add(_e.target_id)
+            if start_executor_id not in edge_executor_ids:
+                raise GraphConnectivityError(
+                    f"Start executor '{start_executor_id}' is not present in the workflow graph"
+                )
+
         # Run all checks
         self._validate_edge_duplication()
         self._validate_handler_output_annotations()

python/packages/workflow/agent_framework_workflow/_workflow.py

@@ -5,11 +5,13 @@ import logging
 import sys
 import uuid
 from collections.abc import AsyncIterable, Awaitable, Callable, Sequence
-from typing import TYPE_CHECKING, Any
+from typing import Any
 
+from agent_framework import AgentProtocol
 from agent_framework._pydantic import AFBaseModel
 from pydantic import Field
 
+from ._agent import WorkflowAgent
 from ._checkpoint import CheckpointStorage
 from ._const import DEFAULT_MAX_ITERATIONS
 from ._edge import (
@@ -24,7 +26,7 @@ from ._edge import (
     SwitchCaseEdgeGroupDefault,
 )
 from ._events import RequestInfoEvent, WorkflowCompletedEvent, WorkflowEvent
-from ._executor import Executor, RequestInfoExecutor
+from ._executor import AgentExecutor, Executor, RequestInfoExecutor
 from ._runner import Runner
 from ._runner_context import CheckpointState, InProcRunnerContext, RunnerContext
 from ._shared_state import SharedState
@@ -39,9 +41,6 @@ else:
 
 logger = logging.getLogger(__name__)
 
-if TYPE_CHECKING:  # Avoid runtime import cycles; enables proper type checking of as_agent return type
-    from ._agent import WorkflowAgent
-
 
 class WorkflowRunResult(list[WorkflowEvent]):
     """A list of events generated during the workflow execution in non-streaming mode."""
@@ -368,8 +367,48 @@ class Workflow(AFBaseModel):
         Returns:
             A WorkflowRunResult instance containing a list of events generated during the workflow execution.
         """
-        events = [event async for event in self.run_stream(message)]
-        return WorkflowRunResult(events)
+        from agent_framework import AgentRunResponse, AgentRunResponseUpdate
+
+        from ._events import AgentRunEvent, AgentRunUpdateEvent  # Local import to avoid cycles
+
+        raw_events = [event async for event in self.run_stream(message)]
+
+        # Coalesce streaming update events into a single AgentRunEvent per executor sequence.
+        coalesced: list[WorkflowEvent] = []  # type: ignore[name-defined]
+        pending_updates: list[AgentRunResponseUpdate] = []
+        pending_executor: str | None = None
+
+        def _flush_pending() -> None:
+            nonlocal pending_updates, pending_executor
+            if pending_executor is None or not pending_updates:
+                return
+            # Aggregate updates into a final AgentRunResponse using existing helper
+            aggregated = AgentRunResponse.from_agent_run_response_updates(pending_updates)
+            coalesced.append(AgentRunEvent(pending_executor, aggregated))
+            pending_updates = []
+            pending_executor = None
+
+        for ev in raw_events:
+            if isinstance(ev, AgentRunUpdateEvent):
+                # Start new grouping or continue existing if same executor
+                if pending_executor is None:
+                    pending_executor = ev.executor_id
+                if ev.executor_id != pending_executor:
+                    # Different executor encountered; flush previous first
+                    _flush_pending()
+                    pending_executor = ev.executor_id
+                if ev.data is not None:
+                    pending_updates.append(ev.data)
+                # Do NOT append update event itself (non-streaming contract)
+                continue
+            # Flush before adding any non-update event
+            _flush_pending()
+            coalesced.append(ev)
+
+        # Flush any trailing updates
+        _flush_pending()
+
+        return WorkflowRunResult(coalesced)
 
     async def run_from_checkpoint(
         self,
@@ -423,7 +462,7 @@ class Workflow(AFBaseModel):
             raise ValueError(f"Executor with ID {executor_id} not found.")
         return self.executors[executor_id]
 
-    def _find_request_info_executor(self) -> "RequestInfoExecutor | None":
+    def _find_request_info_executor(self) -> RequestInfoExecutor | None:
         """Find the RequestInfoExecutor instance in this workflow.
 
         Returns:
@@ -537,7 +576,7 @@ class Workflow(AFBaseModel):
                     )
                 )
 
-    def as_agent(self, name: str | None = None) -> "WorkflowAgent":
+    def as_agent(self, name: str | None = None) -> WorkflowAgent:
         """Create a WorkflowAgent that wraps this workflow.
 
         Args:
@@ -568,18 +607,57 @@ class WorkflowBuilder:
         self._start_executor: Executor | str | None = None
         self._checkpoint_storage: CheckpointStorage | None = None
         self._max_iterations: int = max_iterations
+        # Maps underlying AgentProtocol object id -> wrapped Executor so we reuse the same wrapper
+        # across set_start_executor / add_edge calls. Without this, unnamed agents (which receive
+        # random UUID based executor ids) end up wrapped multiple times, giving different ids for
+        # the start node vs edge nodes and triggering a GraphConnectivityError during validation.
+        self._agent_wrappers: dict[int, Executor] = {}
+
+    # Agents auto-wrapped by builder now always stream incremental updates.
 
     def _add_executor(self, executor: Executor) -> str:
         """Add an executor to the map and return its ID."""
         self._executors[executor.id] = executor
         return executor.id
 
+    def _maybe_wrap_agent(self, candidate: Executor | AgentProtocol) -> Executor:
+        """If the provided object implements AgentProtocol, wrap it in an AgentExecutor.
+
+        This allows fluent builder APIs to directly accept agents instead of
+        requiring callers to manually instantiate AgentExecutor.
+        """
+        try:  # Local import to avoid hard dependency at import time
+            from agent_framework import AgentProtocol  # type: ignore
+        except Exception:  # pragma: no cover - defensive
+            AgentProtocol = object  # type: ignore
+
+        if isinstance(candidate, Executor):  # Already an executor
+            return candidate
+        if isinstance(candidate, AgentProtocol):  # type: ignore[arg-type]
+            # Reuse existing wrapper for the same agent instance if present
+            existing = self._agent_wrappers.get(id(candidate))
+            if existing is not None:
+                return existing
+            # Use agent name if available and unique among current executors
+            name = getattr(candidate, "name", None)
+            proposed_id: str | None = None
+            if name:
+                proposed_id = str(name)
+                if proposed_id in self._executors:
+                    proposed_id = f"{proposed_id}-{uuid.uuid4().hex[:8]}"
+            wrapper = AgentExecutor(candidate, id=proposed_id, streaming=True)
+            self._agent_wrappers[id(candidate)] = wrapper
+            return wrapper
+        raise TypeError(
+            f"WorkflowBuilder expected an Executor or AgentProtocol instance; got {type(candidate).__name__}."
+        )
+
     def add_edge(
         self,
-        source: Executor,
-        target: Executor,
+        source: Executor | AgentProtocol,
+        target: Executor | AgentProtocol,
         condition: Callable[[Any], bool] | None = None,
-    ) -> "Self":
+    ) -> Self:
         """Add a directed edge between two executors.
 
         The output types of the source and the input types of the target must be compatible.
@@ -591,12 +669,18 @@ class WorkflowBuilder:
                        should be traversed based on the message type.
         """
         # TODO(@taochen): Support executor factories for lazy initialization
-        source_id = self._add_executor(source)
-        target_id = self._add_executor(target)
+        source_exec = self._maybe_wrap_agent(source)
+        target_exec = self._maybe_wrap_agent(target)
+        source_id = self._add_executor(source_exec)
+        target_id = self._add_executor(target_exec)
         self._edge_groups.append(SingleEdgeGroup(source_id, target_id, condition))
         return self
 
-    def add_fan_out_edges(self, source: Executor, targets: Sequence[Executor]) -> "Self":
+    def add_fan_out_edges(
+        self,
+        source: Executor | AgentProtocol,
+        targets: Sequence[Executor | AgentProtocol],
+    ) -> Self:
         """Add multiple edges to the workflow where messages from the source will be sent to all target.
 
         The output types of the source and the input types of the targets must be compatible.
@@ -605,13 +689,19 @@ class WorkflowBuilder:
             source: The source executor of the edges.
             targets: A list of target executors for the edges.
         """
-        source_id = self._add_executor(source)
-        target_ids = [self._add_executor(target) for target in targets]
+        source_exec = self._maybe_wrap_agent(source)
+        target_execs = [self._maybe_wrap_agent(t) for t in targets]
+        source_id = self._add_executor(source_exec)
+        target_ids = [self._add_executor(t) for t in target_execs]
         self._edge_groups.append(FanOutEdgeGroup(source_id, target_ids))
 
         return self
 
-    def add_switch_case_edge_group(self, source: Executor, cases: Sequence[Case | Default]) -> "Self":
+    def add_switch_case_edge_group(
+        self,
+        source: Executor | AgentProtocol,
+        cases: Sequence[Case | Default],
+    ) -> Self:
         """Add an edge group that represents a switch-case statement.
 
         The output types of the source and the input types of the targets must be compatible.
@@ -629,10 +719,13 @@ class WorkflowBuilder:
             source: The source executor of the edges.
             cases: A list of case objects that determine the target executor for each message.
         """
-        source_id = self._add_executor(source)
+        source_exec = self._maybe_wrap_agent(source)
+        source_id = self._add_executor(source_exec)
         # Convert case data types to internal types that only uses target_id.
         internal_cases: list[SwitchCaseEdgeGroupCase | SwitchCaseEdgeGroupDefault] = []
         for case in cases:
+            # Allow case targets to be agents
+            case.target = self._maybe_wrap_agent(case.target)  # type: ignore[attr-defined]
             self._add_executor(case.target)
             if isinstance(case, Default):
                 internal_cases.append(SwitchCaseEdgeGroupDefault(target_id=case.target.id))
@@ -644,10 +737,10 @@ class WorkflowBuilder:
 
     def add_multi_selection_edge_group(
         self,
-        source: Executor,
-        targets: Sequence[Executor],
+        source: Executor | AgentProtocol,
+        targets: Sequence[Executor | AgentProtocol],
         selection_func: Callable[[Any, list[str]], list[str]],
-    ) -> "Self":
+    ) -> Self:
         """Add an edge group that represents a multi-selection execution model.
 
         The output types of the source and the input types of the targets must be compatible.
@@ -662,13 +755,19 @@ class WorkflowBuilder:
             targets: A list of target executors for the edges.
             selection_func: A function that selects target executors for messages.
         """
-        source_id = self._add_executor(source)
-        target_ids = [self._add_executor(target) for target in targets]
+        source_exec = self._maybe_wrap_agent(source)
+        target_execs = [self._maybe_wrap_agent(t) for t in targets]
+        source_id = self._add_executor(source_exec)
+        target_ids = [self._add_executor(t) for t in target_execs]
         self._edge_groups.append(FanOutEdgeGroup(source_id, target_ids, selection_func))
 
         return self
 
-    def add_fan_in_edges(self, sources: Sequence[Executor], target: Executor) -> "Self":
+    def add_fan_in_edges(
+        self,
+        sources: Sequence[Executor | AgentProtocol],
+        target: Executor | AgentProtocol,
+    ) -> Self:
         """Add multiple edges from sources to a single target executor.
 
         The edges will be grouped together for synchronized processing, meaning
@@ -702,13 +801,15 @@ class WorkflowBuilder:
             sources: A list of source executors for the edges.
             target: The target executor for the edges.
         """
-        source_ids = [self._add_executor(source) for source in sources]
-        target_id = self._add_executor(target)
+        source_execs = [self._maybe_wrap_agent(s) for s in sources]
+        target_exec = self._maybe_wrap_agent(target)
+        source_ids = [self._add_executor(s) for s in source_execs]
+        target_id = self._add_executor(target_exec)
         self._edge_groups.append(FanInEdgeGroup(source_ids, target_id))
 
         return self
 
-    def add_chain(self, executors: Sequence[Executor]) -> "Self":
+    def add_chain(self, executors: Sequence[Executor | AgentProtocol]) -> 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.
@@ -719,20 +820,30 @@ class WorkflowBuilder:
         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])
+        # Wrap each candidate first to ensure stable IDs before adding edges
+        wrapped: list[Executor] = [self._maybe_wrap_agent(e) for e in executors]
+        for i in range(len(wrapped) - 1):
+            self.add_edge(wrapped[i], wrapped[i + 1])
         return self
 
-    def set_start_executor(self, executor: Executor | str) -> "Self":
+    def set_start_executor(self, executor: Executor | AgentProtocol | 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
+        if isinstance(executor, str):
+            self._start_executor = executor
+        else:
+            wrapped = self._maybe_wrap_agent(executor)  # type: ignore[arg-type]
+            self._start_executor = wrapped
+            # Ensure the start executor is present in the executor map so validation succeeds
+            # even if no edges are added yet, or before edges wrap the same agent again.
+            if wrapped.id not in self._executors:
+                self._executors[wrapped.id] = wrapped
         return self
 
-    def set_max_iterations(self, max_iterations: int) -> "Self":
+    def set_max_iterations(self, max_iterations: int) -> Self:
         """Set the maximum number of iterations for the workflow.
 
         Args:
@@ -741,7 +852,9 @@ class WorkflowBuilder:
         self._max_iterations = max_iterations
         return self
 
-    def with_checkpointing(self, checkpoint_storage: CheckpointStorage) -> "Self":
+    # Removed explicit set_agent_streaming() API; agents always stream updates.
+
+    def with_checkpointing(self, checkpoint_storage: CheckpointStorage) -> Self:
         """Enable checkpointing with the specified storage.
 
         Args:

python/packages/workflow/tests/test_validation.py

@@ -318,7 +318,7 @@ def test_logging_for_missing_input_types(caplog: Any) -> None:
 
     class NoInputTypesExecutor(Executor):
         # Handler without type annotation for input parameter
-        async def handle_message(self, message: Any, ctx: WorkflowContext) -> None:
+        async def handle_message(self, message: Any, ctx: WorkflowContext[Any]) -> None:
             await ctx.send_message("processed")
 
         def _discover_handlers(self) -> None:
@@ -581,7 +581,7 @@ def test_handler_ctx_missing_annotation_raises() -> None:
 def test_handler_ctx_unsubscripted_workflow_context_raises() -> None:
     class BadExecutor(Executor):
         @handler
-        async def handle(self, message: str, ctx: WorkflowContext) -> None:  # missing T
+        async def handle(self, message: str, ctx: WorkflowContext) -> None:  # type: ignore # missing T
             pass
 
     start = StringExecutor(id="s")

python/packages/workflow/tests/test_workflow_builder.py

@@ -4,7 +4,43 @@ from dataclasses import dataclass
 from typing import Any
 
 import pytest
-from agent_framework.workflow import Executor, WorkflowBuilder, WorkflowContext, handler
+from agent_framework import AgentRunResponse, AgentRunResponseUpdate, AgentThread, BaseAgent, ChatMessage, Role
+from agent_framework.workflow import AgentExecutor, Executor, WorkflowBuilder, WorkflowContext, handler
+
+
+class DummyAgent(BaseAgent):
+    async def run(self, messages=None, *, thread: AgentThread | None = None, **kwargs):  # type: ignore[override]
+        norm: list[ChatMessage] = []
+        if messages:
+            for m in messages:  # type: ignore[iteration-over-optional]
+                if isinstance(m, ChatMessage):
+                    norm.append(m)
+                elif isinstance(m, str):
+                    norm.append(ChatMessage(role=Role.USER, text=m))
+        return AgentRunResponse(messages=norm)
+
+    async def run_stream(self, messages=None, *, thread: AgentThread | None = None, **kwargs):  # type: ignore[override]
+        # Minimal async generator
+        yield AgentRunResponseUpdate()
+
+
+def test_builder_accepts_agents_directly():
+    agent1 = DummyAgent(id="agent1", name="writer")
+    agent2 = DummyAgent(id="agent2", name="reviewer")
+
+    wf = WorkflowBuilder().set_start_executor(agent1).add_edge(agent1, agent2).build()
+
+    # Confirm auto-wrapped executors use agent names as IDs
+    assert wf.start_executor_id == "writer"
+    assert any(isinstance(e, AgentExecutor) and e.id in {"writer", "reviewer"} for e in wf.executors.values())
+
+
+def test_builder_agents_always_stream():
+    agent = DummyAgent(id="agentX", name="streamer")
+    wf = WorkflowBuilder().set_start_executor(agent).build()
+    exec_obj = wf.get_start_executor()
+    assert isinstance(exec_obj, AgentExecutor)
+    assert getattr(exec_obj, "_streaming", False) is True
 
 
 @dataclass