Error Handling HITL

Pause on step failures to let users retry or skip.

Steps can pause when they encounter errors, letting users decide to retry or skip the failed step.

Error Pause Mode

Set on_error=OnError.pause to pause when a step fails:

from agno.workflow import Workflow, OnError
from agno.workflow.step import Step
from agno.workflow.types import StepInput, StepOutput
from agno.db.sqlite import SqliteDb
import random

def unreliable_api_call(step_input: StepInput) -> StepOutput:
    if random.random() < 0.7:  # 70% failure rate
        raise Exception("API call failed: Connection timeout")
    return StepOutput(content="API call succeeded")

def process_data(step_input: StepInput) -> StepOutput:
    return StepOutput(content=f"Processed: {step_input.previous_step_content}")

workflow = Workflow(
    name="api_workflow",
    db=SqliteDb(db_file="workflow.db"),
    steps=[
        Step(
            name="fetch_data",
            executor=unreliable_api_call,
            on_error=OnError.pause,
        ),
        Step(name="process", executor=process_data),
    ],
)

run_output = workflow.run("Fetch and process")

while run_output.is_paused:
    for req in run_output.steps_with_errors:
        print(f"Step '{req.step_name}' failed")
        print(f"Error: {req.error_message}")
        
        choice = input("Retry or skip? (r/s): ").lower()
        if choice == "r":
            req.retry()
        else:
            req.skip()
    
    run_output = workflow.continue_run(run_output)

print(run_output.content)

OnError Options

Value	Behavior
OnError.fail	Fail the workflow immediately
OnError.skip	Skip the step and continue (default)
OnError.pause	Pause for user decision (retry or skip)

ErrorRequirement Properties

When a step fails with on_error=OnError.pause, an ErrorRequirement is created:

Property	Type	Description
step_name	str	Name of the failed step
error_message	str	The exception message
error_type	str	Exception class name (e.g., “ValueError”)
retry_count	int	Currently always 0. Track attempts in your resolution loop

ErrorRequirement Methods

Method	Description
req.retry()	Retry the failed step
req.skip()	Skip the step and continue

Retry Behavior

When you call req.retry():

1. The step executes again with the same input
2. If it fails again, the workflow pauses again
3. You can retry indefinitely or skip after some attempts

Track attempts in your resolution loop to cap retries:

attempts = 0
while run_output.is_paused:
    for req in run_output.steps_with_errors:
        if attempts < 3:
            attempts += 1
            print(f"Retrying (attempt {attempts}/3)")
            req.retry()
        else:
            print("Max retries reached, skipping")
            req.skip()

    run_output = workflow.continue_run(run_output)

Skip Behavior

When you call req.skip():

1. The failed step produces no output and is not re-executed
2. The workflow continues with the next step
3. The next step’s step_input.previous_step_content comes from the last successful step, or is None if there is none

Combining with Confirmation

A step can have both error handling and confirmation:

Step(
    name="risky_operation",
    executor=risky_function,
    requires_confirmation=True,
    confirmation_message="Execute risky operation?",
    on_error=OnError.pause,
)

The confirmation happens first. If confirmed and the step fails, the error pause activates.

Streaming

Handle error HITL in streaming workflows:

from agno.run.workflow import StepErrorEvent

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

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

while run_output.is_paused:
    for req in run_output.steps_with_errors:
        print(f"Error: {req.error_message}")
        req.retry()  # or req.skip()
    
    for event in workflow.continue_run(run_output, stream=True, stream_events=True):
        pass
    
    session = workflow.get_session()
    run_output = session.runs[-1]

Error Types

Common error scenarios and handling:

Scenario	Recommended Action
Network timeout	Retry a few times, then skip
Rate limit	Retry after delay
Invalid input	Skip (retry won’t help)
Resource unavailable	Retry or skip based on criticality

for req in run_output.steps_with_errors:
    if "timeout" in req.error_message.lower():
        if timeout_attempts < 3:  # counter tracked in your resolution loop
            timeout_attempts += 1
            req.retry()
        else:
            req.skip()
    elif "rate limit" in req.error_message.lower():
        import time
        time.sleep(5)  # Wait before retry
        req.retry()
    else:
        req.skip()  # Unknown error, skip

Developer Resources

• Workflow HITL overview
• Step reference