Async & Webhook Integration Snippets

import httpx

API_KEY = "rph_live_..."
BASE = "https://api.creativeai.run"

resp = httpx.post(
    f"{BASE}/v1/video/generations",
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={
        "model": "seedance-1.5-pro",   # or "kling-v3", "auto"
        "prompt": "A golden retriever running on the beach at sunset",
        "aspect_ratio": "16:9",
        "duration": 5,
        "webhook_url": "https://your-server.com/webhooks/creativeai",
    },
)
data = resp.json()
print(f"Job created: {data['id']}  status={data['status']}")
# → Job created: gen_abc123  status=pending

resp = httpx.post(
    f"{BASE}/v1/video/generations",
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={
        "model": "auto",             # auto-selects best provider
        "prompt": "Timelapse of a flower blooming in a garden",
        "aspect_ratio": "9:16",
        "duration": 5,
    },
)
generation_id = resp.json()["id"]
print(f"Polling mode — track: {generation_id}")

import time

def poll_until_done(generation_id: str, timeout: int = 600) -> dict:
    """Poll GET /v1/video/generations/{id} until terminal state."""
    start = time.time()
    while time.time() - start < timeout:
        resp = httpx.get(
            f"{BASE}/v1/video/generations/{generation_id}",
            headers={"Authorization": f"Bearer {API_KEY}"},
        )
        data = resp.json()
        status = data["status"]

        if status == "completed":
            print(f"Done! Video URL: {data['output_url']}")
            return data
        elif status == "failed":
            raise Exception(f"Generation failed: {data.get('error')}")

        # Still processing — wait before next poll
        print(f"  status={status} — waiting 5s...")
        time.sleep(5)

    raise TimeoutError(f"Generation {generation_id} timed out after {timeout}s")

result = poll_until_done(generation_id)

"""Flask / FastAPI webhook receiver with HMAC-SHA256 verification."""
import hmac, hashlib, json
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()
WEBHOOK_SECRET = "your-api-key"  # use your API key as signing secret

@app.post("/webhooks/creativeai")
async def handle_creativeai_webhook(request: Request):
    body = await request.body()

    # 1. Verify HMAC-SHA256 signature
    signature = request.headers.get("X-CreativeAI-Signature", "")
    expected = "sha256=" + hmac.new(
        WEBHOOK_SECRET.encode(), body, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        raise HTTPException(status_code=401, detail="Invalid signature")

    # 2. Parse the payload
    payload = json.loads(body)
    gen_id = payload["id"]          # e.g. "gen_abc123"
    status = payload["status"]      # "completed" or "failed"
    model  = payload["model"]       # requested model name
    event  = request.headers.get("X-CreativeAI-Event")
    # → "video.generation.completed"

    # 3. Handle the result
    if status == "completed":
        video_url = payload["output_url"]
        credits   = payload["credits"]
        print(f"Video ready: {video_url} ({credits} credits)")
        # → download, store, notify user, etc.
    else:
        error = payload["error_message"]
        print(f"Generation {gen_id} failed: {error}")
        # → alert, retry, etc.

    return {"received": True}

"""Complete async video generation with webhook — production pattern."""
import httpx

API_KEY = "rph_live_..."
BASE = "https://api.creativeai.run"

# Step 1: Submit job with webhook
resp = httpx.post(
    f"{BASE}/v1/video/generations",
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={
        "model": "auto",
        "prompt": "Drone shot over a misty mountain valley at sunrise",
        "aspect_ratio": "21:9",
        "duration": 5,
        "resolution": "1080p",
        "webhook_url": "https://your-app.com/webhooks/creativeai",
        "failover": True,   # auto-retry on alternate provider if primary fails
    },
)
resp.raise_for_status()
job = resp.json()

print(f"Submitted: id={job['id']}  model={job['model_actual']}  "
      f"credits={job['credits']}  failover={job['failover_used']}")

# Step 2: Your webhook handler receives the result asynchronously.
# No polling needed — just handle the POST to your webhook_url.
#
# Or, if you prefer polling, omit webhook_url and use:
#   GET /v1/video/generations/{job['id']}
# until status is "completed" or "failed".