Orchestrations, human input and checkpoints

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.

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.")

Expose a workflow as an agent

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:

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()))

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.

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.