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 (Step.requires_confirmation) instead.
from agno.workflow import Workflow, OnReject
from agno.workflow.step import Step
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,
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://..."), ...)
HITL Types
| Type | Use Case | Flag |
|---|
| Confirmation | Approve/reject before step execution | requires_confirmation=True |
| User Input | Collect parameters before step execution | requires_user_input=True |
| Route Selection | User chooses which path(s) to execute | Router with requires_user_input=True |
| Error Handling | Retry or skip failed steps | on_error=OnError.pause |
Supported Primitives
HITL is supported on all workflow primitives:
| Primitive | Confirmation | User Input | Route Selection |
|---|
| Step | ✓ | ✓ | - |
| Steps | ✓ | - | - |
| Condition | ✓ | - | - |
| Loop | ✓ | - | - |
| Router | ✓ | - | ✓ |
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_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,
requires_confirmation=True,
confirmation_message="Delete 1000 records?",
on_reject=OnReject.skip, # cancel | skip (default)
)
for req in run_output.steps_requiring_confirmation:
print(req.confirmation_message)
if user_approves():
req.confirm()
else:
req.reject()
from agno.workflow.types import UserInputField
Step(
name="generate_report",
agent=report_agent,
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),
],
)
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"].
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", ...),
],
requires_user_input=True,
user_input_message="Select analysis type:",
allow_multiple_selections=False,
)
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
)
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.else_branch | For Condition only: execute else_steps (default for Condition) |
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
Developer Resources
