Python httpx and asyncio: submit and poll a Sume image job

To run a Sume image job from Python, POST to /v1/images with mode: "async" and an Idempotency-Key, read status_url and next_poll_after_seconds off the 202 envelope, poll until terminal is true, then fetch result_url. The recipe below does that with one httpx.AsyncClient and asyncio.run(main()), in 29 lines.

The submit and poll contract comes from Jobs and results and the OpenAPI file in the API reference. The client usage follows HTTPX's async support page, which says to use one client rather than creating clients inside a hot loop. All read 2026-10-02.

What does the full recipe look like?

Set SUME_API_KEY in your environment and run it. The model id sume/auto is the one the image docs recommend for new integrations; the job still runs in the background, so the script never holds a request open for the whole generation.

import asyncio, os
import httpx
BASE = "https://api.sume.com"
AUTH = {"Authorization": f"Bearer {os.environ['SUME_API_KEY']}"}
async def main(prompt: str = "a red panda astronaut", key: str = "panda-order-8823-v1") -> list[str]:
    async with httpx.AsyncClient(base_url=BASE, headers=AUTH, timeout=30) as c:
        r = await c.post("/v1/images", headers={"Idempotency-Key": key},
                         json={"model": "sume/auto", "prompt": prompt, "mode": "async"})
        r.raise_for_status()
        job = r.json()["data"]
        delay = job["next_poll_after_seconds"] or 2
        for _ in range(200):  # your own deadline, not the API's
            await asyncio.sleep(delay)
            s = await c.get(job["status_url"])
            if s.status_code == 429:
                delay = float(s.headers.get("retry-after", delay * 2))
                continue
            s.raise_for_status()
            st = s.json()["data"]
            if st["terminal"]:
                break
            delay = st["next_poll_after_seconds"] or min(delay * 2, 30)
        else:
            raise TimeoutError(f"still running: {job['status_url']}")
        if st["sume_status"] != "completed":
            raise RuntimeError((await c.get(f"/v1/jobs/{job['request_id']}")).text)
        res = await c.get(job["result_url"])
        return [a["url"] for a in res.json()["data"]["result"]["artifacts"]]
print(asyncio.run(main()))

Why is each step written that way?

The table maps each line of behavior to the page that justifies it.

What would I change for production?

Replace the fixed key with one derived from your business intent, and store the job id before you start polling so a restart can resume from status_url. If many jobs run at once, bound concurrency with a semaphore and read generation_limits from the submit response, as in the pacing post.

For long video jobs, prefer a webhook and keep this loop as the backup. Job webhooks are terminal-only: job.completed, job.failed, and job.canceled.

What does this recipe not cover?

Three things are deliberately left out to stay under 30 lines.

• Network-level retries on the submit; add them only together with the idempotency key.
• Webhook signature checks; see the webhooks page.
• A 200 response: POST /v1/images can return a finished image inline instead of a job envelope, so the recipe expects the async 202.
• Image parameters such as aspect_ratio; read the image model list in the docs.