Kerem Kaynak
05/16/2026, 12:17 AMMarvin
05/16/2026, 12:18 AMMarvin
05/16/2026, 12:19 AMpython
from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.states import Scheduled
from prefect.client.schemas.responses import SetStateStatus
async def retry_flow_run(flow_run_id: str):
async with get_client() as client:
result = await client.set_flow_run_state(
flow_run_id=UUID(flow_run_id),
state=Scheduled(message="Retrying via API"),
force=True,
)
if result.status == SetStateStatus.ACCEPT:
return result.state
else:
raise RuntimeError(f"Retry was not accepted: {result.details}")
What this does
- set_flow_run_state(...) changes the run’s state
- using Scheduled(...) puts it back in the queue
- force=True bypasses orchestration checks that might otherwise reject the transition
REST API endpoint:
text
POST /flow_runs/{flow_run_id}/set_state
Example body:
json
{
"state": {
"type": "SCHEDULED",
"name": "Scheduled",
"message": "Retrying"
},
"force": true
}
Important caveats:
- This is effectively retrying the entire flow run, not just failed tasks.
- In practice, retries make sense for runs in terminal states like FAILED, CRASHED, CANCELLED, or COMPLETED.
- If the run came from a deployment, a worker should pick it up after it is rescheduled.
- If it did not come from a deployment, rescheduling alone may not be enough to actually execute it.
There’s also a built-in CLI for this:
bash
prefect flow-run retry <flow-run-id>
I verified that command exists in Prefect 3.x.
Useful references:
- CLI help for flow-run retry: run prefect flow-run retry --help
- Prefect orchestration client docs/source are reflected in client.set_flow_run_state(...)
If you want, I can also give you:
1. a sync Python example, or
2. a raw curl example against the REST API.