Integrating an AI Video Generation API in Python: Polling, Webhooks, and Cost Control

Choosing your result-retrieval pattern

Video generation is slow compared with most API calls. A request that produces a few seconds of footage can take anywhere from tens of seconds to several minutes, which means the HTTP response that creates the job is almost never the response that contains the video. Before you write a line of Python, decide how you will retrieve the result. There are three practical patterns.

Pattern 1: Submit and poll

You send a generation request, receive a job handle back immediately, then call a status endpoint on an interval until the job reaches a terminal state. This is the simplest pattern to implement and debug: it is plain request/response code, it works behind any firewall, and it has no inbound networking requirements. Its downside is latency floor — the result is only as fresh as your polling interval — and wasted calls when you poll faster than the job can progress.

Polling fits interactive workloads: a user is waiting in a UI, the job is expected to finish in a bounded time, and you want a single code path you can reason about.

Pattern 2: Webhook callback

You submit the job with a callback URL, return control to your caller, and let the platform POST the result to your endpoint when the job completes. This removes polling entirely and gives you near-instant notification. It also survives restarts of the process that submitted the job, because delivery is driven from outside.

The cost is operational: you need a publicly reachable HTTPS endpoint, you need to verify that callbacks are genuinely from the provider, and you need to handle the case where a callback never arrives — because networks fail, endpoints get misconfigured, and payloads get lost.

Webhooks fit long-running or batch workloads: large queues of jobs, asynchronous pipelines, and anything where holding a connection or spinning a loop for minutes is wasteful.

Pattern 3: Hybrid with fallback

The pattern most production systems end up with: use webhooks as the primary signal, but keep a slow periodic reconciliation that polls any job still marked pending after a staleness threshold. If a callback is missed, the poller catches it. If the poller is delayed, the callback usually gets there first and marks the job done.

Pick per workload, not per codebase. An interactive endpoint can poll; a background queue can use callbacks; the shared job store and terminal-state logic should be identical for both, so the retrieval strategy can change without touching the rest of the system.

Setting up the client and submitting a job

The submission skeleton should stay small. Everything below assumes a single place where your API key and base URL live, and a single function that turns a prompt plus parameters into a job handle.

Config and key handling

Never hardcode the key. Read it from the environment and fail loudly at startup if it is missing, so you do not discover the problem on the first customer request.

import os

BASE_URL = os.environ.get("HIAPI_BASE_URL", "https://api.hiapi.ai")
API_KEY = os.environ["HIAPI_API_KEY"]  # KeyError at boot is intentional

Keep the key out of logs. If you log requests for debugging, log the prompt and parameters but redact the authorization header.

A single submit call

The submit call should do exactly one thing: validate the parameters you control, send the request, and return the job identifier. Do not start polling inside it, and do not download anything inside it.

import requests

def submit_video_job(prompt: str, *, idempotency_key: str, **params) -> dict:
    payload = {"prompt": prompt, **params}
    resp = requests.post(
        f"{BASE_URL}/v1/video/generations",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
            "Idempotency-Key": idempotency_key,
        },
        json=payload,
        timeout=(5, 30),  # connect, read
    )
    resp.raise_for_status()
    return resp.json()

The idempotency key matters. If your submit call times out, you do not actually know whether the job was created. Retrying with the same key lets you resend safely instead of generating — and paying for — a duplicate.

What to persist immediately

The moment submission returns, write a row to your own database before doing anything else:

• the job ID and idempotency key
• your own internal request ID and the user or batch it belongs to
• the exact request parameters, so you can reproduce or audit it
• a status of pending and a timestamp

If your process crashes after submission but before persisting, the job exists upstream with no local record. Persisting first, then proceeding, keeps your database the source of truth for what you have to reconcile.

Polling for results the right way

A polling loop is easy to write badly. The naive version — a while True with no sleep — burns quota, hammers the API, and makes your own logs unreadable. The correct loop has four properties: an interval, backoff, a hard cap, and a terminal-state check.

The loop

import time, random

def poll_job(job_id: str, *, max_attempts: int = 60, base: float = 2.0, cap: float = 20.0):
    for attempt in range(1, max_attempts + 1):
        resp = requests.get(
            f"{BASE_URL}/v1/video/generations/{job_id}",
            headers={"Authorization": f"Bearer {API_KEY}"},
            timeout=(5, 15),
        )
        resp.raise_for_status()
        body = resp.json()
        status = body["status"]

        if status == "succeeded":
            return body
        if status in {"failed", "cancelled"}:
            raise RuntimeError(f"job {job_id} ended as {status}: {body.get('error')}")

        delay = min(cap, base * (2 ** (attempt - 1)))
        time.sleep(delay + random.uniform(0, 1.0))

    raise TimeoutError(f"job {job_id} still pending after {max_attempts} attempts")

