Principle:Microsoft Agent framework Workflow Streaming Execution

Property	Value
Principle Name	Workflow Streaming Execution
SDK	Microsoft Agent Framework
Repository	Microsoft Agent Framework
Source Reference	python/packages/core/agent_framework/_workflows/_workflow.py:L449-535
Import	from agent_framework import Workflow
Domains	Agent_Architecture, Workflow_Engine

Overview

Workflow Streaming Execution is the principle governing how workflows emit real-time events during execution. Calling Workflow.run(stream=True) returns a ResponseStream[WorkflowEvent, WorkflowRunResult] that yields WorkflowEvent objects as the workflow progresses, enabling callers to monitor executor invocations, completions, outputs, and errors in real time. This pattern applies uniformly to sequential, concurrent, and graph-based workflows.

Description

Workflow Streaming Execution defines the contract between the workflow engine and its callers for incremental event delivery. Rather than waiting for an entire workflow run to complete before returning results, the streaming mode surfaces each significant lifecycle transition as it occurs. This makes it possible to build responsive user interfaces, emit telemetry, and implement early-exit logic without polling or post-hoc inspection of result objects.

The streaming interface is activated by passing stream=True to Workflow.run(). When enabled, the method returns a ResponseStream that the caller iterates with async for. Each iteration yields a WorkflowEvent whose type discriminator identifies the kind of event. After the stream is exhausted, the caller can obtain the final WorkflowRunResult via get_final_response().

When streaming is disabled (the default), the same internal event pipeline is still used, but the ResponseStream is automatically consumed and its events are collected into a WorkflowRunResult that is returned as an awaitable.

Event Types

The WorkflowEvent class uses a type discriminator (a Literal union called WorkflowEventType) to distinguish the following categories:

Category	Event Type	Description
Lifecycle	started	Workflow run has begun.
Lifecycle	status	Workflow state changed (access via event.state). States include STARTED, IN_PROGRESS, IDLE, FAILED, and others.
Lifecycle	failed	Workflow terminated with an error (access via event.details).
Data	output	An executor yielded a final output via ctx.yield_output() (access via event.executor_id, event.data).
Data	data	An executor emitted intermediate data during execution (access via event.executor_id, event.data).
Request	request_info	An executor requests external information, enabling human-in-the-loop patterns (access via event.request_id, event.source_executor_id).
Diagnostic	warning	A warning emitted from user code (access via event.data as str).
Diagnostic	error	A non-fatal error from user code (access via event.data as Exception).
Iteration	superstep_started	A Pregel superstep has begun (access via event.iteration).
Iteration	superstep_completed	A Pregel superstep has ended (access via event.iteration).
Executor Lifecycle	executor_invoked	An executor's handler was called (access via event.executor_id).
Executor Lifecycle	executor_completed	An executor's handler completed successfully (access via event.executor_id).
Executor Lifecycle	executor_failed	An executor's handler raised an error (access via event.executor_id, event.details).

Dual Return Type Strategy

The run() method uses Python @overload signatures to provide a type-safe dual return:

• When stream=True (typed as Literal[True]): returns ResponseStream[WorkflowEvent, WorkflowRunResult].
• When stream=False (typed as Literal[False], the default): returns Awaitable[WorkflowRunResult].

Both paths share the same internal _run_core() async generator, ensuring identical event semantics regardless of the caller's consumption mode.

Event Filtering

In non-streaming mode, certain events are filtered before being placed into the WorkflowRunResult:

• started events are always omitted (they serve telemetry purposes only).
• status events are excluded from the main list by default but are accessible via WorkflowRunResult.status_timeline(). Passing include_status_events=True includes them inline.

In streaming mode, all events are yielded to the caller without filtering, giving full visibility into the execution lifecycle.

Theoretical Basis

The streaming execution pattern implements the Observer Pattern at the workflow level. The ResponseStream acts as an observable sequence, and the caller's async for loop acts as the observer. This decouples the workflow engine's internal execution from the consumption strategy chosen by the caller.

The internal execution follows a Pregel-like superstep model where executors run in synchronized iterations. Messages between executors are delivered at superstep boundaries, and only workflow-level events (outputs, data, lifecycle transitions) are surfaced to the stream. This separation between internal message-passing and external event emission ensures that the streaming interface remains clean and predictable, even for complex graph topologies with concurrent executors.

The pattern also supports checkpoint-and-resume semantics. A workflow run can be started from a checkpoint by passing checkpoint_id, and responses to pending request_info events can be supplied via the responses parameter, enabling durable, interruptible workflows.

Usage Examples

Streaming Execution

from agent_framework import Workflow

workflow = Workflow(...)

async for event in workflow.run("hello", stream=True):
    if event.type == "executor_invoked":
        print(f"Executor started: {event.executor_id}")
    elif event.type == "output":
        print(f"Output from {event.executor_id}: {event.data}")
    elif event.type == "executor_completed":
        print(f"Executor finished: {event.executor_id}")

Non-Streaming Execution

from agent_framework import Workflow

workflow = Workflow(...)

result = await workflow.run("hello")
outputs = result.get_outputs()
final_state = result.get_final_state()

Checkpoint Restoration

from agent_framework import Workflow

workflow = Workflow(...)

# Resume from a checkpoint and supply responses to pending requests
result = await workflow.run(
    checkpoint_id="abc-123",
    responses={"req-1": "approved"},
    checkpoint_storage=my_storage,
)

I/O Contract

Inputs

Parameter	Type	Default	Description
message	None	None	Initial message for the start executor. Required for new workflow runs. Mutually exclusive with responses.
stream	bool	False	When True, returns a ResponseStream yielding WorkflowEvent objects in real time. When False, returns an Awaitable[WorkflowRunResult].
responses	None	None	Responses to pending request_info events. Keys are request IDs, values are the response data. Mutually exclusive with message.
checkpoint_id	None	None	ID of a checkpoint to restore from. Can be combined with responses to restore and respond in one call.
checkpoint_storage	None	None	Runtime checkpoint storage backend.
include_status_events	bool	False	Whether to include status events inline in non-streaming results.

Outputs

Condition	Return Type	Description
stream=True	ResponseStream[WorkflowEvent, WorkflowRunResult]	An asynchronous iterator yielding WorkflowEvent objects as the workflow executes. Call get_final_response() after exhaustion to obtain the WorkflowRunResult.
stream=False (default)	Awaitable[WorkflowRunResult]	An awaitable that resolves to a WorkflowRunResult containing all collected events. Use get_outputs() to extract output data, get_final_state() for terminal state, and status_timeline() for the full status event history.

Related Pages

• Implementation:Microsoft_Agent_framework_Workflow_Run
• Heuristic:Microsoft_Agent_framework_Silent_Workflow_Failures

Sources

Type	Name	URL
Repo	Microsoft Agent Framework	https://github.com/microsoft/agent-framework