# How to Implement Human-in-the-Loop Workflows with Approval Gates in the Microsoft Agent Framework > Implement human-in-the-loop workflows with approval gates in Microsoft Agent Framework. Use Durable-Task and external events for human reviewer approval of actions. - Repository: [Microsoft/agent-framework](https://github.com/microsoft/agent-framework) - Tags: how-to-guide - Published: 2026-04-05 --- **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`](https://github.com/microsoft/agent-framework/blob/main/python/samples/03-workflows/tool-approval/sequential_builder_tool_approval.py), the framework demonstrates this pattern: ```python 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`](https://github.com/microsoft/agent-framework/blob/main/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: ```python 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: ```python 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: ```python 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: ```python 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`](https://github.com/microsoft/agent-framework/blob/main/worker.py), the `notify_user_for_approval` activity logs the pending review: ```python 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: ```python 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`: ```python 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`](https://github.com/microsoft/agent-framework/blob/main/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: ```python 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.