Implementation:Microsoft Agent framework Workflow Run

Property	Value
Implementation Name	Workflow Run
SDK	Microsoft Agent Framework
Repository	Microsoft Agent Framework
Source File	`python/packages/core/agent_framework/_workflows/_workflow.py`
Line Range	L449-535
Import	`from agent_framework import Workflow`
Type	Instance method (sync entry, async internals)
Domains	Agent_Architecture, Workflow_Engine

Overview

The Workflow.run() method is the primary entry point for executing a workflow in the Microsoft Agent Framework. It accepts an initial message (or checkpoint/response parameters for resumption), orchestrates the graph-based execution of connected executors, and returns either a streaming ResponseStream[WorkflowEvent, WorkflowRunResult] for real-time event monitoring or an Awaitable[WorkflowRunResult] for batch-style consumption. The method validates parameter combinations, constructs the internal ResponseStream, and delegates to the shared _run_core() async generator for actual execution.

Code Reference

Source Location

Property	Value
File	`python/packages/core/agent_framework/_workflows/_workflow.py`
Class	`Workflow`
Method	`run`
Lines	449-535

Signature

def run(
    self,
    message: Any | None = None,
    *,
    stream: bool = False,
    responses: dict[str, Any] | None = None,
    checkpoint_id: str | None = None,
    checkpoint_storage: CheckpointStorage | None = None,
    include_status_events: bool = False,
    **kwargs: Any,
) -> ResponseStream[WorkflowEvent, WorkflowRunResult] | Awaitable[WorkflowRunResult]:

The method also provides two @overload signatures for precise static typing:

# Overload 1: stream=True returns a ResponseStream
@overload
def run(
    self,
    message: Any | None = None,
    *,
    stream: Literal[True],
    responses: dict[str, Any] | None = None,
    checkpoint_id: str | None = None,
    checkpoint_storage: CheckpointStorage | None = None,
    **kwargs: Any,
) -> ResponseStream[WorkflowEvent, WorkflowRunResult]: ...

# Overload 2: stream=False (default) returns an Awaitable
@overload
def run(
    self,
    message: Any | None = None,
    *,
    stream: Literal[False] = ...,
    responses: dict[str, Any] | None = None,
    checkpoint_id: str | None = None,
    checkpoint_storage: CheckpointStorage | None = None,
    include_status_events: bool = False,
    **kwargs: Any,
) -> Awaitable[WorkflowRunResult]: ...

Import Statement

