Prefect 3 task retries for a Sume job: same Idempotency-Key

In a Prefect 3 flow, wrap the Sume submit in a @task with retries, a retry_delay_seconds list and a retry_condition_fn that refuses to retry a permanent 4xx, and build the Idempotency-Key from your own order id, outside anything random. A Prefect retry runs the task function again, so a key made with uuid4() inside the task would create a new paid job on every attempt, while a key derived from the order makes each retry return the original job.

The sample below is two tasks and a flow. I ran it with Prefect 3.8.7 against a local stand-in for the API: after two injected 429 responses, one job was created and all three attempts carried the same key.

What does a Prefect retry actually repeat?

Prefect's retries documentation describes retries and retry_delay_seconds (a number or a list such as [1, 2, 4, 8]) on a task, and a retry_condition_fn that decides whether a failed run should try again. The task body runs from the top each time. That is fine for a read and dangerous for a paid create, unless the create is replay-safe, and Sume's OpenAPI says it is when you send a key: a replay with the same key and payload returns the original job with idempotency_hit: true, and a 5xx never proves the work was refused.

So the rule is the same as for any retrying runner: the key is a function of the thing being made, computed from the task arguments. The retry policy then only needs to separate two cases, which the table spells out.

What does the flow look like?

submit returns the job id, and finish polls until the job is terminal, then fetches the first artifact URL. The condition function reads the exception from the failed state, so a Refused stops retrying at once while a 429 or 5xx goes through the delay list. I also tested the conflict path: calling the flow again with the same order and a different prompt raised Refused after exactly one request, with no retry.

import os, time, requests
from prefect import flow, task
API = os.environ.get("SUME_API", "https://api.sume.com")
H = {"Authorization": f"Bearer {os.environ['SUME_API_KEY']}"}
class Refused(Exception):  # a 4xx other than 429: the same request cannot succeed
    pass
def retry_unless_refused(task, task_run, state) -> bool:
    return not isinstance(state.result(raise_on_failure=False), Refused)
@task(retries=4, retry_delay_seconds=[2, 5, 10, 20], retry_condition_fn=retry_unless_refused)
def submit(order_id: str, prompt: str) -> str:
    r = requests.post(f"{API}/v1/image-1.0/generate", timeout=30,
                      json={"prompt": prompt, "mode": "async"},
                      headers={**H, "Idempotency-Key": f"order-{order_id}-hero-v1"})
    if 400 <= r.status_code < 500 and r.status_code != 429:
        raise Refused(r.text)
    r.raise_for_status()  # 429 and 5xx raise here and retry with the SAME key
    return r.json()["data"]["request_id"]
@task(retries=3, retry_delay_seconds=5)
def finish(job_id: str) -> str:
    while True:
        s = requests.get(f"{API}/v1/jobs/{job_id}/status", headers=H, timeout=30).json()["data"]
        if s["terminal"]:
            break
        time.sleep(s.get("next_poll_after_seconds") or 3)
    res = requests.get(f"{API}/v1/jobs/{job_id}/result", headers=H, timeout=30).json()["data"]
    return res["result"]["artifacts"][0]["url"]
@flow
def hero_image(order_id: str, prompt: str) -> str:
    return finish(submit(order_id, prompt))

Where does this still go wrong?

Two details matter. First, finish polls inside one task, so a Prefect retry restarts the polling loop from the beginning. That is harmless, because status reads create nothing, but it means the loop should honour next_poll_after_seconds rather than sleeping a fixed interval. Second, if the whole flow is ever run again for the same order, for example by hand after an outage, submit is called again. That is also safe here, because the key is derived from the order, and it is the reason the version suffix in the key matters: bump it only when you want a new image, and the old job is returned otherwise, with no second charge. Log the returned idempotency_hit value if you want to see how often this happens.

Rate limits deserve a note. The delay list backs off blindly, while Sume sends retry-after on a 429 when it can. If many flows run at once, a shared concurrency limit in Prefect will protect the write budget better than any retry delay.

• Derive the key from task arguments, never from uuid4() or the clock.
• Stop retrying permanent 4xx responses with retry_condition_fn.
• Keep the poll loop on next_poll_after_seconds, and keep waits short per request.
• For a webhook instead of polling, see Sume's webhook docs and verify the signature before trusting a delivery.
• Return the artifact URL, not the bytes, so the flow result stays small.

Polling or webhook inside Prefect?

Polling inside a task is the simplest thing that works, and a Prefect worker can afford to wait while it runs. A webhook is better when a job runs long or you run thousands: the flow submits, stores the job id, and a separate small receiver resumes the work when Sume calls, which keeps workers free. Both read the same job object, so you can start with the poll and move later without changing the submit task.