How to Implement Human-in-the-Loop Workflows with Approval Gates in the Microsoft Agent Framework

The Microsoft Agent Framework enables human-in-the-loop workflows by combining tool-level approval declarations with Durable-Task orchestrations that pause execution via wait_for_external_event, allowing human reviewers to approve or reject actions via external HTTP endpoints.

The Microsoft Agent Framework provides native primitives for inserting human oversight into automated agent workflows. By leveraging the @tool decorator with approval modes and Durable-Task orchestration patterns, developers can build reliable approval gates that pause execution indefinitely until a human reviewer responds, while maintaining state across retries and timeouts.

Declaring Tools with Approval Requirements

Tools are any callable functions that agents invoke. By adding approval_mode="always_require" to the @tool decorator, the framework automatically blocks execution and emits a HumanApprovalRequest when that tool is called.

In python/samples/03-workflows/tool-approval/sequential_builder_tool_approval.py, the framework demonstrates this pattern:

from agent_framework import tool

@tool(approval_mode="always_require")
def submit_refund(order_id: str, amount: float) -> str:
    """Submit a refund request – must be approved by a human reviewer."""
    # business logic here

    return f"Refund of ${amount:.2f} for order {order_id} submitted."

When the agent invokes submit_refund, the framework creates a HumanApprovalRequest and pauses the orchestration until a HumanApprovalResponse arrives.

Building the Durable-Task Orchestration

The core implementation for approval gates resides in python/samples/04-hosting/durabletask/07_single_agent_orchestration_hitl/worker.py. This file defines a generator-based orchestration that combines agent execution with human review loops.

Structuring the Review Loop

The orchestration function signature accepts an OrchestrationContext and returns a generator yielding Task objects:

from typing import Generator, Any
from durabletask.task import Task, when_any
from datetime import timedelta

def content_generation_hitl_orchestration(
    context: OrchestrationContext, payload_raw: Any
) -> Generator[Task[Any], Any, dict[str, str]]:
    """Human-in-the-loop orchestration with an approval gate."""
    # Validation and agent initialization...

The pattern implements a retry loop that continues until content is approved or the maximum attempts are exhausted.

Modeling the Approval Payload

Before waiting for human input, define the data models for the request and response. The HumanApprovalRequest class carries the tool name, arguments, and description to the reviewer:

from pydantic import BaseModel
from typing import Any

class HumanApprovalRequest(BaseModel):
    """Payload sent to the reviewer."""
    tool_name: str
    args: dict[str, Any]
    description: str

The response model uses HumanApproval (or HumanApprovalResponse in some samples) to capture the decision:

class HumanApproval(BaseModel):
    """Response from the human reviewer."""
    approved: bool
    feedback: str = ""

Implementing Wait and Timeout Logic

Inside the orchestration, the review loop races human input against a timer using when_any. This prevents workflows from hanging indefinitely:

from datetime import timedelta

# Inside the review loop

approval_task = context.wait_for_external_event("HumanApproval")
timeout_task = context.create_timer(
    context.current_utc_datetime + timedelta(seconds=payload.approval_timeout_seconds)
)
winner = yield when_any([approval_task, timeout_task])

if winner == approval_task:
    approval_data = approval_task.get_result()
    approval = HumanApproval.model_validate(approval_data)
    
    if approval.approved:
        yield context.call_activity("publish_content", input=content.model_dump())
        return {"title": content.title, "content": content.content}
    else:
        # Regenerate content with feedback

        rewrite_prompt = f"The content was rejected. Feedback: {approval.feedback}"
        # ... yield writer.run() again

else:
    raise TimeoutError(f"Human approval timed out after {payload.approval_timeout_seconds}s")

If the timeout task wins, the orchestration raises a TimeoutError. After exhausting max_review_attempts, it raises a RuntimeError.

Notifying Reviewers with Activity Functions

Activity functions are regular Python callables (not requiring async) that run inside the Durable-Task worker process. They handle side effects like sending emails or posting to Teams.

