Async Queue

import json
import requests
import threading
import time

API_KEY = "YOUR_API_KEY"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

def listen(results):
    with requests.get(
        "https://hub.oxen.ai/api/events",
        headers=HEADERS,
        stream=True,
    ) as resp:
        event_name = None
        for line in resp.iter_lines(decode_unicode=True):
            if line is None or line == "":
                event_name = None
                continue
            if line.startswith(":"):
                continue  # keep-alive comment
            if line.startswith("event:"):
                event_name = line[6:].strip()
            elif line.startswith("data:") and event_name == "media_generation_completed":
                payload = json.loads(line[5:].strip())
                results[payload["generation_id"]] = payload

# Start listening BEFORE enqueuing
results = {}
threading.Thread(target=listen, args=(results,), daemon=True).start()
print("SSE listener connected, waiting for events...")

# Enqueue
prompt = "A red cube"
model = "black-forest-labs-flux-2-klein-4b"
print(f"\nEnqueuing {model}")
print(f"  prompt: \"{prompt}\"")
resp = requests.post(
    "https://hub.oxen.ai/api/ai/queue",
    headers={**HEADERS, "Content-Type": "application/json"},
    json={
        "model": model,
        "prompt": prompt,
    },
)
gen_id = resp.json()["generations"][0]["generation_id"]
print(f"  generation_id: {gen_id}")

# Wait for completion event
start = time.time()
while gen_id not in results:
    elapsed = time.time() - start
    print(f"  Waiting for SSE completion event... ({elapsed:.1f}s)")
    time.sleep(2)

elapsed = time.time() - start
event = results[gen_id]
if event["status"] == "succeeded":
    print(f"\nGeneration succeeded in {elapsed:.1f}s")
    print(f"  URL: {event['url']}")
else:
    print(f"\nGeneration failed after {elapsed:.1f}s")
    print(f"  Error: {event['error']}")

import requests
import time

API_KEY = "YOUR_API_KEY"
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

# Enqueue 4 images
model = "black-forest-labs-flux-2-klein-4b"
prompt = "Abstract geometric pattern in blue and gold"
print(f"Enqueuing 4 generations of {model}")
print(f"  prompt: \"{prompt}\"")
response = requests.post(
    "https://hub.oxen.ai/api/ai/queue",
    headers=HEADERS,
    json={
        "model": model,
        "prompt": prompt,
        "num_generations": 4,
    },
)
gen_ids = [g["generation_id"] for g in response.json()["generations"]]
for gid in gen_ids:
    print(f"  generation_id: {gid}")

# Poll individual generations until all reach a terminal status
terminal = {"succeeded", "failed", "cancelled"}
start = time.time()
while True:
    statuses = {}
    for gid in gen_ids:
        resp = requests.get(
            f"https://hub.oxen.ai/api/ai/queue/{gid}",
            headers=HEADERS,
        ).json()
        statuses[gid] = resp["status"]
    done = sum(1 for s in statuses.values() if s in terminal)
    elapsed = time.time() - start
    print(f"  [{elapsed:5.1f}s] {done}/{len(gen_ids)} complete")
    if done == len(gen_ids):
        break
    time.sleep(5)

elapsed = time.time() - start
print(f"\nAll {len(gen_ids)} images generated in {elapsed:.1f}s!")
for gid, s in statuses.items():
    print(f"  {gid}: {s}")

import requests

response = requests.post(
    "https://hub.oxen.ai/api/ai/queue",
    headers={
        "Authorization": "Bearer YOUR_API_KEY",
        "Content-Type": "application/json",
    },
    json={
        "model": "kling-video-o3-pro-reference-to-video",
        "multi_prompt": [
            {"prompt": "Aerial view of waves crashing on a rocky shore", "duration": 5},
            {"prompt": "Camera pulls back to reveal the full coastline", "duration": 5},
        ],
        "aspect_ratio": "16:9",
    },
)

print(response.json())

import requests
import time

API_KEY = "YOUR_API_KEY"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

# After enqueuing, grab a generation ID
generation_id = "YOUR_GENERATION_ID"
print(f"Polling generation {generation_id}")

terminal = {"succeeded", "failed", "cancelled"}
start = time.time()
poll_count = 0
while True:
    poll_count += 1
    data = requests.get(
        f"https://hub.oxen.ai/api/ai/queue/{generation_id}",
        headers=HEADERS,
    ).json()
    elapsed = time.time() - start
    status = data["status"]
    if status in terminal:
        print(f"  [{elapsed:5.1f}s] Generation {status}")
        if status == "succeeded":
            print(f"  result_url: {data['result_url']}")
        elif status == "failed":
            print(f"  error: {data['error_message']}")
        break
    print(f"  [{elapsed:5.1f}s] Poll #{poll_count} — {status} ({data['model_name']}, {data['media_type']})")
    time.sleep(10)