Router HITL

Let users select routes or confirm automated routing decisions.

Routers support two HITL modes: user selection (user chooses routes) and confirmation (user approves automated routing).

User Selection

Let users choose which route(s) to execute. The router pauses and presents available choices.

from agno.workflow import Workflow
from agno.workflow.router import Router
from agno.workflow.step import Step
from agno.workflow.types import StepInput, StepOutput
from agno.db.sqlite import SqliteDb

def quick_analysis(step_input: StepInput) -> StepOutput:
    return StepOutput(content="Quick analysis: Basic metrics computed")

def deep_analysis(step_input: StepInput) -> StepOutput:
    return StepOutput(content="Deep analysis: Full statistical analysis")

def custom_analysis(step_input: StepInput) -> StepOutput:
    return StepOutput(content="Custom analysis: User-defined parameters")

workflow = Workflow(
    name="analysis_workflow",
    db=SqliteDb(db_file="workflow.db"),
    steps=[
        Step(name="prepare", executor=prepare_data),
        Router(
            name="analysis_router",
            choices=[
                Step(name="quick", description="Fast analysis (2 min)", executor=quick_analysis),
                Step(name="deep", description="Full analysis (10 min)", executor=deep_analysis),
                Step(name="custom", description="Custom parameters", executor=custom_analysis),
            ],
            requires_user_input=True,
            user_input_message="Select analysis type:",
            allow_multiple_selections=False,
        ),
        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_route:
        print(f"Router: {req.step_name}")
        print(f"Message: {req.user_input_message}")
        print(f"Options: {req.available_choices}")
        
        choice = input("Select: ")
        req.select(choice)
    
    run_output = workflow.continue_run(run_output)

print(run_output.content)

Parameters

Parameter	Type	Description
requires_user_input	bool	Pause for user to select route(s)
user_input_message	str	Message shown to the user
allow_multiple_selections	bool	Allow selecting multiple routes (default: False)

Selection Methods

Method	Description
req.select("route_name")	Select a single route
req.select_single("route_name")	Select exactly one route
req.select_multiple(["a", "b"])	Select multiple routes (requires allow_multiple_selections=True)

Multiple Selection

Allow users to select multiple routes. Selected routes execute in sequence.

Router(
    name="processing_pipeline",
    choices=[
        Step(name="clean", description="Clean data", executor=clean_data),
        Step(name="validate", description="Validate data", executor=validate_data),
        Step(name="enrich", description="Enrich data", executor=enrich_data),
        Step(name="transform", description="Transform data", executor=transform_data),
    ],
    requires_user_input=True,
    user_input_message="Select processing steps:",
    allow_multiple_selections=True,
)

Handle multiple selections:

for req in run_output.steps_requiring_route:
    print(f"Available: {req.available_choices}")
    
    # User selects: "clean, validate, transform"
    selections = input("Select (comma-separated): ").split(",")
    selections = [s.strip() for s in selections]
    
    req.select_multiple(selections)

The selected steps execute in the order they appear in choices, not the selection order.

Confirmation Mode

Confirm automated routing decisions. A selector function determines the route, but the user must approve before execution.

def route_by_priority(step_input: StepInput) -> str:
    content = step_input.previous_step_content or ""
    if "urgent" in content.lower():
        return "urgent_handler"
    elif "billing" in content.lower():
        return "billing_handler"
    return "general_handler"

Router(
    name="request_router",
    choices=[
        Step(name="urgent_handler", executor=handle_urgent),
        Step(name="billing_handler", executor=handle_billing),
        Step(name="general_handler", executor=handle_general),
    ],
    selector=route_by_priority,
    requires_confirmation=True,
    confirmation_message="Proceed with the selected route?",
)

Handle confirmation:

for req in run_output.steps_requiring_confirmation:
    print(f"Router: {req.step_name}")
    print(f"Message: {req.confirmation_message}")
    
    if input("Confirm? (y/n): ").lower() == "y":
        req.confirm()
    else:
        req.reject()

Confirmation Parameters

Parameter	Type	Description
requires_confirmation	bool	Pause for user to confirm routing decision
confirmation_message	str	Message shown to the user
on_reject	OnReject	Action when rejected: skip (default), cancel

User Selection vs Confirmation

Mode	Selector	User Action	Use Case
User Selection	None	Chooses route(s)	Interactive wizards, user-driven workflows
Confirmation	Function	Approves/rejects	Oversight of automated decisions

Use user selection when the user should decide the path. Use confirmation when the system decides but needs human approval.

Streaming

Handle router 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 router: {event.step_name}")

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

while run_output.is_paused:
    for req in run_output.steps_requiring_route:
        req.select(req.available_choices[0])
    
    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

• Workflow HITL overview
• Router reference
• Branching workflow pattern