Early Stopping

Stop a workflow from any step by returning StepOutput with stop=True.

Return StepOutput(stop=True) from a custom step to skip the remaining workflow steps.

from agno.workflow import Step, Workflow, StepInput, StepOutput


def security_scan(step_input: StepInput) -> StepOutput:
    source = str(step_input.input)
    result = "VULNERABLE" if "exec(" in source else "SECURE"
    return StepOutput(content=result)


def security_gate(step_input: StepInput) -> StepOutput:
    security_result = step_input.previous_step_content or ""

    if "VULNERABLE" in security_result.upper():
        return StepOutput(
            content="SECURITY ALERT: Critical vulnerabilities detected. Deployment blocked.",
            stop=True,
        )

    return StepOutput(content="Security check passed.")


def deploy_code(step_input: StepInput) -> StepOutput:
    return StepOutput(content="Code deployed")


def configure_monitoring(step_input: StepInput) -> StepOutput:
    return StepOutput(content="Monitoring configured")


workflow = Workflow(
    name="secure_deployment",
    steps=[
        Step(name="security_scan", executor=security_scan),
        Step(name="security_gate", executor=security_gate),
        Step(name="deploy_code", executor=deploy_code),
        Step(name="configure_monitoring", executor=configure_monitoring),
    ],
)

run_output = workflow.run("exec(input('Enter command: '))")
print(run_output.content)
# SECURITY ALERT: Critical vulnerabilities detected. Deployment blocked.

The workflow records the stopping step’s output and does not start later top-level steps. Steps, Condition, Router, and Loop propagate stop=True from nested steps. Other branches in a Parallel may already be running when one branch requests a stop.

Developer Resources

• Early Stop Workflow