HappyHorse 1.1 reference-to-video API: a working curl + Python tutorial

HappyHorse 1.1 R2V turns up to nine reference images plus a text prompt into a short video, keeping subject, wardrobe and scene style consistent across shots. On HiAPI it is served through the unified async task endpoint and exposed as model id happyhorse-1.1/reference-to-video. This tutorial walks through the smallest request that works, then shows the production-shape version with a callback.

Goal and prerequisites

You will:

1. Upload (or already host) 1-9 reference images on an HTTPS URL the API can reach.
2. POST a task to https://api.hiapi.ai/v1/tasks referencing those URLs.
3. Read the final MP4 URL out of data.output[0].url once the task reaches success.

Prerequisites:

• An HiAPI account and an API key starting with sk-. Grab one from the HiAPI dashboard and read Authentication for header rules. Keys are sent as Authorization: Bearer sk-... and must never be shipped client-side.
• Image hosting: each reference image must be JPEG, PNG or WEBP, shortest side ≥ 400px, ≤ 20MB. Public CDN URLs work; signed S3 URLs work as long as they are reachable from the task worker.

Minimum working request (curl)

This is the smallest payload the schema accepts. It uses the default 1080p / 16:9 / 5s output, so the only fields you have to set are prompt and reference_image.

curl -X POST https://api.hiapi.ai/v1/tasks \
  -H "Authorization: Bearer sk-YOUR_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "happyhorse-1.1/reference-to-video",
    "input": {
      "prompt": "The character in [Image 1] walks through the scene styled like [Image 2], cinematic camera",
      "reference_image": [
        "https://static.hiapi.ai/example/ref-1.jpg",
        "https://static.hiapi.ai/example/ref-2.jpg"
      ]
    }
  }'

The response is the task envelope; the only field you actually need is data.taskId:

{
  "code": 200,
  "data": { "taskId": "task_xxx", "status": "queued" }
}

Poll for the terminal state:

curl https://api.hiapi.ai/v1/tasks/task_xxx \
  -H "Authorization: Bearer sk-YOUR_KEY"

When data.status becomes success, your video URL is data.output[0].url. The URL has an expireAt timestamp, so download the bytes or promote it to persistent storage immediately - do not hot-link it from your site.

Parameters

The R2V endpoint validates inputs strictly; unknown fields return 400 INVALID_REQUEST. The full schema (verified live against POST /v1/tasks):

Field	Type	Required	Allowed values	Default
model	string	yes	happyhorse-1.1/reference-to-video (fixed)	-
input.prompt	string	yes	Reference images with [Image 1], [Image 2], ... in the same order as reference_image	-
input.reference_image	string[]	yes	1-9 HTTPS URLs, JPEG/PNG/WEBP, shortest side ≥ 400px, ≤ 20MB each	-
input.resolution	enum	no	720p, 1080p	1080p
input.aspect_ratio	enum	no	16:9, 9:16, 3:4, 4:3, 4:5, 5:4, 1:1, 9:21, 21:9	16:9
input.duration	integer	no	3-15 (seconds)	5
callback.url	string	no	HTTPS URL receiving terminal task events	-
callback.when	enum	no	final (only value today)	final

A few non-obvious validation rules worth knowing before they bite you:

• Pass reference_image as an array even if you only have one URL. A bare string returns reference_image: got string, want array.
• duration must be an integer between 3 and 15 inclusive. 999 and 1 both fail with explicit bounds messages.
• There is no top-level seed, negative_prompt, fps, cfg_scale or steps. The API rejects unknown properties.

For full pricing per resolution and duration combination, see the HiAPI pricing page - cost scales with both resolution and duration.

Python: end-to-end with polling

The block below is a self-contained generator that submits a task, polls until terminal, and downloads the video. It is deliberately dependency-free (requests only) so you can copy it into any service. Read Quickstart for the broader request lifecycle.

import os
import time
import requests

API = "https://api.hiapi.ai/v1/tasks"
HEADERS = {"Authorization": f"Bearer {os.environ['HIAPI_API_KEY']}"}


def submit_r2v(prompt: str, reference_images: list[str], duration: int = 5,
               resolution: str = "1080p", aspect_ratio: str = "16:9") -> str:
    if not 1 <= len(reference_images) <= 9:
        raise ValueError("reference_image must contain 1-9 URLs")
    payload = {
        "model": "happyhorse-1.1/reference-to-video",
        "input": {
            "prompt": prompt,
            "reference_image": reference_images,
            "resolution": resolution,
            "aspect_ratio": aspect_ratio,
            "duration": duration,
        },
    }
    r = requests.post(API, json=payload, headers=HEADERS, timeout=60)
    r.raise_for_status()
    data = r.json()
    if data.get("code") != 200:
        raise RuntimeError(f"create failed: {data}")
    return data["data"]["taskId"]


