Workflow Mechanics — Microsoft Agent Framework in Python

Post 8 of 12 in Learning the Microsoft Agent Framework — Python.

Up to now every lesson hardened one agent. A workflow is the next layer up: it orchestrates many executors — plain functions or whole agents — as a directed graph. A message enters, flows along edges, and each executor transforms or routes it. The best way I found to learn it was to start with no model at all — pure functions — so the orchestration mechanics stand on their own with zero credentials.

Executors are nodes, edges are data flow

An executor is either a class Executor with an @handler method, or an @executor async function. Its handler always takes the message plus a typed WorkflowContext:

from agent_framework import Executor, WorkflowContext, handler
from typing_extensions import Never

class Intake(Executor):
    @handler
    async def run(self, n: int, ctx: WorkflowContext[int]) -> None:
        await ctx.send_message(n)          # forwards n along its outgoing edges

class HandleEven(Executor):
    @handler
    async def run(self, n: int, ctx: WorkflowContext[Never, str]) -> None:
        await ctx.yield_output(f"{n} is even")   # terminal: yields, sends nothing

The type parameters carry intent. WorkflowContext[int] means “this node will send_message(int)”; WorkflowContext[Never, str] means “terminal — it only yield_output(str).” That’s not decoration: the builder uses those types to validate the graph before it ever runs.

The builder wires the graph

WorkflowBuilder describes the shape, not the execution. The start node is a constructor kwarg — there is no set_start_executor() in this build:

from agent_framework import Case, Default, WorkflowBuilder

workflow = (
    WorkflowBuilder(start_executor=intake)
    .add_switch_case_edge_group(intake, [
        Case(condition=lambda x: x % 2 == 0, target=even),
        Default(target=odd),
    ])
    .build()
)

add_edge(src, dst) is the plain directed edge. add_switch_case_edge_group gives you data-dependent routing — cases are evaluated in order, first match wins, else Default, and only one branch runs. There’s also add_fan_out_edges → add_fan_in_edges for parallelism: fan-in hands the target a list of all sources’ messages, once, after every branch completes.

flowchart LR
    I[Intake] --> S{switch-case}
    S -->|x % 2 == 0| E[HandleEven]
    S -->|Default| O[HandleOdd]
    E --> OUT([output])
    O --> OUT

Running: block, or stream typed events

await workflow.run(x) returns a WorkflowRunResult; .get_outputs() is the list of yielded outputs. But the more interesting mode is streaming:

async for event in workflow.run(x, stream=True):
    if event.type == "executor_completed":
        print("done:", event.executor_id)
    elif event.type == "output":
        print("output:", event.data)

What tripped me up: there are no per-kind event classes and no run_stream(). You pass stream=True and branch on event.type — "executor_completed", "output", and friends. Each executor reports as it finishes, so you watch the graph light up node by node rather than waiting on one blocking answer.

Functions first, agents later

Everything above is model-free — 01_control_flow.py routes evens and odds with zero credentials. That’s deliberate. Once the wiring is second nature, dropping an agent into the graph is anticlimactic: WorkflowBuilder auto-wraps any agent in an AgentExecutor, so a FoundryChatClient-backed writer sits on exactly the same add_edge chain as the pure functions. Same edges, same events — the only new thing is that some nodes now cost a model call. That’s the whole point of learning the mechanics with functions first: the graph doesn’t care what a node is, only what it sends.

Next: Workflows with Agents — Microsoft Agent Framework in Python

Key takeaways

• An executor is a node — either a class Executor with an @handler or an @executor async function — and edges carry the messages it emits with ctx.send_message; a terminal node uses ctx.yield_output and sends nothing onward.
• The WorkflowContext[...] type parameters are load-bearing, not decoration: WorkflowContext[int] declares “sends an int,” WorkflowContext[Never, str] declares “terminal, yields a str,” and the builder validates the graph against them before the first run.
• WorkflowBuilder takes the start node as the start_executor= constructor kwarg (there is no set_start_executor() here); add_switch_case_edge_group gives first-match-wins routing, and add_fan_out_edges → add_fan_in_edges hands the fan-in target one list of all branch messages after every branch completes.
• Streaming has no per-kind event classes and no run_stream() — you pass stream=True to workflow.run and branch on event.type ("executor_completed", "output"), watching the graph light up node by node.
• Learn the mechanics with pure functions first; dropping in an agent is anticlimactic because WorkflowBuilder auto-wraps it in an AgentExecutor that sits on the same add_edge chain — the graph cares only what a node sends, not what it is.

Further reading

• Observability, Safety & Providers — Microsoft Agent Framework in Python
• Workflows with Agents — Microsoft Agent Framework in Python
• Workflow Mechanics — Microsoft Agent Framework Go — the same graph model in the Go SDK