In Prefect 3.x, use a task’s

retry_condition_fn

to decide whether to retry based on the raised exception. The function receives

(task, task_run, state)

and must return

True

to retry or

False

to skip retries. Example: don’t retry on

ValueError

, retry on everything else.

from prefect import task, flow

def retry_condition(task, task_run, state) -> bool:
    # Get the exception without raising it
    exc = state.result(raise_on_failure=False)
    return not isinstance(exc, ValueError)

@task(retries=3, retry_condition_fn=retry_condition)
def might_fail(x: int):
    if x < 0:
        raise ValueError("bad input")
    raise RuntimeError("transient issue")

@flow
def run():
    might_fail(-1)  # Will NOT retry (ValueError)
    might_fail(1)   # Will retry (RuntimeError)

Variant for multiple non-retry exceptions:

def retry_condition(task, task_run, state) -> bool:
    exc = state.result(raise_on_failure=False)
    non_retry = (ValueError, KeyError)
    return not isinstance(exc, non_retry)

Notes: - This is task-level only; flows do not have

retry_condition_fn

. - The function can be sync or async. Docs: Retry with a custom condition If you share which exceptions you want to skip, I can tailor the snippet.