Assuming Prefect 3.x: a flow only fails if an exception escapes the flow function or you return a failed State. To keep the flow green when a task fails, handle the failure explicitly and return a normal value (or a Completed state) instead of letting the exception propagate. Here are the main patterns: - Catch exceptions around a task call (simple and explicit)

from prefect import flow, task

@task
def risky():
    raise ValueError("boom")

@flow
def my_flow():
    try:
        value = risky()  # raises if it fails
    except Exception as e:
        # Handle and continue — flow will be marked COMPLETED if you return normally
        return {"status": "partial-success", "error": str(e)}

    return {"status": "ok", "value": value}

- Use .submit() and control raising with Future.result(raise_on_failure=False)

from prefect import flow, task

@task
def risky():
    raise RuntimeError("oops")

@flow
def my_flow():
    fut = risky.submit()
    out = fut.result(raise_on_failure=False)  # returns Exception instead of raising
    if isinstance(out, Exception):
        # decide how to proceed
        return {"status": "partial-success", "error": str(out)}
    return {"status": "ok", "value": out}

- Inspect task State with return_state=True

from prefect import flow, task

@task
def risky():
    raise ValueError("bad")

@flow
def my_flow():
    state = risky(return_state=True)  # returns a State, not the value
    if state.is_failed():
        exc = state.result(raise_on_failure=False)  # get the exception without raising
        return {"status": "partial-success", "error": str(exc)}
    return {"status": "ok", "value": state.result()}

- Allow failed upstreams to be passed downstream with allow_failure - Use this when you want downstream tasks to run even if an upstream task failed.

from prefect import flow, task
from prefect.utilities.annotations import allow_failure

@task
def primary():
    raise ConnectionError("primary down")

@task
def fallback():
    return {"data": "from backup"}

@task
def choose(data):
    # data will be an Exception if primary failed
    if isinstance(data, Exception):
        return fallback()
    return data

@flow
def my_flow():
    p = primary.submit()
    result = choose.submit(allow_failure(p)).result()
    return {"status": "ok", "result": result}

Important notes: - What you return from the flow determines the flow’s final state. If you return a Failed state (or an uncaught exception escapes), the flow fails. If you want the flow to succeed despite task failures, return plain data (e.g., a dict summarizing successes/failures) or a Completed state. - If you return State objects from the flow, Prefect will aggregate them. Including a failed State in the return will fail the flow. To keep the flow green, avoid returning failed State objects directly—convert to plain data first. If you’re on Prefect 2.x, the tools above still work, but behavior differs in some details. If that’s your setup, let me know and I’ll tailor the examples.