Python

The official Python SDK is published on PyPI as hyperclipco (the bare hyperclip was already taken). The module imports as hyperclipco. This page shows production patterns built on top of it. For the full method reference see SDKs.

Install

pip install hyperclipco

Requires Python 3.9+. Pulls in httpx as the only runtime dependency.

Basic usage

import uuid
from hyperclipco import Hyperclip

# reads HYPERCLIP_API_KEY from os.environ
# base URL defaults to Hyperclip production
hc = Hyperclip()

ref = hc.runs.create(
    flow_id="a3c7f1ea-...",
    inputs={"prompts": {0: "A lighthouse keeper hears footsteps upstairs"}},
    idempotency_key=str(uuid.uuid4()),
)

run = hc.runs.wait(
    ref["id"],
    on_update=lambda r: print(f"step {r.get('current_step')} · {r['status']}"),
)

assert run["status"] == "completed", run
print(f"Video ready: {run['video_url']} ({run['credits_charged']} credits)")

Download helper

The video_url is a permanent public URL on video.hyperclip.co. Stream it to disk if your workflow needs a local copy:

import httpx

def download(url: str, dest: str) -> None:
    with httpx.stream("GET", url, timeout=120.0) as r:
        r.raise_for_status()
        with open(dest, "wb") as f:
            for chunk in r.iter_bytes(chunk_size=1 << 16):
                f.write(chunk)

download(run["video_url"], "output.mp4")

CSV-driven batch jobs

Process a CSV of prompts with a thread pool. Each row gets a deterministic idempotency key so crashes are safe to retry — already-processed rows return their existing completed run.

Show batch.py

import csv, sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from hyperclipco import Hyperclip, HyperclipError

CONCURRENCY = 4
FLOW_ID = "a3c7f1ea-..."


def process_row(hc: Hyperclip, row: dict) -> tuple[str, str]:
    row_id = row["id"]
    try:
        ref = hc.runs.create(
            flow_id=FLOW_ID,
            inputs={"prompts": {0: row["prompt"]}},
            idempotency_key=f"row-{row_id}",
        )
        run = hc.runs.wait(ref["id"])
        if run["status"] != "completed":
            raise RuntimeError(run.get("error_message") or run["status"])
        print(f"✓ {row_id} · {run['credits_charged']} credits")
        return row_id, run.get("video_url") or ""
    except HyperclipError as err:
        print(f"✗ {row_id} · {err}", file=sys.stderr)
        return row_id, ""


def main(csv_path: str) -> None:
    hc = Hyperclip()  # reads HYPERCLIP_API_KEY from env; base URL defaults to production

    with open(csv_path, newline="") as f:
        rows = list(csv.DictReader(f))

    with ThreadPoolExecutor(max_workers=CONCURRENCY) as pool:
        futures = [pool.submit(process_row, hc, r) for r in rows]
        with open("results.csv", "w", newline="") as out:
            writer = csv.writer(out)
            writer.writerow(["id", "video_url"])
            for future in as_completed(futures):
                writer.writerow(future.result())


if __name__ == "__main__":
    main(sys.argv[1])

The SDK’s Hyperclip instance is thread-safe — share one across worker threads instead of creating one per task. The underlying httpx.Client uses a connection pool you’d otherwise reset on every call.

FastAPI proxy

Expose a thin proxy so your frontend never touches the Hyperclip key.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from hyperclipco import Hyperclip, HyperclipError

app = FastAPI()
hc = Hyperclip()  # reads env vars


class GenerateRequest(BaseModel):
    user_id: str
    prompt: str


@app.post("/generate")
def generate(req: GenerateRequest):
    try:
        ref = hc.runs.create(
            flow_id="a3c7f1ea-...",
            inputs={"prompts": {0: req.prompt}},
            idempotency_key=f"user-{req.user_id}",
        )
        return {"run_id": ref["id"]}
    except HyperclipError as err:
        raise HTTPException(status_code=err.status, detail={"code": err.code, "message": err.message})


@app.get("/generate/{run_id}")
def status(run_id: str):
    run = hc.runs.get(run_id)
    return {
        "status": run["status"],
        "progress": run.get("current_step"),
        "video_url": run.get("video_url"),
        "error": (
            {"code": run["error_code"], "message": run["error_message"]}
            if run.get("error_code") else None
        ),
    }

