Yes — in Prefect 3.x you can make task retries fully dynamic by passing them as flow parameters and applying them at runtime with

Task.with_options(...)

. Then each deployment can set different default parameters. Here’s a simple pattern you can drop in:

from prefect import flow, task
from prefect.tasks import exponential_backoff

@task
def run_dbt(cmd: list[str]):
    import subprocess
    # Example: cmd could be ["run", "--select", "model_name"]
    subprocess.run(["dbt"] + cmd, check=True)

@flow
def dbt_flow(
    command: list[str],
    task_retries: int = 2,
    retry_delay_seconds: float | list[float] = 30,
):
    # Apply runtime options based on deployment/parameters
    dbt_task = run_dbt.with_options(
        retries=task_retries,
        retry_delay_seconds=retry_delay_seconds,
        # optional: jitter or custom condition
        # retry_jitter_factor=0.5,
        # retry_condition_fn=my_condition,
    )
    dbt_task(command)

Examples of per-deployment defaults - Deployment A:

parameters={"task_retries": 2, "retry_delay_seconds": 30}

- Deployment B:

parameters={"task_retries": 4, "retry_delay_seconds": [30, 60, 120, 300]}

- Deployment C:

parameters={"task_retries": 0}

If you prefer exponential backoff:

@flow
def dbt_flow(
    command: list[str],
    task_retries: int = 4,
    backoff_factor: float = 10.0,  # seconds base
):
    dbt_task = run_dbt.with_options(
        retries=task_retries,
        retry_delay_seconds=exponential_backoff(backoff_factor),  # e.g., [10, 20, 40, 80]
    )
    dbt_task(command)

How to set per-deployment parameters - In code: when defining deployments, pass

parameters

to `.deploy(...)`:

if __name__ == "__main__":
    # A
    dbt_flow.deploy(
        name="dbt-A",
        parameters={"task_retries": 2, "retry_delay_seconds": 30},
    )
    # B
    dbt_flow.deploy(
        name="dbt-B",
        parameters={"task_retries": 4, "retry_delay_seconds": [30, 60, 120, 300]},
    )
    # C
    dbt_flow.deploy(
        name="dbt-C",
        parameters={"task_retries": 0},
    )

- Or set default parameters for each deployment in the UI or via

prefect deploy

(you’ll be prompted for parameters). Notes -

retry_delay_seconds

can be a single number, a list of per-retry delays, or a callable like

exponential_backoff

. - There’s a cap of 50 retry delays per task. -

with_options()

returns a new configured task instance; call that inside your flow for the dbt step you want to control. Docs for deeper reference: - Retries how-to: Retries (how-to) - Tasks concept/API (includes

with_options

and

exponential_backoff

): Tasks API - Flow deploy API (parameters on deployments): Flow.deploy If you share how you’re creating your deployments (UI, CLI, or code) and how you pass the dbt command, I can tailor the exact snippet for your setup.