def wait_for_video(task_id: str, timeout_s: int = 600, poll_s: int = 5) -> str:
    deadline = time.time() + timeout_s
    while time.time() < deadline:
        r = requests.get(f"{API}/{task_id}", headers=HEADERS, timeout=30)
        r.raise_for_status()
        task = r.json()["data"]
        status = task["status"]
        if status == "success":
            return task["output"][0]["url"]
        if status == "fail":
            err = task.get("error") or {}
            raise RuntimeError(f"task failed: {err.get('code')} {err.get('message')}")
        time.sleep(poll_s)
    raise TimeoutError(f"task {task_id} did not finish in {timeout_s}s")


def download(url: str, path: str) -> None:
    with requests.get(url, stream=True, timeout=300) as r:
        r.raise_for_status()
        with open(path, "wb") as f:
            for chunk in r.iter_content(1 << 16):
                f.write(chunk)


if __name__ == "__main__":
    task_id = submit_r2v(
        prompt="Keep the subject design from [Image 1] moving through the mood of [Image 2]",
        reference_images=[
            "https://static.hiapi.ai/example/char.jpg",
            "https://static.hiapi.ai/example/mood.jpg",
        ],
        duration=5,
        resolution="720p",
        aspect_ratio="9:16",
    )
    print("submitted", task_id)
    video_url = wait_for_video(task_id)
    print("ready", video_url)
    download(video_url, "out.mp4")

A few things this code does on purpose:

• Reads the key from HIAPI_API_KEY so it never lands in git or in container logs.
• Uses r.raise_for_status() to catch transport errors before parsing JSON.
• Treats success and fail as the only terminal states - everything else (queued, processing, ...) keeps the loop alive.
• Downloads the output URL right away. Output URLs are signed and expire; if you need the file later, promote it to persistent storage or copy it into your own bucket.

Production: switch from polling to a webhook

Polling every 5 seconds works for a notebook or a low-volume worker. For a real service, register a callback when you create the task so HiAPI pushes terminal events to your endpoint - your worker stays idle and you stop paying for empty GET traffic. See Create Task for the full request schema and Get Task Detail for the fallback poll.

payload = {
    "model": "happyhorse-1.1/reference-to-video",
    "input": {
        "prompt": "The character in [Image 1] walks across the city street in [Image 2]",
        "reference_image": [
            "https://cdn.example.com/char.jpg",
            "https://cdn.example.com/street.jpg",
        ],
        "resolution": "1080p",
        "aspect_ratio": "16:9",
        "duration": 8,
    },
    "callback": {
        "url": "https://your-domain.com/hiapi/callback",
        "when": "final",
    },
}

Two things to wire into the receiving endpoint:

1. Idempotency by taskId. Both success and fail are terminal, and a retry from HiAPI (or a duplicate from your own retry layer) will repost the same taskId. Dedupe before triggering any side effect (S3 copy, Slack message, billing event).
2. Reconciliation poll. If your endpoint is down, the callback is gone. Keep a thin reconciliation job that calls GET /v1/tasks/<id> for any task that has been in processing longer than your worst-case generation time.

Errors you will actually see

These are the ones worth handling explicitly; everything else is a transient 5xx that the SDK or your HTTP client should retry with backoff.

• 401 permission_denied - the Authorization header is missing, malformed or the key has been revoked. Re-issue the key from the dashboard rather than patching the header.
• 400 INVALID_REQUEST with field path - the message tells you exactly what failed, e.g. reference_image: maxItems: got 10, want 9 or aspect_ratio: value must be one of .... Validate inputs in your worker before submitting so you fail fast locally.
• task failed (terminal) - poll/callback returns status: "fail" with error.code and error.message. Common causes: a reference image URL the worker cannot fetch, an NSFW moderation hit, or images below the 400px shortest-side threshold.

FAQ

Can I reference more than 9 images? No. The schema caps reference_image at 9. If you need more variety, compose a higher-density reference image off-platform (a grid or storyboard) and treat it as a single input.

Do I have to use [Image N] markers in the prompt? They are how you point at a specific reference. A prompt that never names an image is legal but will let the model blend the references freely - usually not what you want for multi-shot consistency.

Why am I getting 503 nsfw_moderation_unavailable? That is a transient platform error from the moderation service, not your request. Retry after a short backoff.

Should I use 720p or 1080p? 1080p is the default and looks better, but it costs more and renders slower. For vertical social cuts where the player downscales anyway, 720p with aspect_ratio: 9:16 is usually the right tradeoff.

Can I pass a seed to make output reproducible? Not currently. The API rejects seed as an unknown property. If you need reproducibility, pin your prompt, reference_image, aspect_ratio, resolution and duration exactly and accept some run-to-run variation.

Where do the output URLs live, and for how long? They are signed URLs with a short TTL; the response includes expireAt. Download or promote them to persistent storage the moment a task succeeds.