Skip to main content
Steps support three HITL modes: confirmation (approve/reject before execution), user input (collect parameters before execution), and output review (review/edit/reject after execution). All HITL settings are configured via HumanReview.

​
Confirmation

Pause before executing a step. The user confirms to proceed or rejects to skip/cancel. 
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"),
    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:
        print(f"Step: {req.step_name}")
        print(f"Message: {req.confirmation_message}")
        
        if input("Confirm? (y/n): ").lower() == "y":
            req.confirm()
        else:
            req.reject()
    
    run_output = workflow.continue_run(run_output)

print(run_output.content)

​
Parameters

FieldTypeDescription
requires_confirmationboolPause for user confirmation before execution
confirmation_messagestrMessage shown to the user
on_rejectOnRejectAction when rejected: skip (default), cancel

​
OnReject Options

ValueBehavior
OnReject.skipSkip this step and continue with the next (default)
OnReject.cancelCancel the entire workflow
OnReject.retryRe-execute the step. Used with output review

​
User Input

Collect parameters from the user before step execution. Input values are passed to the step via step_input.additional_data["user_input"]. 
from agno.workflow import Workflow
from agno.workflow.step import Step
from agno.workflow.types import HumanReview, StepInput, StepOutput, UserInputField
from agno.db.sqlite import SqliteDb

def process_with_params(step_input: StepInput) -> StepOutput:
    user_input = step_input.additional_data.get("user_input", {})
    threshold = user_input.get("threshold", 0.5)
    mode = user_input.get("mode", "fast")
    
    return StepOutput(content=f"Processed with threshold={threshold}, mode={mode}")

workflow = Workflow(
    name="configurable_pipeline",
    db=SqliteDb(db_file="workflow.db"),
    steps=[
        Step(name="analyze", agent=analyze_agent),
        Step(
            name="process",
            executor=process_with_params,
            human_review=HumanReview(
                requires_user_input=True,
                user_input_message="Configure processing:",
                user_input_schema=[
                    UserInputField(
                        name="threshold",
                        field_type="float",
                        description="Processing threshold (0.0-1.0)",
                        required=True,
                    ),
                    UserInputField(
                        name="mode",
                        field_type="str",
                        description="Mode: 'fast' or 'accurate'",
                        required=True,
                    ),
                    UserInputField(
                        name="batch_size",
                        field_type="int",
                        description="Records per batch",
                        required=False,
                    ),
                ],
            ),
        ),
        Step(name="report", agent=report_agent),
    ],
)

run_output = workflow.run("Process Q4 data")

if run_output.is_paused:
    for req in run_output.steps_requiring_user_input:
        print(f"Step: {req.step_name}")
        print(f"Message: {req.user_input_message}")
        
        values = {}
        for field in req.user_input_schema:
            marker = "*" if field.required else ""
            prompt = f"{field.name}{marker} ({field.field_type}): "
            value = input(prompt)
            
            if value:
                if field.field_type == "int":
                    values[field.name] = int(value)
                elif field.field_type == "float":
                    values[field.name] = float(value)
                elif field.field_type == "bool":
                    values[field.name] = value.lower() in ("true", "yes", "1")
                else:
                    values[field.name] = value
        
        req.set_user_input(**values)
    
    run_output = workflow.continue_run(run_output)

print(run_output.content)

​
Parameters

FieldTypeDescription
requires_user_inputboolPause to collect user input before execution
user_input_messagestrMessage shown to the user
user_input_schemaList[UserInputField]Schema defining expected input fields

​
UserInputField

FieldTypeDescription
namestrField name (key in user input dict)
field_typestrType: str, int, float, bool
descriptionstrDescription shown to user
requiredboolWhether field is required (default: True)
allowed_valuesList[Any]Optional list of valid values

​
Accessing User Input

User input is available in the step function via step_input.additional_data["user_input"]: 
def my_step(step_input: StepInput) -> StepOutput:
    user_input = step_input.additional_data.get("user_input", {})
    
    threshold = user_input.get("threshold")
    mode = user_input.get("mode")
    
    return StepOutput(content=f"Done with {threshold}, {mode}")
