Workflows for sequential and concurrent orchestration

After this chapter, you will be able to model repeatable multi-step agent systems as Microsoft Agent Framework workflows: explicit executors connected by edges for sequential, concurrent, and conditional paths.

You will use the verified agent-framework==1.9.0 Python API: WorkflowBuilder, Executor handlers, edge builders, and Workflow.run(...).

A workflow is an explicit graph of work. Each node performs a step; each edge says where output should go next. That makes workflows a good fit when the process is known: normalize input, run independent checks, aggregate results, then route by rule.

This is the deterministic side of the pattern map from Chapter 2. Sequential orchestration chains stages when the next step depends on the previous one, while concurrent orchestration fans out independent work and fans results back into an aggregator. Microsoft Learn describes sequential orchestration as a pipeline and concurrent orchestration as multiple agents working in parallel with collected results. Sequential source; concurrent source

The boundary from Chapter 1 still matters: agents choose steps dynamically from context and tools; workflows define a predefined flow that can include agents, functions, integrations, and human interactions. Use workflow control when predictability, traceability, and typed handoffs matter more than autonomous path selection. Source

In MAF Python, a workflow is built with WorkflowBuilder and run as a Workflow. You give the builder a start_executor, connect executors or agent nodes with edges, call build(), and then call await workflow.run(message). The inspected API version for this chapter is agent-framework==1.9.0.

Executors are async workflow nodes. For custom code, subclass Executor and decorate handler methods with @handler, or convert a standalone function with @executor. Inside a handler, use WorkflowContext.send_message(...) to send data along graph edges or WorkflowContext.yield_output(...) to emit final workflow output. Agents can participate as nodes through AgentExecutor, which receives AgentExecutorRequest and emits AgentExecutorResponse.

The framework also exposes streaming, checkpoints, and human-in-the-loop workflow features; those operational concerns are covered in Chapter 8. Source

Choose a workflow when the path is mostly known before the model runs. Choose a single agent when one role can solve the task with tools. Choose handoff, group chat, or planner-style orchestration when the right specialist or next step should emerge during the conversation.

• Use workflows for repeatable processes: approvals, enrichment pipelines, validation gates, deterministic routing, and fan-out/fan-in analysis.
• Use sequential edges when order is real: later steps should not start until earlier steps produce validated output.
• Use concurrent edges when independence is real: branches should not depend on each other's intermediate state, and you have a clear aggregator.
• Use agent nodes selectively: wrap an agent with AgentExecutor when model judgment is one step inside a larger deterministic workflow.

Common pitfalls

• Over-engineering graphs: if a function or a single agent call solves the task, a workflow adds unnecessary build, test, and observability overhead. Microsoft recommends using the lowest level of complexity that reliably meets the requirement. Source
• Forgetting the plural fan-in API: in Python agent-framework==1.9.0, the builder methods are add_fan_out_edges and add_fan_in_edges. The fan-in target must handle list[T].
• Missing the switch fallback: add_switch_case_edge_group expects Case entries plus exactly one Default, so define an intentional fallback path.
• Letting branch outputs conflict silently: concurrent orchestration needs an aggregation rule: sort, vote, rank, merge, or reject inconsistent outputs.
• Assuming graph control removes model risk: workflows make routing explicit, but any LLM-backed executor still needs instructions, tool boundaries, validation, telemetry, and human checkpoints for sensitive work.

The verified example below runs without model credentials. It uses trivial executors so you can focus on workflow shape rather than provider setup.

"""Chapter 6: graph workflows with sequential, fan-out/fan-in, and conditional edges.

Runs without model credentials.
"""

import asyncio

from agent_framework import Case, Default, Executor, WorkflowBuilder, WorkflowContext, handler


class UpperCase(Executor):
    def __init__(self) -> None:
        super().__init__(id="upper")

    @handler(input=str, output=str)
    async def handle(self, text: str, ctx: WorkflowContext[str, str]) -> None:
        await ctx.send_message(text.upper())


