Skip to main content

Documentation Index

Fetch the complete documentation index at: https://agno-v2-rbac-doc-update.mintlify.app/llms.txt

Use this file to discover all available pages before exploring further.

Conditions support confirmation HITL, allowing users to decide which branch to execute at runtime. All HITL settings are configured via HumanReview. Conditions only support requires_confirmation. Passing unsupported fields (like requires_output_review) raises a ValueError.

​
User-Controlled Branching

When requires_confirmation=True, the condition pauses for user decision:
  • Confirm: Execute the steps branch (if branch)
  • Reject: Behavior depends on on_reject setting
from agno.workflow import Workflow, OnReject
from agno.workflow.condition import Condition
from agno.workflow.step import Step
from agno.workflow.types import HumanReview, StepInput, StepOutput
from agno.db.sqlite import SqliteDb

def detailed_analysis(step_input: StepInput) -> StepOutput:
    return StepOutput(content="Detailed analysis: Full review completed")

def quick_summary(step_input: StepInput) -> StepOutput:
    return StepOutput(content="Quick summary: Key highlights identified")

workflow = Workflow(
    name="analysis_workflow",
    db=SqliteDb(db_file="workflow.db"),
    steps=[
        Step(name="analyze", executor=analyze_data),
        Condition(
            name="analysis_depth",
            steps=[Step(name="detailed", executor=detailed_analysis)],
            else_steps=[Step(name="quick", executor=quick_summary)],
            human_review=HumanReview(
                requires_confirmation=True,
                confirmation_message="Perform detailed analysis?",
                on_reject=OnReject.else_branch,
            ),
        ),
        Step(name="report", executor=generate_report),
    ],
)

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

if run_output.is_paused:
    for req in run_output.steps_requiring_confirmation:
        print(f"Decision: {req.step_name}")
        print(f"Message: {req.confirmation_message}")
        
        if input("Confirm? (y/n): ").lower() == "y":
            req.confirm()
            print("Executing 'if' branch")
        else:
            req.reject()
            print("Executing 'else' branch")
    
    run_output = workflow.continue_run(run_output)

print(run_output.content)

​
Parameters

ParameterTypeDescription
requires_confirmationboolPause for user decision
confirmation_messagestrMessage shown to the user
on_rejectOnRejectAction when rejected

​
OnReject Options

ValueBehavior
OnReject.else_branchExecute else_steps (default)
OnReject.skipSkip the entire condition
OnReject.cancelCancel the workflow

​
Branch Execution

User Actionon_rejectResult
ConfirmanyExecute steps
Rejectelse_branchExecute else_steps
RejectskipSkip condition, continue workflow
RejectcancelCancel workflow

​
Without else_steps

If no else_steps are defined and user rejects with on_reject=OnReject.else_branch, the condition is skipped: 
Condition(
    name="optional_processing",
    steps=[Step(name="process", executor=process)],
    # No else_steps defined
    human_review=HumanReview(
        requires_confirmation=True,
        confirmation_message="Run optional processing?",
        on_reject=OnReject.else_branch,  # Will skip if rejected
    ),
)

​
Combining with Evaluator

When requires_confirmation=True, the evaluator is ignored. The user’s decision takes precedence: 
Condition(
    name="user_controlled",
    # evaluator is ignored when requires_confirmation=True
    steps=[Step(name="if_branch", ...)],
    else_steps=[Step(name="else_branch", ...)],
    human_review=HumanReview(
        requires_confirmation=True,
        confirmation_message="Proceed with if branch?",
    ),
)

​
Streaming

Handle condition 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()  # or req.reject()
    
    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