Step HITL

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"),
    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:
        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)

from agno.workflow import Workflow
from agno.workflow.step import Step
from agno.workflow.types import 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,
            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)
            
            # Convert to appropriate type
            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)

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")
    
    # Process with user-provided values
    return StepOutput(content=f"Done with {threshold}, {mode}")

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),
    ],
    ...
)

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]