from agent_framework import Workflow

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` that yields `WorkflowEvent` objects incrementally. When `False` (default), returns an `Awaitable[WorkflowRunResult]` that resolves after all events are collected.
`responses`	None	`None`	Responses to pending `request_info` events from a prior run. Keys are request IDs, values are the corresponding response data. Mutually exclusive with `message`. Can be combined with `checkpoint_id` to restore a checkpoint and send responses in a single call.
`checkpoint_id`	None	`None`	ID of a previously saved checkpoint to restore from. Can be used alone (resume from checkpoint) or with `responses` (restore then send responses). Cannot be combined with `message`.
`checkpoint_storage`	None	`None`	Runtime checkpoint storage backend. When provided, enables checkpoint persistence during the run and is cleaned up after the stream is consumed.
`include_status_events`	`bool`	`False`	Whether to include `status` events inline in the `WorkflowRunResult` for non-streaming runs. Has no effect in streaming mode (all events are yielded).
`**kwargs`	`Any`	--	Additional keyword arguments passed through to agent/executor invocations during the workflow run.

Output

Condition	Return Type	Description
`stream=True`	`ResponseStream[WorkflowEvent, WorkflowRunResult]`	An asynchronous iterator yielding `WorkflowEvent` objects as the workflow progresses through supersteps. Each event carries a `type` discriminator and associated payload fields. After iteration completes, call `get_final_response()` to retrieve the `WorkflowRunResult`.
`stream=False` (default)	`Awaitable[WorkflowRunResult]`	An awaitable that resolves to a `WorkflowRunResult` (a list subclass of `WorkflowEvent`). Key methods on the result: `get_outputs()` - Returns `list[Any]` of all output data from `output` events. `get_request_info_events()` - Returns `list[WorkflowEvent]` of pending external requests. `get_final_state()` - Returns the terminal `WorkflowRunState` (e.g., `IDLE`, `FAILED`). `status_timeline()` - Returns the ordered list of all `status` events.

Exceptions

Exception	Condition
`ValueError`	Raised when parameter combinations are invalid (e.g., both `message` and `responses` provided, or `message` combined with `checkpoint_id`).
`RuntimeError`	Raised by `_ensure_not_running()` if the workflow is already executing (concurrent runs on the same instance are not permitted).

Execution Flow

The run() method implements the following sequence:

Parameter Validation: _validate_run_params() checks that message and responses are mutually exclusive and that checkpoint_id is not combined with message.
Concurrency Guard: _ensure_not_running() verifies that no other run is active on this workflow instance.
ResponseStream Construction: A ResponseStream[WorkflowEvent, WorkflowRunResult] is created, wrapping the _run_core() async generator. It is configured with:
- A finalizer (_finalize_events) that converts the collected event list into a WorkflowRunResult, filtering out started events and optionally including status events.
- A cleanup hook (_run_cleanup) that clears runtime checkpoint storage and resets the running flag.
Mode Selection: If stream=True, the ResponseStream is returned directly for incremental consumption. If stream=False, response_stream.get_final_response() is called, which returns an awaitable that consumes all events internally and returns the finalized WorkflowRunResult.
Core Execution (inside _run_core()):
- Checkpoint storage is attached to the runner context if provided.
- The execution mode is resolved (fresh run, checkpoint restoration, or response delivery).
- The workflow graph is executed via _run_workflow_with_tracing(), which runs supersteps until the graph reaches an idle state.
- Each WorkflowEvent is yielded as it is produced, with output events filtered through _should_yield_output_event().
Cleanup: After the stream is fully consumed (or on error), the cleanup hook clears checkpoint storage and resets the running flag, allowing the workflow instance to be reused.

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"Started: {event.executor_id}")
    elif event.type == "output":
        print(f"Output: {event.data}")
    elif event.type == "executor_completed":
        print(f"Completed: {event.executor_id}")

Non-Streaming Execution

from agent_framework import Workflow

workflow = Workflow(...)

result = await workflow.run("hello")
outputs = result.get_outputs()
print(f"Final state: {result.get_final_state()}")

Checkpoint Restoration with Responses

from agent_framework import Workflow

workflow = Workflow(...)

# First run encounters a request_info event and goes idle
result = await workflow.run("process this", include_status_events=True)
requests = result.get_request_info_events()

# Resume from checkpoint with a response to the pending request
result2 = await workflow.run(
    checkpoint_id="saved-checkpoint-id",
    responses={requests[0].request_id: "approved"},
    checkpoint_storage=my_storage,
)
final_outputs = result2.get_outputs()

Streaming with Extra Kwargs

from agent_framework import Workflow

workflow = Workflow(...)

# Pass additional arguments through to executor invocations
async for event in workflow.run("analyze data", stream=True, temperature=0.7):
    if event.type == "data":
        print(event.data)

Internal Dependencies

Component	Role
`_run_core()`	Async generator that drives the actual workflow execution, yielding `WorkflowEvent` objects from the superstep loop.
`_validate_run_params()`	Validates mutual exclusivity of `message`/`responses` and checks `checkpoint_id` constraints.
`_ensure_not_running()`	Guards against concurrent runs on the same workflow instance.
`_finalize_events()`	Converts raw event list into `WorkflowRunResult`, filtering `started` and optionally `status` events.
`_run_cleanup()`	Clears runtime checkpoint storage and resets the running flag after stream consumption.
`_resolve_execution_mode()`	Determines whether the run is a fresh start, checkpoint restoration, or response delivery.
`_run_workflow_with_tracing()`	Wraps the superstep execution loop with OpenTelemetry tracing spans.
`ResponseStream`	Generic async stream wrapper from `agent_framework._types` that supports both iteration and `get_final_response()`.

Related Pages

Sources

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

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment