# Build a workflow

Agent Mission Control developer docs > Microsoft Agent Framework (Python)

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

**Before you start:** A Foundry project endpoint, a chat model deployment name, and an az login session.

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`
```text
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](https://www.agentmissioncontrol.ai/developers/af-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`
```python
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`
```python
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`
```python
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](https://www.agentmissioncontrol.ai/developers/af-amc-hardening/).

## 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](https://github.com/microsoft/agent-framework/tree/main/python/samples/03-workflows)) 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](https://www.agentmissioncontrol.ai/developers/af-orchestration-and-checkpoints/).

## Related

- [Orchestrations, human input and checkpoints](https://www.agentmissioncontrol.ai/developers/af-orchestration-and-checkpoints/)
- [Harden an agent or workflow for AMC](https://www.agentmissioncontrol.ai/developers/af-amc-hardening/)
- [Host and deploy on Foundry](https://www.agentmissioncontrol.ai/developers/af-host-and-deploy/)

---
Source: https://www.agentmissioncontrol.ai/developers/af-build-a-workflow/ · Last updated 2026-10-21 · Full index: https://www.agentmissioncontrol.ai/llms.txt