Why each piece matters

Backoff. Generation jobs make progress in stages: queued, then running, then rendering. Polling every second in the early queued phase returns the same answer twenty times. Starting at a couple of seconds and doubling until a cap keeps call volume proportional to job duration rather than to your patience.

Jitter. Adding a random fraction of a second prevents every client you run from polling in lockstep after a deploy or a retry storm.

Max attempts and a wall-clock budget. Exponential backoff without a cap on attempts can run for hours. Pick a ceiling based on your expected worst-case job duration, and treat exceeding it as a failure you surface, not a loop that continues silently. Track elapsed time too — attempt count alone is misleading once the interval cap kicks in.

Terminal-state detection. Distinguish success, failure, and cancellation explicitly. A job that failed should not be retried by the poller; it should be reported with the provider's error so you can decide whether to resubmit with different parameters.

One more rule: do not poll from inside a web request handler. Move the loop to a background worker so a slow job does not tie up your request thread pool.

Receiving results with webhook callbacks

A webhook endpoint is a public, unauthenticated-by-default HTTP route that anyone can POST to. Treat it accordingly.

Verify the sender

If the platform signs callbacks, verify the signature over the raw request body before parsing JSON — signature checks run against bytes, and re-serializing the payload can change them. Compare with a constant-time function, and reject mismatches with a non-2xx status. If signatures are not available, at minimum use a secret path segment plus an allowlist or a shared header token, and never trust the payload's contents until you have matched it to a job you created.

Return fast, work later

Your endpoint's only job is to acknowledge receipt. Parse, verify, enqueue, respond. Anything slow — downloading the video, calling your database under lock, invoking downstream services — happens in a worker, not in the handler. Providers retry callbacks that do not get a prompt success response, and a slow handler turns one event into several.

from flask import Flask, request, abort

app = Flask(__name__)

@app.post("/callbacks/video")
def video_callback():
    raw = request.get_data()
    if not verify_signature(raw, request.headers):
        abort(401)
    event = request.get_json(force=True)
    job_id = event.get("job_id")
    if not job_id:
        abort(400)
    queue.enqueue("process_video_event", event)  # returns immediately
    return {"ok": True}, 200

Reconcile against your record

Before acting on a callback, look up the job ID in your own table. Three things follow from that lookup:

• Unknown job ID: this may be a stale event from a deleted job or a forged request. Log it and drop it. Do not create records from inbound payloads.
• Job already marked complete: this is a duplicate delivery. Acknowledge it and do nothing else. Re-downloading and re-billing against the same job is the classic webhook bug.
• Job pending: mark it complete (or failed), store the asset reference, and hand off to the download step.

Make the state transition conditional on the current status — an update that only succeeds if the job is still pending — so two concurrent deliveries cannot both win. Combine that with a hybrid poller as described earlier, and no job goes missing even when a callback never arrives.

Timeouts, retries, and failure handling

Retries are where integrations quietly break. The rules below apply across submit, poll, and callback, but the safe boundaries differ for each.

Classify the error first

Transient: connection resets, 5xx responses, rate-limit replies, read timeouts. These may succeed on a second attempt.

Terminal: 4xx validation errors, unsupported parameters, authentication failures, and job-level failures such as a rejected prompt. Retrying these just repeats the same failure and, for submitted jobs, repeats the cost.

When in doubt, inspect the status code and error body rather than retrying everything by default.

Backoff with jitter, and a hard stop

Use exponential backoff with random jitter for transient errors, and cap both the number of attempts and the total time spent. A retry policy without a ceiling is an outage generator: when a dependency degrades, your retry volume multiplies load exactly when it can least afford it.

def with_retries(fn, *, attempts: int = 4, base: float = 0.5, cap: float = 8.0):
    for i in range(attempts):
        try:
            return fn()
        except TransientError:
            if i == attempts - 1:
                raise
            delay = min(cap, base * (2 ** i))
            time.sleep(delay + random.uniform(0, delay / 2))

Idempotency across the boundaries

• Submit: always send the same idempotency key when retrying a submit. This is the only safe way to recover from a timeout on a request that may have created a job. Generate the key once, store it with your intended request, and reuse it.
• Poll: naturally idempotent — reads only. Safe to retry freely within your time budget.
• Callback: deliveries are at-least-once by nature. Deduplicate on job ID plus event type, and make your handler's state transition conditional so a repeat delivery is a no-op.

Surface a terminal state

Every job path should end in exactly one stored outcome: succeeded with an asset reference, or failed with a reason and the attempt history. Never let a job sit in pending indefinitely — your hybrid poller exists precisely to force a decision after a staleness threshold. Expose that state to whatever consumes the job so upstream code can decide to resubmit, alert, or refund attention rather than waiting forever.