In worker.py, the notify_user_for_approval activity logs the pending review:

def notify_user_for_approval(context: ActivityContext, content: dict[str, str]) -> str:
    """Log the draft and tell the reviewer how to approve."""
    model = GeneratedContent.model_validate(content)
    logger.info("Review needed – Title: %s", model.title)
    return "Notification sent"

The publish_content activity handles the final approved output:

def publish_content(context: ActivityContext, content: dict[str, str]) -> str:
    """Placeholder for publishing to a CMS."""
    model = GeneratedContent.model_validate(content)
    logger.info("Publishing approved article – %s", model.title)
    return "Published"

Wiring Together Agents and Orchestrations

The setup_worker function registers agents, activities, and orchestrators with the DurableTaskSchedulerWorker:

def setup_worker(worker: DurableTaskSchedulerWorker) -> DurableAIAgentWorker:
    agent_worker = DurableAIAgentWorker(worker)
    
    # Register the writer agent

    agent_worker.add_agent(create_writer_agent())
    
    # Register activities

    worker.add_activity(notify_user_for_approval)
    worker.add_activity(publish_content)
    
    # Register the orchestrator

    worker.add_orchestrator(content_generation_hitl_orchestration)
    return agent_worker

Receiving Human Decisions via HTTP

The final component is an external endpoint that receives the reviewer’s decision and raises it as an orchestration event. The sample in python/samples/04-hosting/azure_functions/12_workflow_hitl/function_app.py demonstrates this pattern using Azure Functions, but a FastAPI implementation follows the same logic:

from fastapi import FastAPI, Body
from pydantic import BaseModel

app = FastAPI()

class HumanApprovalResponse(BaseModel):
    approved: bool
    feedback: str = ""

@app.post("/approval")
async def approve(
    payload: HumanApprovalResponse,
    instance_id: str = Body(..., embed=True),
):
    """Forward the human decision to the paused orchestration."""
    async with httpx.AsyncClient() as client:
        await client.post(
            f"http://localhost:8080/orchestrations/{instance_id}/raiseEvent",
            json={"eventName": "HumanApproval", "eventData": payload.dict()},
        )
    return {"status": "approval forwarded"}

When the human submits their decision, the orchestration resumes from the yield context.wait_for_external_event("HumanApproval") line with the provided data.

Summary

• Tool-level approval: Use @tool(approval_mode="always_require") to automatically generate approval requests for sensitive operations.
• Durable-Task orchestration: Implement review loops as Python generators using OrchestrationContext to maintain state across human delays.
• Timeout protection: Race wait_for_external_event against create_timer using when_any to prevent indefinite blocking.
• Activity functions: Encapsulate side effects like notifications in regular Python functions registered via worker.add_activity.
• External events: Expose HTTP endpoints that call raiseEvent on the orchestration instance to feed human decisions back into the workflow.

Frequently Asked Questions

How does the framework handle timeouts when waiting for human approval?

The orchestration uses context.create_timer alongside context.wait_for_external_event, then yields when_any([approval_task, timeout_task]). If the timer task completes first, the orchestration raises a TimeoutError with the configured approval_timeout_seconds.

Can I customize what information the human reviewer sees?

Yes. The HumanApprovalRequest model accepts any Pydantic-defined schema. Add fields like risk_score, customer_tier, or custom UI hints to the model, and the framework serializes this data when emitting the approval request.

What is the difference between tool-level and orchestration-level approval?

Tool-level approval uses the @tool decorator with approval_mode="always_require" for automatic gating of specific functions. Orchestration-level approval manually implements the review loop with wait_for_external_event, offering granular control over timeouts, retry logic, and conditional branching based on feedback.

How is state maintained while waiting for human input?

The Durable-Task scheduler persists the orchestration's execution state, including local variables and generator position, to durable storage. When wait_for_external_event yields, the worker process can terminate or scale down, and the workflow resumes exactly where it left off once the external event arrives.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}