Streaming, checkpointing, and human-in-the-loop

After this chapter, you will be able to make longer-running Microsoft Agent Framework workflows observable with streaming events, resumable with checkpoints, and controllable by people through human-in-the-loop request/response pauses.

Chapters 6 and 7 focused on workflow shape: fixed sequential or concurrent graphs, then handoff and group collaboration when multiple agents need to coordinate. Chapter 2 called the features in this chapter operational overlays: they do not choose the next agent or solve the task themselves, but they make whichever orchestration pattern you choose safer to run in production.

Three concerns show up as soon as a workflow lasts longer than a quick request. First, operators and users need visibility while work is happening; streaming turns an opaque wait into an event stream. Second, useful work should not be lost when a process stops, a deployment restarts, or a person takes time to respond; checkpointing records workflow state at safe boundaries. Third, sensitive or ambiguous steps often need a person or external system to answer before the workflow continues; human-in-the-loop pauses make that dependency explicit.

Microsoft Learn describes MAF workflows as predefined flows that can include agents, human interactions, integrations, events, checkpointing, and streaming or non-streaming execution modes.^{Microsoft Learn} Treat these as production controls around the workflow patterns you already learned: add them when latency, durability, auditability, or approval boundaries matter; skip them when a one-shot call is simpler and enough.

This chapter targets the verified Python package agent-framework==1.9.0. The same workflow can usually run in non-streaming mode for simple callers or streaming mode for progress-aware callers; checkpoint storage and human responses are opt-in run/build parameters.

For stateful executors, pair workflow-level checkpointing with Executor.on_checkpoint_save() and Executor.on_checkpoint_restore(state) so local executor fields are rehydrated along with the graph. Microsoft Learn positions these workflow features as part of building reliable multi-step and multi-agent processes rather than as separate agent patterns.^{Microsoft Learn}

Common pitfalls

• Inventing the wrong HITL API. The verified Python 1.9.0 API does not include RequestInfoExecutor. Call await ctx.request_info(...) from an executor and handle the response with @response_handler.
• Forgetting durable storage. InMemoryCheckpointStorage is useful for tests and demos, but it disappears with the process. Use FileCheckpointStorage or another trusted persistent CheckpointStorage implementation when resume must survive restarts.
• Blocking the workflow on a person. Do not wait synchronously for approval inside the executor. Emit a request_info event, persist or record the request id, return control to the caller, then resume with Workflow.run(responses=...) when the answer arrives.
• Streaming without consuming final results. A streaming run yields events; if the caller needs the final WorkflowRunResult, keep the returned ResponseStream and use its final response API or run a flow that emits the output event you need.
• Loading untrusted file checkpoints. Treat persisted checkpoints as private application state. Do not load checkpoint files from untrusted or tampered locations.

The example below is the verified file at backend/examples/ch08/stream_checkpoint_hitl.py. It runs without model credentials because it uses a custom workflow executor rather than a live LLM call.

"""Chapter 8: streaming events, checkpoint storage, and HITL request/response."""

import asyncio
from dataclasses import dataclass

from agent_framework import (
    Executor,
    InMemoryCheckpointStorage,
    WorkflowBuilder,
    WorkflowContext,
    handler,
    response_handler,
)


@dataclass
class ApprovalRequest:
    item: str


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

    @handler(input=str)
    async def start(self, item: str, ctx: WorkflowContext[ApprovalRequest, str]) -> None:
        await ctx.request_info(ApprovalRequest(item), response_type=bool)

    @response_handler(request=ApprovalRequest, response=bool, workflow_output=str)
    async def on_response(self, request: ApprovalRequest, approved: bool, ctx: WorkflowContext[None, str]) -> None:
        verdict = "approved" if approved else "rejected"
        await ctx.yield_output(f"{request.item}: {verdict}")


async def main() -> None:
    checkpoint_storage = InMemoryCheckpointStorage()
    approval = ApprovalExecutor()
    workflow = WorkflowBuilder(
        name="approval_workflow",
        start_executor=approval,
        checkpoint_storage=checkpoint_storage,
        output_from=[approval],
    ).build()

    responses: dict[str, bool] = {}
    async for event in workflow.run("deploy production", stream=True):
        if event.type == "request_info":
            print(f"stream_request: {event.request_id} -> {event.data}")
            responses[event.request_id] = True
        elif event.type in {"started", "status", "superstep_completed"}:
            print(f"stream_event: {event.type}")

    checkpoints = await checkpoint_storage.list_checkpoints(workflow_name=workflow.name)
    print("checkpoint_count:", len(checkpoints))

    result = await workflow.run(responses=responses)
    print("final_outputs:", result.get_outputs())


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

The first run uses workflow.run("deploy production", stream=True), so the caller can inspect events as they happen. When the executor calls ctx.request_info(ApprovalRequest(item), response_type=bool), the stream emits a WorkflowEvent whose type is request_info. The caller prints the request id and data, then records an approval response in a responses dictionary.

The workflow was built with InMemoryCheckpointStorage, so the pause creates checkpoint state that can be listed with checkpoint_storage.list_checkpoints(workflow_name=workflow.name). In a durable application, use persistent checkpoint storage so this request id and pending state can survive a restart.

Finally, the example resumes by calling await workflow.run(responses=responses). MAF routes the boolean response back to the matching @response_handler, which yields the final output deploy production: approved.

• Streaming exposes progress and incremental output through WorkflowEvent, AgentResponseUpdate, and ResponseStream.
• Checkpointing makes workflow state resumable with CheckpointStorage, including in-memory storage for tests and file storage for trusted local persistence.
• Human-in-the-loop is a request/response pause: call ctx.request_info(), observe the request_info event, and resume with Workflow.run(responses=...).

Next, Chapter 9 turns to middleware and observability: the cross-cutting pipelines and telemetry you add around these workflows to diagnose failures, latency, cost, and behavior.