For agent-based steps, user input is automatically appended to the message.

​
Output Review

Pause after a step executes so the user can review the output before the workflow continues. If the user rejects, the step re-executes (up to max_retries). See the dedicated Output Review page for full details including editing output and reject with feedback. 
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="content_pipeline",
    db=SqliteDb(db_file="workflow.db"),
    steps=[
        Step(name="research", agent=research_agent),
        Step(
            name="draft",
            agent=draft_agent,
            human_review=HumanReview(
                requires_output_review=True,
                output_review_message="Review the draft. Approve to continue or reject to regenerate.",
                on_reject=OnReject.retry,
                max_retries=3,
            ),
        ),
        Step(name="publish", agent=publish_agent),
    ],
)

run_output = workflow.run("Write a blog post about AI agents")

while run_output.is_paused:
    for req in run_output.steps_requiring_output_review:
        print(f"Output: {req.step_output.content}")
        print(f"Message: {req.output_review_message}")
        
        if input("Approve? (y/n): ").lower() == "y":
            req.confirm()
        else:
            feedback = input("Feedback (optional): ").strip()
            if feedback:
                req.reject(feedback=feedback)
            else:
                req.reject()
    
    run_output = workflow.continue_run(run_output)

print(run_output.content)

​
Parameters

FieldTypeDescription
requires_output_reviewboolPause after execution for user review
output_review_messagestrMessage shown during review
on_rejectOnRejectAction when rejected. Use OnReject.retry to re-execute
max_retriesintMax re-executions on rejection (default: 3)

​
Timeout

Set a deadline for the user to respond. If it expires, on_timeout fires automatically. See the dedicated Timeout page for full details. 
from agno.workflow import OnTimeout

Step(
    name="review_report",
    agent=report_agent,
    human_review=HumanReview(
        requires_output_review=True,
        output_review_message="Review the report.",
        on_reject=OnReject.retry,
        max_retries=2,
        timeout=120,  # 2 minutes
        on_timeout=OnTimeout.approve,
    ),
)
OnTimeout ValueBehavior
OnTimeout.approveAuto-approve and continue
OnTimeout.rejectAuto-reject (triggers retry if on_reject=OnReject.retry)
OnTimeout.cancelCancel the workflow

​
The @pause Decorator

Use the @pause decorator to configure HITL directly on functions: 
from agno.workflow.decorators import pause
from agno.workflow.types import StepInput, StepOutput, UserInputField

@pause(
    requires_confirmation=True,
    confirmation_message="Execute this step?",
)
def step_with_confirmation(step_input: StepInput) -> StepOutput:
    return StepOutput(content="Executed after confirmation")

@pause(
    requires_user_input=True,
    user_input_message="Enter parameters:",
    user_input_schema=[
        UserInputField(name="value", field_type="str", required=True),
    ],
)
def step_with_input(step_input: StepInput) -> StepOutput:
    value = step_input.additional_data["user_input"]["value"]
    return StepOutput(content=f"Received: {value}")

# Decorator config is auto-detected when used in a Step
workflow = Workflow(
    steps=[
        Step(name="confirm_step", executor=step_with_confirmation),
        Step(name="input_step", executor=step_with_input),
    ],
    ...
)

​
Streaming

Handle HITL in streaming workflows: 
from agno.run.workflow import StepPausedEvent

for event in workflow.run("input", stream=True, stream_events=True):
    if isinstance(event, StepPausedEvent):
        print(f"Paused at: {event.step_name}")

session = workflow.get_session()
run_output = session.runs[-1]

while run_output.is_paused:
    for req in run_output.steps_requiring_confirmation:
        req.confirm()
    
    for event in workflow.continue_run(run_output, stream=True, stream_events=True):
        pass
    
    session = workflow.get_session()
    run_output = session.runs[-1]

​
Developer Resources