Makernice

用 FastAPI 接入 Makernice API

这个 FastAPI 应用有两个端点：POST /videos 在服务端报价并创建生成，POST /makernice-webhook 接收完成的生成并验证签名。调用 API 用 requests，除 FastAPI 外不需要别的依赖。

Makernice 团队 ·

开始之前

  • 在控制台创建 API Key，设成 MAKERNICE_API_KEY。
  • 把控制台设置里的 webhook 签名密钥设成 MAKERNICE_WEBHOOK_SECRET。
  • 把 APP_URL 设成应用的公网 https 地址，然后 pip install fastapi uvicorn requests，运行 uvicorn main: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}

代码做了什么

  1. POST /videos 接收 { "prompt": "…" }，提示词为空时校验失败，返回 422。
  2. 报价：POST /v1/generations/quote 返回精确金额（micro-USD 整数字符串），锁定 12 分钟，这一步不扣费。
  3. 创建：POST /v1/generations 带上 quote_id、同一份输入和 Idempotency-Key 请求头。用同一个键重试，拿回的是同一个生成，不会重复下单。
  4. 等待：创建时带了 callback_url，生成变成 succeeded、failed 或 canceled 时，我们把它 POST 到 /makernice-webhook。
  5. 402 insufficient_balance 等 API 错误以同样的状态码抛出 HTTPException。

webhook 要注意的地方

  • 先对原始请求体验证 Makernice-Signature，再解析 JSON。签名密钥在控制台设置里，callback_url 的推送用同一个密钥。
  • callback_url 必须是公网 https 地址，本地开发时 localhost 收不到推送：可以先用轮询，或用隧道把开发服务器暴露出去。
  • 尽快返回任意 2xx，重活放到响应之后做。推送失败会在 1 分钟、5 分钟、30 分钟、2 小时、6 小时、12 小时 后重试（共 7 次），同一事件可能收到多次：按事件 id 去重（与 Makernice-Event-Id 请求头相同）。
  • create_video 是普通 def，FastAPI 会放到工作线程里跑，阻塞的 requests 调用不会卡住事件循环。webhook 用 await request.body() 先拿到原始字节再解析。

常见问题

Django 或 Flask 能用同一套代码吗？

能。verify_webhook 只要原始字节、签名头和密钥，不用改：Django 传 request.body（视图加 @csrf_exempt），Flask 传 request.get_data()。

可以用 httpx 代替 requests 吗？

都可以。用 httpx.AsyncClient 的话，create_video 可以改成 async def；保留 30 秒超时，错误状态码照样透传。

生成失败要付钱吗？

不用。创建时冻结金额，失败或取消就全额释放；生成完成后被内容审核拦截的仍会扣费。详见失败不扣费。

相关链接