# Orchestrations, human input and checkpoints

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

> Use the high-level orchestration builders, expose a workflow as an agent, pause for a person with request_info, and checkpoint so a run can resume.

**Before you start:** A working workflow; see Build a workflow.

## Orchestration builders

For common multi-agent shapes, use a builder from `agent_framework.orchestrations` instead of wiring edges. Each takes `participants=[...]` (agents and executors) and returns a `Workflow` from `.build()`, which you run like any workflow.

```python
from agent_framework.orchestrations import (
    ConcurrentBuilder, GroupChatBuilder, HandoffBuilder, MagenticBuilder, SequentialBuilder,
)

workflow = SequentialBuilder(participants=[writer, reviewer], output_from="all").build()
result = await workflow.run("Draft a launch slogan.")
```

| Builder | Shape | Default output |
| --- | --- | --- |
| `SequentialBuilder` | Agents in a chain with a shared conversation. | The last participant. `output_from="all"` surfaces every one. |
| `ConcurrentBuilder` | Fan out to all participants, fan in with an aggregator you can override. | The aggregator. Participants are hidden unless selected. |
| `GroupChatBuilder` | A manager picks the next speaker: `.with_orchestrator(agent=manager_agent)` or a function selector. | See the samples. |
| `HandoffBuilder` | A triage agent routes to specialists. `.with_autonomous_mode()` lets specialists iterate until they hand off. | The participants. |
| `MagenticBuilder` | A Magentic manager plans and drives the agents. Takes `participants={...}`. | See the samples. |

> **Important:** Treat `MagenticBuilder.participants` keys as stable ids. When you resume from a checkpoint, rebuild with the same keys or the run fails fast.

## Expose a workflow as an agent

```python
agent = workflow.as_agent(name="content_pipeline")
response = await agent.run("...")   # an AgentResponse
```

Wrapped, workflow output becomes the normal `AgentResponse.text` and intermediate output becomes `text_reasoning` content. Foundry hosting serves a workflow this way, and a wrapped workflow can be a participant in another workflow.

## Pause for a person

An executor asks with `ctx.request_info(...)`. The workflow goes idle with a pending request (`IDLE_WITH_PENDING_REQUESTS`). When the answer arrives, the executor's `@response_handler` receives it and the graph continues. This start node asks a free-text question:

`workflow.py`
```python
from dataclasses import dataclass

from agent_framework import Executor, Message, WorkflowContext, handler, response_handler


@dataclass
class TopicRequest:
    question: str        # AMC shows this field to the person


@dataclass
class Topic:
    topic: str


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

    async def _ask(self, ctx: WorkflowContext[Topic, Message]) -> None:
        await ctx.request_info(
            request_data=TopicRequest(question="What would you like me to write about?"),
            response_type=str,
        )

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

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

    @response_handler
    async def on_topic(self, original_request: TopicRequest, answer: str,
                       ctx: WorkflowContext[Topic, Message]) -> None:
        await ctx.send_message(Topic(topic=(answer or "").strip()))
```

> **Important:** Don't put `from __future__ import annotations` in a module that uses `@response_handler`. Stringized annotations break its signature check, which inspects the raw `WorkflowContext[...]` annotation.

Every orchestration builder also supports `.with_request_info()` to pause for review before continuing. For approval before sensitive tool calls, see the skill's `tool-approval/` samples. How AMC shows the pause and sends the answer back is in [Report steps, tokens and human input](https://www.agentmissioncontrol.ai/developers/af-report-to-amc/).

## Checkpoint and resume

A checkpoint persists workflow state so a run can pause, including for human input, and resume later, even in another process. Locally, pass a checkpoint store to the builder with `checkpoint_storage=...`, or to `workflow.as_agent(checkpoint_storage=...)`. For durable storage the skill points to `CosmosCheckpointStorage` and its samples (`checkpoint/cosmos_workflow_checkpointing.py`); take the exact import and constructor from the sample for your version.

Under Foundry hosting, `ResponsesHostServer` owns checkpointing: it restores a paused workflow's checkpoint when the answer arrives. Two rules follow:

- **Default storage is instance-local.** Without process-loss recovery, the pause and the resume must land on the same warm instance, so run the agent on a single replica.
- **Custom types must be allow-listed on core 1.19 and later.** Every custom type a checkpoint carries (the `request_data` payload and each edge dataclass) must be named as `"<module>:<qualname>"` in a `CheckpointStoreProvider`, or the restore fails. See [Harden an agent or workflow for AMC](https://www.agentmissioncontrol.ai/developers/af-amc-hardening/).

## Related

- [Report steps, tokens and human input](https://www.agentmissioncontrol.ai/developers/af-report-to-amc/)
- [Harden an agent or workflow for AMC](https://www.agentmissioncontrol.ai/developers/af-amc-hardening/)
- [Build a workflow](https://www.agentmissioncontrol.ai/developers/af-build-a-workflow/)

---
Source: https://www.agentmissioncontrol.ai/developers/af-orchestration-and-checkpoints/ · Last updated 2026-10-21 · Full index: https://www.agentmissioncontrol.ai/llms.txt