Makernice

Use the Makernice API with FastAPI

This FastAPI app has two endpoints: POST /videos quotes and creates a generation on the server, and POST /makernice-webhook receives the finished generation and verifies its signature. It uses requests for the API calls and nothing else beyond FastAPI.

Makernice Team ·

Before you start

  • Create an API key in the console and set MAKERNICE_API_KEY.
  • Copy the webhook signing secret from console settings into MAKERNICE_WEBHOOK_SECRET.
  • Set APP_URL to the app's public https address, then pip install fastapi uvicorn requests and run uvicorn main:app.

Complete app

main.py
# main.py. Run: pip install fastapi uvicorn requests && uvicorn main:app
import hashlib
import hmac
import json
import os
import time
import uuid

import requests
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, Field


# raw_body: the body exactly as received (bytes). Don't re-serialize parsed JSON.
def verify_webhook(raw_body: bytes, signature_header: str, secret: str, tolerance_seconds: int = 300) -> bool:
    parts = dict(p.split("=", 1) for p in signature_header.split(",") if "=" in p)
    try:
        t = int(parts["t"])
    except (KeyError, ValueError):
        return False
    if abs(time.time() - t) > tolerance_seconds:
        return False
    expected = hmac.new(secret.encode(), f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, parts.get("v1", ""))


API = "https://api.makernice.com/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['MAKERNICE_API_KEY']}"}
WEBHOOK_SECRET = os.environ["MAKERNICE_WEBHOOK_SECRET"]
APP_URL = os.environ["APP_URL"]  # public https address of this app, for callback_url

app = FastAPI()
# Deliveries can repeat (retries, timeouts on your side). Keep seen event ids in your database in production.
seen_event_ids = set()


def post(path, body, headers=None):
    r = requests.post(API + path, json=body, headers={**HEADERS, **(headers or {})}, timeout=30)
    if not r.ok:
        # 402 insufficient_balance, 422 invalid_request, 429 rate_limited… pass the API error through
        raise HTTPException(status_code=r.status_code, detail=r.text)
    return r.json()


class VideoRequest(BaseModel):
    prompt: str = Field(min_length=1)


@app.post("/videos")
def create_video(req: VideoRequest):
    # Same job as the quickstart; swap model, duration or resolution as needed
    job = {"model": "seedance-2.5", "mode": "t2v", "prompt": req.prompt, "duration": 5, "resolution": "720p"}
    # 1. Quote: the exact amount in micro-USD, locked for 12 minutes. Nothing is charged yet
    quote = post("/generations/quote", job)
    # 2. Create. callback_url: we POST the finished generation to /makernice-webhook below
    gen = post(
        "/generations",
        {"quote_id": quote["id"], "input": job, "callback_url": f"{APP_URL}/makernice-webhook"},
        {"Idempotency-Key": str(uuid.uuid4())},
    )
    return {"id": gen["id"], "status": gen["status"], "amount": quote["amount"]}


@app.post("/makernice-webhook")
async def makernice_webhook(request: Request):
    raw = await request.body()  # raw bytes: verify before parsing
    if not verify_webhook(raw, request.headers.get("makernice-signature", ""), WEBHOOK_SECRET):
        raise HTTPException(status_code=401, detail="invalid signature")
    event = json.loads(raw)
    if event["id"] in seen_event_ids:
        return {"ok": True}
    seen_event_ids.add(event["id"])
    if event["type"] == "generation.succeeded":
        # outputs[].url is absolute here. Fetch it with your API key: it answers 302 with a signed file URL.
        # A generation can arrive again with a new event id (an output released after review); the latest one wins.
        print("ready", event["data"]["id"], [o["url"] for o in event["data"].get("outputs", [])])
    # Any 2xx counts as delivered. Answer fast and do heavy work afterwards.
    return {"ok": True}

What the code does

  1. POST /videos takes { "prompt": "…" }. An empty prompt fails validation with 422.
  2. Quote: POST /v1/generations/quote returns the exact amount as an integer string in micro-USD, locked for 12 minutes. Nothing is charged yet.
  3. Create: POST /v1/generations with the quote_id, the same input and an Idempotency-Key header. Retrying with the same key returns the same generation instead of a second one.
  4. Wait: the create call includes callback_url, so when the generation is succeeded, failed or canceled we POST it to /makernice-webhook.
  5. API errors such as 402 insufficient_balance are raised as HTTPException with the same status.

Webhook notes

  • Verify Makernice-Signature over the raw request body before parsing it. The signing secret is in console settings; callback_url deliveries use the same secret.
  • callback_url must be a public https address, so localhost does not receive webhooks during development. Poll instead, or expose your dev server through a tunnel.
  • Answer with any 2xx quickly and do heavy work afterwards. Failed deliveries are retried after 1 min, 5 min, 30 min, 2 h, 6 h, 12 h (7 attempts in total), so the same event can arrive more than once: deduplicate on the event id (it equals the Makernice-Event-Id header).
  • create_video is a plain def, so FastAPI runs it in a worker thread and the blocking requests calls don't stall the event loop. The webhook reads await request.body() to get the raw bytes before parsing.

FAQ

Can I use the same code in Django or Flask?

Yes. verify_webhook takes raw bytes, the header and the secret, so it works unchanged: pass request.body in Django (and mark the view @csrf_exempt) or request.get_data() in Flask.

Should I use httpx instead of requests?

Either works. With httpx.AsyncClient you can make create_video an async def; keep the 30-second timeout and pass the error status through the same way.

Do I pay if a generation fails?

No. The amount is held when you create the generation and released in full if it fails or is canceled. An output blocked by content moderation after the run is still charged. Details: No charge for failed generations.

Related