<@ULVA73B9P> In Prefect 3, given the ID of a flow ...
# ask-marvin
n
@Marvin In Prefect 3, given the ID of a flow run, how do I launch a retry of that flow run using the Python SDK?
m
thought for 129.9 seconds
Yes—set the flow run’s state to AwaitingRetry via the Prefect client. That schedules a new retry attempt for that run. Async example:
Copy code
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")  # 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:
Copy code
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.