Build a workflow

Wire executors and agents into a typed graph with WorkflowBuilder: edges, streaming, output selection, fan-out and fan-in, routing and loops.

Before you startA Foundry project endpoint, a chat model deployment name, and an az login session.View as Markdown

A workflow is a directed graph. Its nodes are executors and its connections are edges. Messages flow along the edges, and terminal nodes yield the workflow's output. Agents inside a workflow are local, ephemeral agents built on FoundryChatClient.

Install

requirements.txt

agent-framework-core==1.19.0
agent-framework-foundry==1.13.1
agent-framework-foundry-hosting==1.0.0b260918
azure-ai-agentserver-responses>=2.2.0b1,<3
azure-ai-projects>=2.3.0
azure-identity>=1.19.0

Set FOUNDRY_PROJECT_ENDPOINT and FOUNDRY_MODEL for local runs. The orchestration builders live in agent-framework-orchestrations. For graph diagrams, the skill installs agent-framework[viz] and GraphViz, but keep that out of anything Foundry builds (see the meta-package warning in the overview).

Executors and the workflow context

An executor is an Executor subclass with @handler methods, or an @executor-decorated async function. Every handler takes the typed input first and a WorkflowContext second.

Context type	The node can
`WorkflowContext[T_Out]`	Send `T_Out` downstream with `await ctx.send_message(...)`.
`WorkflowContext[T_Out, T_W_Out]`	Send downstream and also yield workflow output with `await ctx.yield_output(...)`.
`WorkflowContext`	Neither. Same as `WorkflowContext[Never, Never]`.

workflow.py

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


class UpperCase(Executor):
    def __init__(self, id: str):
        super().__init__(id=id)

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


@executor(id="reverse_text")
async def reverse_text(text: str, ctx: WorkflowContext[Never, str]) -> None:
    await ctx.yield_output(text[::-1])


def create_workflow():
    upper = UpperCase(id="upper")
    return WorkflowBuilder(start_executor=upper).add_edge(upper, reverse_text).build()

To bypass type introspection, for example for a union input, declare types on the decorator: @handler(input=str, output=str). It is all or nothing: if you set any, set every relevant one, and input is required.

Run it

python

events = await create_workflow().run("hello world")
print(events.get_outputs())       # ['DLROW OLLEH']
print(events.get_final_state())   # WorkflowRunState.IDLE

Note

Build a fresh workflow for every run with a helper like create_workflow(). Executor instances hold state, and reusing them leaks it between runs. Agents can be built once and shared.

Agents as nodes

python

import os
from typing import cast

from agent_framework import Agent, AgentResponse, WorkflowBuilder
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    credential=AzureCliCredential(),
)
writer = Agent(client=client, name="writer", instructions="You write and edit content.")
reviewer = Agent(client=client, name="reviewer", instructions="You give concise, actionable feedback.")

workflow = WorkflowBuilder(start_executor=writer).add_edge(writer, reviewer).build()
events = await workflow.run("Create a slogan for an affordable electric bike.")
for out in cast(list[AgentResponse], events.get_outputs()):
    print(f"{out.messages[0].author_name}: {out.text}")

Don't mix FoundryChatClient agents with persistent server-side agents in one workflow. If you need persistent agents, use the single-agent flow.

Typed edges with your own executors

For AMC you will usually wrap each agent in your own executor, so the executor can report a step, catch failures and pass a typed payload on. Use dataclasses for edge payloads.

workflow.py

from dataclasses import dataclass

from agent_framework import Executor, Message, Workflow, WorkflowBuilder, WorkflowContext, handler
from typing_extensions import Never


@dataclass
class Draft:
    prompt: str
    draft: str


def _prompt_from_messages(messages: list[Message]) -> str:
    for message in reversed(messages):
        if message.text:
            return message.text
    return ""


class WriterExecutor(Executor):
    def __init__(self, agent, id: str = "writer") -> None:
        super().__init__(id=id)
        self._agent = agent

    async def _draft(self, prompt: str, ctx: WorkflowContext[Draft, Message]) -> None:
        response = await self._agent.run(prompt)
        await ctx.send_message(Draft(prompt=prompt, draft=response.text))

    @handler
    async def handle_prompt(self, prompt: str, ctx: WorkflowContext[Draft, Message]) -> None:
        await self._draft(prompt, ctx)

    @handler
    async def handle_messages(self, messages: list[Message], ctx: WorkflowContext[Draft, Message]) -> None:
        await self._draft(_prompt_from_messages(messages), ctx)


class PackageExecutor(Executor):
    def __init__(self, id: str = "package") -> None:
        super().__init__(id=id)

    @handler
    async def handle(self, message: Draft, ctx: WorkflowContext[Never, object]) -> None:
        await ctx.yield_output({"prompt": message.prompt, "draft": message.draft})


