Refer CreativeAI.
Earn rewards.

import openai  # standard OpenAI SDK

client = openai.OpenAI(
    base_url="https://api.creativeai.run/v1",
    api_key="rph_live_YOUR_KEY",  # per-client key
)

# Image generation — synchronous
image = client.images.generate(
    model="gpt-image-1",
    prompt="Product photo: wireless earbuds on marble surface",
)

# Video generation — async with webhook delivery
import requests
resp = requests.post(
    "https://api.creativeai.run/v1/video/generations",
    headers={"Authorization": "Bearer rph_live_YOUR_KEY"},
    json={
        "model": "kling-v2-master",
        "image_url": image.data[0].url,
        "prompt": "Slow 360-degree rotation, studio lighting",
        "webhook_url": "https://your-app.com/hooks/render-done",
    },
)
# → webhook fires on completion with signed payload

import asyncio, httpx

API = "https://api.creativeai.run/v1"
KEY = "rph_live_YOUR_KEY"
HEADERS = {"Authorization": f"Bearer {KEY}"}
WEBHOOK = "https://your-app.com/hooks/render-done"

# Your product catalog — CSV, database, or Shopify API
products = [
    {"sku": "WE-100", "image": "https://cdn.shop.com/earbuds.jpg"},
    {"sku": "WE-200", "image": "https://cdn.shop.com/headphones.jpg"},
    {"sku": "WE-300", "image": "https://cdn.shop.com/speaker.jpg"},
    # ... hundreds of SKUs
]

async def generate_product_video(client, product):
    """Submit one product video — webhook delivers result."""
    resp = await client.post(f"{API}/video/generations", json={
        "model": "kling-v2-master",
        "image_url": product["image"],
        "prompt": "Slow 360-degree rotation, studio lighting",
        "webhook_url": WEBHOOK,
        "metadata": {"sku": product["sku"]},
    }, headers=HEADERS)
    return {"sku": product["sku"], "job_id": resp.json()["id"]}

async def main():
    async with httpx.AsyncClient() as client:
        # Process in bounded batches to respect rate limits
        batch_size = 10
        for i in range(0, len(products), batch_size):
            batch = products[i:i + batch_size]
            jobs = await asyncio.gather(
                *[generate_product_video(client, p) for p in batch]
            )
            print(f"Submitted {len(jobs)} jobs: {jobs}")
            await asyncio.sleep(1)  # pace between batches

asyncio.run(main())
# → each webhook fires independently as renders complete
# → match by metadata.sku to update your product catalog