Robert Szomolanyi
04/21/2026, 6:59 PMMarvin
04/21/2026, 6:59 PMMarvin
04/21/2026, 7:07 PMfrom prefect import flow
@flow(retries=3, retry_delay_seconds=10)
def my_flow():
# your logic
...
Notes:
- retries = number of times to retry the whole flow if it fails
- retry_delay_seconds = fixed delay (int/float) between retries
- Advanced strategies like exponential backoff or jitter are not applied at the flow level; use task-level retries for those
- Docs: Retries how-to
2) Manually retry an existing failed flow run (Python API)
Set the run back to Scheduled so a worker will pick it up again:
import asyncio
from prefect.client.orchestration import get_client
from prefect.states import Scheduled
async def retry_flow_run(flow_run_id: str):
async with get_client() as client:
await client.set_flow_run_state(flow_run_id, state=Scheduled())
# asyncio.run(retry_flow_run("<your-flow-run-id>"))
Docs: Manually retry a flow run
3) Adjust the current flow’s retry policy at runtime (advanced)
You can update the current run’s empirical policy while the flow is running:
from prefect import flow
from prefect.context import FlowRunContext
from prefect.client.orchestration import get_client
from prefect.server.schemas.core import FlowRunPolicy
from prefect.server.schemas.actions import FlowRunUpdate
@flow
async def my_flow():
ctx = FlowRunContext.get()
async with get_client() as client:
await client.update_flow_run(
ctx.flow_run.id,
FlowRunUpdate(empirical_policy=FlowRunPolicy(retries=5, retry_delay=20)),
)
...
A few clarifications
- Flow-level retries only support a single fixed delay; exponential backoff and jitter are task-only features. For advanced retry strategies, set them on tasks within the flow.
- API refs: prefect.flows.Flow, FlowRunPolicy schema
If you had a different scenario in mind (e.g., conditional retry, rescheduling at a specific time, or retrying from within a task), let me know and I can tailor examples.