Nicholas Pini
03/18/2026, 4:36 PMMarvin
03/18/2026, 4:37 PMMarvin
03/18/2026, 4:39 PMfrom uuid import UUID
from datetime import datetime, timezone
from prefect.client.orchestration import get_client
from prefect.states import AwaitingRetry
flow_run_id = UUID("00000000-0000-0000-0000-000000000000") # your flow run ID
async with get_client() as client:
result = await client.set_flow_run_state(
flow_run_id=flow_run_id,
state=AwaitingRetry(
scheduled_time=datetime.now(timezone.utc), # or a future time
message="Manual retry",
),
force=True, # bypasses orchestration guards if needed
)
if result.status.value == "ACCEPT":
print("Retry scheduled.")
else:
print(f"Retry not accepted: {result.details.reason}")
Sync example:
from uuid import UUID
from datetime import datetime, timezone
from prefect.client.orchestration import get_client
from prefect.states import AwaitingRetry
flow_run_id = UUID("00000000-0000-0000-0000-000000000000")
with get_client(sync_client=True) as client:
result = client.set_flow_run_state(
flow_run_id=flow_run_id,
state=AwaitingRetry(
scheduled_time=datetime.now(timezone.utc),
message="Manual retry",
),
force=True,
)
if result.status.value == "ACCEPT":
print("Retry scheduled.")
else:
print(f"Retry not accepted: {result.details.reason}")
Notes:
- Use a timezone-aware scheduled_time (e.g., datetime.now(timezone.utc)) for immediate retry, or set a future time to delay it.
- force=True is often necessary to override orchestration rules that might otherwise reject the transition. The worker will pick up the run when it becomes due.