class AddSuffix(Executor):
    def __init__(self, id: str, suffix: str) -> None:
        self.suffix = suffix
        super().__init__(id=id)

    @handler(input=str, output=str)
    async def handle(self, text: str, ctx: WorkflowContext[str, str]) -> None:
        await ctx.send_message(f"{text}{self.suffix}")


class FinalSuffix(Executor):
    def __init__(self) -> None:
        self.suffix = "!"
        super().__init__(id="exclaim")

    @handler(input=str, workflow_output=str)
    async def handle(self, text: str, ctx: WorkflowContext[None, str]) -> None:
        await ctx.yield_output(f"{text}{self.suffix}")


class JoinResults(Executor):
    def __init__(self) -> None:
        super().__init__(id="join")

    @handler(input=list[str], workflow_output=str)
    async def handle(self, results: list[str], ctx: WorkflowContext[None, str]) -> None:
        await ctx.yield_output(" | ".join(sorted(results)))


class Classifier(Executor):
    def __init__(self) -> None:
        super().__init__(id="classifier")

    @handler(input=str, output=str)
    async def handle(self, text: str, ctx: WorkflowContext[str, str]) -> None:
        await ctx.send_message(text)


class Label(Executor):
    def __init__(self, id: str, label: str) -> None:
        self.label = label
        super().__init__(id=id)

    @handler(input=str, workflow_output=str)
    async def handle(self, text: str, ctx: WorkflowContext[None, str]) -> None:
        await ctx.yield_output(f"{self.label}: {text}")


async def sequential_demo() -> None:
    upper = UpperCase()
    exclaim = FinalSuffix()
    workflow = WorkflowBuilder(start_executor=upper, output_from=[exclaim]).add_edge(upper, exclaim).build()
    result = await workflow.run("hello")
    print("sequential:", result.get_outputs())


async def fan_out_fan_in_demo() -> None:
    upper = UpperCase()
    a = AddSuffix("branch_a", " from A")
    b = AddSuffix("branch_b", " from B")
    join = JoinResults()
    workflow = (
        WorkflowBuilder(start_executor=upper, output_from=[join])
        .add_fan_out_edges(upper, [a, b])
        .add_fan_in_edges([a, b], join)
        .build()
    )
    result = await workflow.run("parallel")
    print("fan_out_fan_in:", result.get_outputs())


async def conditional_demo() -> None:
    classifier = Classifier()
    urgent = Label("urgent", "urgent")
    normal = Label("normal", "normal")
    workflow = (
        WorkflowBuilder(start_executor=classifier, output_from=[urgent, normal])
        .add_switch_case_edge_group(
            classifier,
            [
                Case(lambda text: "!" in text, urgent),
                Default(normal),
            ],
        )
        .build()
    )
    result = await workflow.run("ship it!")
    print("conditional:", result.get_outputs())


async def main() -> None:
    await sequential_demo()
    await fan_out_fan_in_demo()
    await conditional_demo()


if __name__ == "__main__":
    asyncio.run(main())

The sequential demo uppercases text and sends it to a final executor with add_edge. For longer chains, add_chain([step1, step2, step3]) expresses the same pipeline idea more compactly.

The fan-out/fan-in demo sends one message from UpperCase to two suffix branches, then waits at JoinResults. Because fan-in aggregates branch outputs, the join handler declares @handler(input=list[str], workflow_output=str).

The conditional demo uses add_switch_case_edge_group with a Case for urgent text and one Default for the normal path. The workflow is executed with await workflow.run(...) and final values are read with result.get_outputs().

• Workflow graphs make control explicit: executors do work; edges define sequence, concurrency, and conditions.
• Use the real 1.9.0 builder names: add_edge, add_chain, add_fan_out_edges, add_fan_in_edges, and add_switch_case_edge_group.
• Executors are the unit of typed work: @handler methods send messages or yield final output through WorkflowContext.
• Concurrent branches need a join strategy: fan-in produces a list for an aggregator, not a single merged answer by magic.

Next, Chapter 7 moves from explicit workflow graphs to handoff and group chat orchestration, where agent ownership and collaboration can be more dynamic.