Human-in-the-Loop in Workflows

Pause workflow execution for user confirmation, input, review, or decisions at the step level or executor level.

Human-in-the-Loop (HITL) in Workflows enables you to pause execution to collect user confirmation, input, or decisions. The workflow state is persisted, allowing you to resume execution after the user responds.

Workflows support HITL at two levels:

• Step-level HITL — pauses before or after a workflow step executes. Configured on the workflow primitive (Step, Loop, Router, Condition, Steps).
• Executor-level HITL — pauses during step execution when an agent or team inside the step calls a tool that requires human approval. Configured on the tool itself (e.g. @tool(requires_confirmation=True)). The pause propagates from the agent/team up to the workflow.

Both levels can be combined on the same step. See Executor HITL and Dual-level HITL for details.

User input is currently supported for Step (to collect parameters) and Router (to select routes). Other primitives (Condition, Loop, Steps) support confirmation only.

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:

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 object:

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 page for all fields and validation rules.

HITL Types

Type	Use Case	Field
Confirmation	Approve/reject before step execution	requires_confirmation=True
User Input	Collect parameters before step execution	requires_user_input=True
Output Review	Review/edit/reject output after step execution	requires_output_review=True
Route Selection	User chooses which path(s) to execute	Router with requires_user_input=True
Iteration Review	Review output between loop iterations	requires_iteration_review=True
Timeout	Auto-resolve if no human response in time	timeout=300
Error Handling	Retry or skip failed steps	on_error=OnError.pause

Step-Level vs Executor-Level

Aspect	Step-Level HITL	Executor-Level HITL
Configured on	The workflow primitive (Step, Loop, etc.)	The tool used by the agent/team
When it pauses	Before or after the step executes	During the step, when the agent calls the tool
Pause kind	pause_kind="step"	pause_kind="executor"
Pause event	StepPaused / RouterPaused / StepOutputReview	StepExecutorPaused
Continue event	StepContinued	StepExecutorContinued
Use case	Gate the whole step (“Run this step?”, “Pick a branch”, “Approve this output”)	Gate a specific tool call (“Approve sending money to X”, “Confirm this DB query”)
Resolved by	req.confirm(), req.set_user_input(...), req.select(...) on the step requirement	Confirming each executor_requirements[].tool_execution

The two can be combined on the same step. For example: requires_confirmation=True on a Step (gate before the agent runs) + @tool(requires_confirmation=True) on the agent’s tool (gate the tool call). See Dual-level HITL.

Supported Primitives

Primitive	Confirmation	User Input	Output Review	Iteration Review	Route Selection	Executor HITL
Step	✓	✓	✓	-	-	✓ (when agent / team is set)
Steps	✓	-	-	-	-	✓ (via inner Step)
Condition	✓	-	-	-	-	✓ (via inner Step)
Loop	✓	-	✓	✓	-	✓ (via inner Step)
Router	✓	-	✓	-	✓	✓ (via selected branch)

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.

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:

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.

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:

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 page for full details.

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.

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.

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:

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.

from agno.workflow import OnError

Step(
    name="api_call",
    executor=unreliable_function,
    on_error=OnError.pause,  # fail | skip(default) | pause
)

Handle in your code:

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
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 page for full details.

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:

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:

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

HumanReview Config

All HITL settings in a single config object

Step HITL

Confirmation, user input, and output review on steps

Executor HITL

Tool-level HITL inside agents/teams that propagates to the workflow

Dual-level HITL

Combine step-level and executor-level HITL on the same step

Output Review

Review, edit, or reject output after step execution

Router HITL

Route selection, confirmation, and output review

Condition HITL

User-controlled branching decisions

Loop HITL

Start confirmation and per-iteration review

Steps HITL

Confirm before executing a pipeline

Timeout

Auto-resolve HITL pauses after a deadline

Error Handling

Retry or skip failed steps

Developer Resources

• HumanReview Config
• Cookbooks
• Step reference
• Router reference
• Condition reference
• Loop reference
• Steps reference