The frontend POSTs to /generate, receives a run_id, then polls GET /generate/<id> every few seconds.

Inline FlowSchema (no saved flow)

ref = hc.runs.create(
    flow_schema={
        "version": 1,
        "nodes": [
            {"id": "node-0", "type": "text_input",    "label": "Prompt",   "position": {"x": 0,   "y": 100}, "config": {}},
            {"id": "node-1", "type": "text_to_video", "label": "Generate", "position": {"x": 300, "y": 100}, "config": {"model": "kling-3.0"}},
        ],
        "edges": [
            {"id": "edge-0", "source": "node-0", "target": "node-1", "sourceHandle": "text-out", "targetHandle": "text-in"},
        ],
    },
    inputs={"prompts": {0: "a lighthouse at dawn, 35mm film grain"}},
)

See inline FlowSchema for the full shape and node catalog.

Async with httpx

The SDK is synchronous today. For async codebases, run blocking calls in a thread pool:

import asyncio
from hyperclipco import Hyperclip

hc = Hyperclip()

async def generate(prompt: str) -> dict:
    ref = await asyncio.to_thread(
        hc.runs.create,
        flow_id="a3c7f1ea-...",
        inputs={"prompts": {0: prompt}},
    )
    run = await asyncio.to_thread(hc.runs.wait, ref["id"])
    return run

asyncio.run(generate("a lighthouse at dawn"))

A native async client is on the roadmap — track hyperclip-sdks for releases.

Bring-your-own httpx client

Pass a pre-configured httpx.Client to set transport-level options like proxies, retries, or custom headers:

import httpx
from hyperclipco import Hyperclip

client = httpx.Client(
    timeout=httpx.Timeout(30.0, connect=5.0),
    transport=httpx.HTTPTransport(retries=2),
    proxies="http://corp-proxy:8080",
)

hc = Hyperclip(client=client)

The SDK won’t close a client you supplied — manage its lifecycle yourself (or use it as a context manager: with httpx.Client(...) as c: ...).

Error handling

from hyperclipco import Hyperclip, HyperclipError, HyperclipTimeoutError

try:
    ref = hc.runs.create(flow_id=FLOW_ID, inputs={"prompts": {0: prompt}})
    run = hc.runs.wait(ref["id"])
except HyperclipError as err:
    if err.code == "monthly_cap_exceeded":
        ...  # raise the cap or wait for next month
    elif err.code == "insufficient_credits":
        ...  # enable autorefill or top up
    elif err.code in ("worker_error", "render_failed"):
        ...  # safe to retry with a new idempotency_key
    elif err.status == 401:
        ...  # revoke + reissue
    else:
        raise
except HyperclipTimeoutError:
    ...  # run still going server-side — cancel or extend

See error codes for every case.

Cleanup

If you create the Hyperclip instance in a function (not at module scope), close it when you’re done — or use it as a context manager:

with Hyperclip() as hc:
    ref = hc.runs.create(flow_id=FLOW_ID, inputs=inputs)
    run = hc.runs.wait(ref["id"])
# httpx.Client is closed automatically here

Where next

SDK reference

Every method, parameter, and option.

Inputs reference

All the ways to override a flow’s defaults at run time.

Errors

Catalog of every error you can hit.

Source on GitHub

File issues, read source, watch for releases.

Introduction

Endpoints

Reference

Examples

AI tooling

Install

Basic usage

Download helper

CSV-driven batch jobs

FastAPI proxy

Inline FlowSchema (no saved flow)

Async with httpx

Bring-your-own httpx client

Error handling

Cleanup

Where next

SDK reference

Inputs reference

Errors

Source on GitHub

​Install

​Basic usage

​Download helper

​CSV-driven batch jobs

​FastAPI proxy

​Inline FlowSchema (no saved flow)

​Async with httpx

​Bring-your-own httpx client

​Error handling

​Cleanup

​Where next

SDK reference

Inputs reference

Errors

Source on GitHub

Install

Basic usage

Download helper

CSV-driven batch jobs

FastAPI proxy

Inline FlowSchema (no saved flow)

Async with httpx

Bring-your-own httpx client

Error handling

Cleanup

Where next