Let me provide you with the specific code you need to manually retry a flow run in Prefect 3.x. Here's a complete example:

python
from prefect import get_client
from prefect.states import Retrying
from datetime import datetime, timedelta

async def retry_flow_run(flow_run_id: str):
    # Get the client
    client = get_client()
    
    # Create a Retrying state
    # You can optionally specify a scheduled_time for when the retry should occur
    retry_state = Retrying(
        scheduled_time=datetime.utcnow() + timedelta(seconds=0)  # immediate retry
    )
    
    # Set the flow run state to Retrying
    await client.set_flow_run_state(
        flow_run_id=flow_run_id,
        state=retry_state
    )

To use this code: 1. You'll need the flow run ID (a UUID string) of the run you want to retry 2. Since this uses async functions, you'll need to run it in an async context 3. The retry will be scheduled immediately, but you can adjust the

scheduled_time

if you want to delay it Here's how you might actually execute this:

python
import asyncio

# Replace with your actual flow run ID
flow_run_id = "your-flow-run-id-here"

# Run the retry function
asyncio.run(retry_flow_run(flow_run_id))

A few important notes: - This code assumes you're already authenticated with your Prefect server - The flow run must be in a state that can transition to

Retrying

(typically failed or cancelled states) - If you need to schedule the retry for a future time, simply adjust the

scheduled_time

parameter in the

Retrying

state Would you like me to explain any part of this in more detail or show you how to find the flow run ID?

for posterity, you'll need to use AwaitingRetry for it to get picked up by the infra.