def build_workflow(writer_agent) -> Workflow:
    writer = WriterExecutor(writer_agent)
    package = PackageExecutor()
    return (
        WorkflowBuilder(
            start_executor=writer,
            output_from=[package],
            intermediate_output_from=[writer],
        )
        .add_edge(writer, package)
        .build()
    )

The start executor has a list[Message] handler. Foundry hosting wraps the workflow with as_agent(), which feeds the start executor the Responses conversation as list[Message] and rejects a start executor that can't take one. The str handler keeps local runs simple.
The context is typed WorkflowContext[Draft, Message] so the executor can also yield Message outputs. That is how it emits the AMC step marker.
Terminal nodes must call ctx.yield_output(...), or the workflow produces no output.

Choose which outputs surface

output_from: executors whose yield_output becomes workflow output (type='output' events, get_outputs()). Special value "all".
intermediate_output_from: executors whose output becomes intermediate output (type='intermediate'). Special value "all_other".
output_from is an allow-list, not a routing rule. Output from an executor in neither list is hidden.
These fail at build time: the same executor in both lists, unknown or duplicate executors, output_from="all_other", and intermediate_output_from="all". Omitting both gives a deprecation warning, so set output_from explicitly.

Important

For AMC, list every non-terminal executor in intermediate_output_from. An executor's step markers are only forwarded when it is designated.

Stream events

python

from agent_framework import AgentResponseUpdate, Message

last_author = None
async for event in workflow.run(Message("user", ["Create a slogan."]), stream=True):
    if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
        if event.data.author_name != last_author:
            last_author = event.data.author_name
            print(f"\n{last_author}: ", end="", flush=True)
        print(event.data.text, end="", flush=True)

To observe node inputs and outputs without changing executors, watch type='executor_invoked' and type='executor_completed' events.

Fan-out and fan-in

Send one message to several branches, then join their results. AgentExecutor wraps an agent as a node, AgentExecutorRequest feeds it, and the result is an AgentExecutorResponse.

workflow.py

from agent_framework import (
    Agent, AgentExecutor, AgentExecutorRequest, AgentExecutorResponse,
    Executor, Message, WorkflowBuilder, WorkflowContext, handler,
)
from typing_extensions import Never


class DispatchToExperts(Executor):
    @handler
    async def dispatch(self, prompt: str, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
        msg = Message("user", [prompt])
        await ctx.send_message(AgentExecutorRequest(messages=[msg], should_respond=True))


class AggregateInsights(Executor):
    @handler
    async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, str]) -> None:
        await ctx.yield_output("\n\n".join(r.agent_response.text for r in results))


researcher = AgentExecutor(agent=Agent(client=client, name="researcher", instructions="..."), id="researcher")
marketer = AgentExecutor(agent=Agent(client=client, name="marketer", instructions="..."), id="marketer")
legal = AgentExecutor(agent=Agent(client=client, name="legal", instructions="..."), id="legal")
dispatcher = DispatchToExperts(id="dispatcher")
aggregator = AggregateInsights(id="aggregator")

workflow = (
    WorkflowBuilder(start_executor=dispatcher)
    .add_fan_out_edges(dispatcher, [researcher, marketer, legal])
    .add_fan_in_edges([researcher, marketer, legal], aggregator)
    .build()
)

The fan-in target's handler receives a list of upstream results, so type its input as list[...] matching what the branches emit. AgentExecutor nodes emit no AMC step markers. To see each branch as its own step in AMC, wrap each agent in your own executor that emits a marker with its own step id; see Harden an agent or workflow for AMC.

Conditional routing, switch-case and loops

These use edge groups instead of plain add_edge. The skill names the shapes and points to the official samples (under python/samples/03-workflows in the Agent Framework repository) for the exact method names, which have changed between releases. Read the sample for your pinned version.

Shape	What it does	Sample
Conditional edge	Route on a classifier or agent decision.	`control-flow/edge_condition.py`
Switch-case edge group	Pick one branch by category.	`control-flow/switch_case_edge_group.py`
Multi-selection edge group	Pick one or many targets at run time.	`control-flow/multi_selection_edge_group.py`
Loop	A judge node routes back until a condition is met.	`control-flow/simple_loop.py`
Cancellation	Cancel a running workflow through asyncio tasks.	`control-flow/workflow_cancellation.py`

Other composition options

Sub-workflows: nest a whole workflow as an executor in a parent (composition/sub_workflow_basics.py).
Functional API: write the workflow as plain async Python with if, loops and asyncio.gather; @step adds per-step checkpointing (functional/ samples).
Shared state: store a value once in workflow state and reuse it across agents (state-management/state_with_agents.py).
High-level builders: for sequential, concurrent, group chat, handoff and Magentic shapes, see Orchestrations, human input and checkpoints.

Last updated 2026-10-21

Build a workflow

Install

Executors and the workflow context

Run it

Agents as nodes

Typed edges with your own executors

Choose which outputs surface

Stream events

Fan-out and fan-in

Conditional routing, switch-case and loops

Other composition options

Related