> ## Documentation Index > Fetch the complete documentation index at: https://agno-v2-docs-align-with-readme.mintlify.site/llms.txt > Use this file to discover all available pages before exploring further. # Human-in-the-Loop in Workflows > Pause workflow execution for user confirmation, input, review, or decisions at the step level. Human-in-the-Loop (HITL) in Workflows enables you to pause execution at any step to collect user confirmation, input, or decisions. The workflow state is persisted, allowing you to resume execution after the user responds. User input is currently supported for `Step` (to collect parameters) and `Router` (to select routes). Other primitives (`Condition`, `Loop`, `Steps`) support confirmation only. Agent tool-level HITL (e.g., `@tool(requires_confirmation=True)`) is **not** propagated to the workflow. If an agent inside a step has tool-level HITL, the workflow will continue but the paused tool may not execute. Use workflow-level HITL instead. ```python theme={null} from agno.workflow import Workflow, OnReject from agno.workflow.step import Step from agno.workflow.types import HumanReview from agno.db.sqlite import SqliteDb workflow = Workflow( name="data_pipeline", db=SqliteDb(db_file="workflow.db"), # Required for HITL steps=[ Step(name="fetch_data", agent=fetch_agent), Step( name="process_data", agent=process_agent, human_review=HumanReview( requires_confirmation=True, confirmation_message="Process sensitive data?", on_reject=OnReject.skip, ), ), Step(name="save_results", agent=save_agent), ], ) run_output = workflow.run("Process user data") if run_output.is_paused: for req in run_output.steps_requiring_confirmation: req.confirm() # or req.reject() run_output = workflow.continue_run(run_output) ``` ## Requirements HITL workflows require a database to persist state between pauses: ```python theme={null} from agno.db.sqlite import SqliteDb from agno.db.postgres import PostgresDb # SQLite for development workflow = Workflow(db=SqliteDb(db_file="workflow.db"), ...) # PostgreSQL for production workflow = Workflow(db=PostgresDb(db_url="postgresql://..."), ...) ``` ## HumanReview Config All HITL settings are grouped into a single [`HumanReview`](/workflows/hitl/human-review) object: ```python theme={null} from agno.workflow.types import HumanReview Step( name="review_step", agent=review_agent, human_review=HumanReview( requires_output_review=True, output_review_message="Check the output.", on_reject=OnReject.retry, max_retries=3, timeout=300, ), ) ``` Flat parameters (`requires_confirmation=True`, `confirmation_message="..."`) still work for backward compatibility. See the [HumanReview Config](/workflows/hitl/human-review) page for all fields and validation rules. ## HITL Types | Type | Use Case | Field | | --------------------------------------------------------- | ---------------------------------------------- | ------------------------------------------ | | [Confirmation](#confirmation) | Approve/reject before step execution | `requires_confirmation=True` | | [User Input](#user-input) | Collect parameters before step execution | `requires_user_input=True` | | [Output Review](/workflows/hitl/output-review) | Review/edit/reject output after step execution | `requires_output_review=True` | | [Route Selection](#route-selection) | User chooses which path(s) to execute | **Router** with `requires_user_input=True` | | [Iteration Review](/workflows/hitl/loop#iteration-review) | Review output between loop iterations | `requires_iteration_review=True` | | [Timeout](/workflows/hitl/timeout) | Auto-resolve if no human response in time | `timeout=300` | | [Error Handling](#error-handling) | Retry or skip failed steps | `on_error=OnError.pause` | ## Supported Primitives | Primitive | Confirmation | User Input | Output Review | Iteration Review | Route Selection | | -------------------------------------- | ------------ | ---------- | ------------------------------------------ | ---------------- | --------------- | | [Step](/workflows/hitl/step) | ✓ | ✓ | [✓](/workflows/hitl/output-review) | - | - | | [Steps](/workflows/hitl/steps) | ✓ | - | - | - | - | | [Condition](/workflows/hitl/condition) | ✓ | - | - | - | - | | [Loop](/workflows/hitl/loop) | ✓ | - | [✓](/workflows/hitl/loop#iteration-review) | ✓ | - | | [Router](/workflows/hitl/router) | ✓ | - | [✓](/workflows/hitl/router#output-review) | - | ✓ | ## Run Output Properties When a workflow pauses, check these properties on `WorkflowRunOutput`: | Property | Description | | ------------------------------- | --------------------------------------------- | | `is_paused` | `True` if workflow is waiting for user action | | `steps_requiring_confirmation` | Steps needing confirm/reject | | `steps_requiring_user_input` | Steps needing user input values | | `steps_requiring_output_review` | Steps needing output review after execution | | `steps_requiring_route` | Routers needing route selection | | `steps_with_errors` | Steps that failed with `on_error="pause"` | ## Confirmation Pause before executing a step. User confirms to proceed or rejects to skip/cancel. ```python theme={null} Step( name="delete_records", agent=delete_agent, human_review=HumanReview( requires_confirmation=True, confirmation_message="Delete 1000 records?", on_reject=OnReject.skip, ), ) ``` Handle in your code: ```python theme={null} for req in run_output.steps_requiring_confirmation: print(req.confirmation_message) if user_approves(): req.confirm() else: req.reject() ``` ## User Input Collect parameters from the user before step execution. ```python theme={null} from agno.workflow.types import HumanReview, UserInputField Step( name="generate_report", agent=report_agent, human_review=HumanReview( requires_user_input=True, user_input_message="Configure report settings:", user_input_schema=[ UserInputField(name="format", field_type="str", required=True), UserInputField(name="include_charts", field_type="bool", required=False), ], ), ) ``` Handle in your code: ```python theme={null} for req in run_output.steps_requiring_user_input: print(req.user_input_message) for field in req.user_input_schema: value = get_user_value(field.name, field.field_type) req.set_user_input(**{field.name: value}) ``` User input is available in the custom function step via `step_input.additional_data["user_input"]`. ## Output Review Pause after a step executes so the user can review the output. Supported on Step and Router. See the dedicated [Output Review](/workflows/hitl/output-review) page for full details. ```python theme={null} Step( name="draft_email", agent=draft_agent, human_review=HumanReview( requires_output_review=True, output_review_message="Review the draft email.", on_reject=OnReject.retry, max_retries=3, ), ) ``` When rejected with `on_reject=OnReject.retry`, the step re-executes up to `max_retries` times. Pair with `reject(feedback=...)` to send feedback to the agent. ## Iteration Review Pause after each loop iteration for review. Supported on Loop only. ```python theme={null} from agno.workflow.loop import Loop Loop( name="refinement_loop", steps=[Step(name="refine", executor=refine_fn)], max_iterations=5, human_review=HumanReview( requires_iteration_review=True, iteration_review_message="Review this iteration.", on_reject=OnReject.retry, ), ) ``` ## Route Selection Let users choose which path(s) a Router executes. ```python theme={null} from agno.workflow.router import Router Router( name="analysis_router", choices=[ Step(name="quick_analysis", ...), Step(name="deep_analysis", ...), ], human_review=HumanReview( requires_user_input=True, user_input_message="Select analysis type:", ), allow_multiple_selections=False, ) ``` Handle in your code: ```python theme={null} for req in run_output.steps_requiring_route: print(req.available_choices) # ["quick_analysis", "deep_analysis"] req.select("deep_analysis") # or req.select_multiple(["quick_analysis", "deep_analysis"]) ``` ## Error Handling Pause when a step fails, letting the user retry or skip. `This is only at the Step level.` ```python theme={null} from agno.workflow import OnError Step( name="api_call", executor=unreliable_function, on_error=OnError.pause, # fail | skip(default) | pause ) ``` Handle in your code: ```python theme={null} for req in run_output.steps_with_errors: print(f"Error: {req.error_message}") if should_retry(): req.retry() else: req.skip() ``` ## OnReject Behavior The `on_reject` parameter controls what happens when a user rejects a step: | Value | Behavior | | ---------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------- | | `OnReject.skip` | Skip the step and continue with the next (default for most primitives) | | `OnReject.cancel` | Cancel the entire workflow | | `OnReject.retry` | Re-execute the step. Pair with `reject(feedback=...)` to send feedback to the agent. See [Output Review](/workflows/hitl/output-review#reject-with-retry) | | `OnReject.else_branch` | For Condition only: execute `else_steps` (default for Condition) | ## Timeout Set a timeout for HITL pauses. If the user does not respond in time, `on_timeout` determines the action. See the dedicated [Timeout](/workflows/hitl/timeout) page for full details. ```python theme={null} from agno.workflow import OnTimeout Step( name="approve_deploy", agent=deploy_agent, human_review=HumanReview( requires_confirmation=True, confirmation_message="Approve deployment?", timeout=300, # 5 minutes on_timeout=OnTimeout.approve, ), ) ``` | `OnTimeout` Value | Behavior | | ------------------- | ------------------------- | | `OnTimeout.approve` | Auto-approve and continue | | `OnTimeout.reject` | Auto-reject | | `OnTimeout.cancel` | Cancel the workflow | ## Streaming HITL works with streaming workflows. Check for pauses in the event stream: ```python theme={null} for event in workflow.run("input", stream=True, stream_events=True): if isinstance(event, StepPausedEvent): # Handle pause pass session = workflow.get_session() run_output = session.runs[-1] if run_output.is_paused: # Handle requirements workflow.continue_run(run_output, stream=True, stream_events=True) ``` ## The @pause Decorator Mark custom function steps with HITL configuration using the `@pause` decorator: ```python theme={null} from agno.workflow.decorators import pause from agno.workflow.types import UserInputField @pause( requires_user_input=True, user_input_message="Enter parameters:", user_input_schema=[ UserInputField(name="threshold", field_type="float", required=True), ], ) def process_data(step_input: StepInput) -> StepOutput: threshold = step_input.additional_data["user_input"]["threshold"] return StepOutput(content=f"Processed with threshold {threshold}") # The decorator config is auto-detected when used in a custom function step Step(name="process", executor=process_data) ``` ## Guides All HITL settings in a single config object Confirmation, user input, and output review on steps Review, edit, or reject output after step execution Route selection, confirmation, and output review User-controlled branching decisions Start confirmation and per-iteration review Confirm before executing a pipeline Auto-resolve HITL pauses after a deadline Retry or skip failed steps ## Developer Resources * [HumanReview Config](/workflows/hitl/human-review) * [Cookbooks](https://github.com/agno-agi/agno/tree/main/cookbook/04_workflows/08_human_in_the_loop) * [Step reference](/reference/workflows/step) * [Router reference](/reference/workflows/router-steps) * [Condition reference](/reference/workflows/conditional-steps) * [Loop reference](/reference/workflows/loop-steps) * [Steps reference](/reference/workflows/steps-step)