> Markdown copy of https://makernice.com/zh-CN/docs/fastapi/ for AI tools. The HTML page is the canonical version.

# 用 FastAPI 接入 Makernice API

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

[Makernice 团队](https://makernice.com/zh-CN/about/) · 更新于 2026 年 11 月 11 日

## 开始之前

-   在[控制台](https://makernice.com/console/keys/)创建 API Key，设成 `MAKERNICE_API_KEY`。
-   把[控制台设置](https://makernice.com/console/settings/)里的 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。签名密钥在[控制台设置](https://makernice.com/console/settings/)里，`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 秒超时，错误状态码照样透传。

### 生成失败要付钱吗？

不用。创建时冻结金额，失败或取消就全额释放；生成完成后被内容审核拦截的仍会扣费。详见[失败不扣费](https://makernice.com/zh-CN/pricing/failed-generations/)。

## 相关链接

-   [API 文档：快速上手](https://makernice.com/zh-CN/docs/#quickstart)
-   [API 文档：Webhooks](https://makernice.com/zh-CN/docs/#webhooks)
-   [API 文档：错误码](https://makernice.com/zh-CN/docs/#errors)
-   [用 Python 调用 Makernice API](https://makernice.com/zh-CN/docs/python/)
-   [用 Node.js 调用 Makernice API](https://makernice.com/zh-CN/docs/node/)
-   [用 Next.js 接入 Makernice API](https://makernice.com/zh-CN/docs/nextjs/)
-   [在 Claude Code、Cursor 等 AI 工具里用（MCP）](https://makernice.com/zh-CN/mcp/)