Downloading and storing finished videos

The URL in a completion payload is temporary. It may be signed, it may expire, and its host is not something your application should depend on at read time. The final step of the integration is moving the artifact into storage you control.

Fetch the asset

Download with a streaming request so large files do not land in memory as a single bytes object, and set a timeout that reflects file size rather than a typical API call.

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

If your storage backend supports streaming writes, hand the response iterator directly to it rather than staging to a local file.

Validate before you trust it

At minimum, check that the response status was successful, the content length matches any declared size, and the file is non-trivial in size. If the completion payload includes duration or format metadata, spot-check that the downloaded artifact is plausibly a video rather than an error page or a truncated partial. A corrupt asset that reaches your users is worse than a job you mark failed and resubmit.

Move it into your own storage

Write the object to your bucket or CDN origin under a key derived from your internal job ID, and record that key on the job row. Serve users from your own URL. The benefits compound:

• Temporary provider URLs never leak into your frontend, your emails, or your logs.
• Expiry stops mattering; you can regenerate derivatives or re-encode on your own schedule.
• Retention, access control, and deletion become your policy rather than the provider's.

Keep the download step idempotent too: if the destination key already exists and is valid, skip the fetch. On a duplicate callback, this check is what prevents a second download and a second storage write.

Keeping generation costs under control

Generation is priced by the work requested — typically billed against output duration or token-equivalent units, with the exact rates shown on the platform's pricing panel. Because each call is materially more expensive than a typical text request, cost control belongs at the call site, not in a monthly review.

Cap concurrency

The single most effective guardrail is a limit on concurrent generation jobs. Without one, a retry storm, a runaway loop, or a burst of user requests can multiply spend faster than any dashboard will show you. Put a semaphore or a bounded worker pool in front of submission, and size it deliberately.

from threading import Semaphore

_submit_gate = Semaphore(5)  # tune to your budget and provider limits

def guarded_submit(prompt, *, idempotency_key, **params):
    with _submit_gate:
        return submit_video_job(prompt, idempotency_key=idempotency_key, **params)

Validate before submitting

Most accidental overspend comes from requests that should never have been sent: a duration parameter far above what the feature needs, a retry loop that resubmits a job that is still running, an empty prompt triggering a client-side bug. Validate parameters — duration, resolution, prompt length, input media references — before the request leaves your process, and reject early with a clear error.

Log job volume, not just errors

Record every submission with its parameters, duration requested, user, and idempotency key. Count jobs per feature and per user per period. Aggregated job volume is the metric that actually tracks spend, and it lets you catch anomaly patterns — one caller generating ten times the normal rate — while they are still small.

Do not pay twice for the same job

The duplicate-generation failure mode has three common sources, and each has a defense:

• Submit timeouts retried without an idempotency key. Reuse the key.
• Webhook redelivery triggering a fresh download or a fresh job. Deduplicate on job ID and make transitions conditional.
• Pollers giving up and resubmitting while the original is still running. Verify the existing job's status before creating a replacement, and track the supersession in your job table.

Treat every resubmission as a decision that needs a reason recorded, not as a default recovery path.

Checklist before shipping

Before the integration goes live, verify these decisions and failure modes end to end.

Retrieval and lifecycle

• A retrieval pattern is chosen per workload, with a hybrid fallback so a missed callback cannot lose a job.
• The job row is written before any other side effect after submission, with parameters, idempotency key, and owner recorded.
• Every job reaches exactly one terminal state; a staleness threshold forces a decision on anything left pending.
• Status transitions are conditional on the current state, so duplicate deliveries and concurrent polls are harmless.

Network behavior

• Connect and read timeouts are set on every request, including downloads.
• Transient and terminal errors are classified, and only transient ones are retried.
• Retries use exponential backoff with jitter and hard caps on attempts and elapsed time.
• Polling backoff has both an interval cap and a wall-clock budget, and poll loops run in workers rather than request handlers.

Callbacks and storage

• The callback endpoint verifies the sender against the raw body, returns fast, and queues the work.
• Every callback is reconciled against your own job record before any action.
• Finished assets are streamed into your own storage, validated, and served from your URLs only; the download step is idempotent.

Cost and operations

• Submission is gated by a concurrency limit and parameter validation on duration, resolution, and inputs.
• Job volume is logged per feature and per user, so spend is visible as a trend rather than a surprise.
• Duplicate-generation paths — submit retries, webhook redelivery, poller resubmission — are each explicitly handled.
• An alert exists for jobs stuck pending past the threshold and for concurrency limits being hit persistently.

Run one real job through each path — success, failure, missed callback, duplicate callback — and confirm the terminal state, storage write, and counters look correct. If every item above holds under that drill, the integration is ready to ship against the HiAPI.ai video generation API.