用 Luma Agents API 提交视频生成任务并轮询下载结果

Dream Machine 的开发者接口现在是 Luma Agents API：提交任务后立即返回 id，需要轮询直到完成，再下载结果。 这篇做 luma_job.py：提交 → 轮询 → 马上下载（输出链接 1 小时后过期），先把任务 id 写进 jobs.jsonl，中断后可以用 --resume 接着等；失败时按 failure_code 告诉你该改提示词还是可以重试。

先看清成品

• 真实接口：无效 key 返回 401，脚本给出“key 以 luma-api- 开头”的提示。
• 替身上跑通：queued → processing → completed，自动下载；content_moderated 失败时提示“修改内容，不要原样重试”。
• 发现并修复：误用旧版 Dream Machine 地址时得到的是 403 而不是 401，原来的提示没有用。pytest 5 项通过。

合规提醒：只上传你有权使用的图片作为关键帧或参考图；不生成真实人物的误导性内容。本文只使用 Luma 官方的 Agents API（https://agents.lumalabs.ai/v1），key 在 Luma 控制台创建，放在环境变量 LUMA_AGENTS_API_KEY 中，不写进代码、不分享给别人。费用以 Luma 官方价格页为准（本文不列价格）。Dream Machine 网页版的套餐、水印和商用权利以官网为准。

运行说明：box 上没有 Luma key。我用无效 key 真实调用了两个接口（2026-10-21 20:35 和 20:47，UTC+8）：Agents API 返回 HTTP 401 {"detail":"Missing or invalid API key"}（key 不是 luma-api- 开头时）或 {"detail":"Invalid API key"}；旧版 Dream Machine API 返回 HTTP 403 {"detail":"Not authenticated"}。其余响应来自我写的本地替身，字段、状态名和 failure_code 取自官方 API 参考。没有真正生成过视频。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

77-luma-dream-machine-A-src/
├── luma_job.py
├── fake_luma.py
├── demo_local.py
└── test_luma.py

第 1 步：主脚本

"""Luma Agents API：提交生成任务 → 轮询 → 立即下载结果（输出链接 1 小时后过期）。
官方文档：https://docs.agents.lumalabs.ai/api/  key 放在环境变量 LUMA_AGENTS_API_KEY。"""
import argparse, json, os, sys, time
from pathlib import Path
import requests

BASE = os.environ.get("LUMA_AGENTS_BASE", "https://agents.lumalabs.ai/v1")
UA = "binbho-tutorial/1.0"
RETRYABLE_FAIL = {"rate_limited", "generation_failed"}  # 可以原样重新提交的失败
FIX_FAIL = {"content_moderated": "提示词或参考图被审核拒绝：修改内容，不要原样重试",
            "budget_exhausted": "额度用完：到 Luma 控制台查看余额",
            "invalid_request": "参数组合不被接受：检查模型、类型、比例和视频选项"}

class LumaError(Exception): pass

def _headers(key): return {"Authorization": f"Bearer {key}", "Content-Type": "application/json", "User-Agent": UA}

def _check(r):
    if r.status_code == 401:
        raise LumaError(f"401 {r.text.strip()}：检查 LUMA_AGENTS_API_KEY（Agents API 的 key 以 luma-api- 开头）")
    if r.status_code == 403 and "Not authenticated" in r.text:
        # 旧版 Dream Machine API（api.lumalabs.ai/dream-machine/v1）对无效 key 返回 403 而不是 401（2026-10-21 实测）
        raise LumaError(f"403 {r.text.strip()}：认证失败。你可能在用旧版 Dream Machine 地址（{BASE}）；"
                        "本脚本面向 Agents API（https://agents.lumalabs.ai/v1），两者的 key 和请求格式不同")
    if r.status_code >= 400:
        raise LumaError(f"HTTP {r.status_code}: {r.text.strip()[:300]}")
    return r.json()

def create(key, body, session=requests):
    return _check(session.post(f"{BASE}/generations", headers=_headers(key), json=body, timeout=60))

def get(key, gid, session=requests):
    return _check(session.get(f"{BASE}/generations/{gid}", headers=_headers(key), timeout=60))

def wait(key, gid, *, interval=3.0, timeout=900, session=requests, sleep=None, log=sys.stderr):
    sleep = sleep or time.sleep  # 调用时再取 time.sleep，测试里才能替换
    t0, last = time.monotonic(), None
    while True:
        g = get(key, gid, session=session)
        if g["state"] != last: print(f"  {time.strftime('%H:%M:%S')} {g['state']}", file=log); last = g["state"]
        if g["state"] in ("completed", "failed"): return g
        if time.monotonic() - t0 > timeout: raise LumaError(f"等待超过 {timeout}s，任务 {gid} 仍是 {g['state']}；稍后用 --resume {gid} 继续")
        sleep(interval)

def download(g, outdir, session=requests):
    Path(outdir).mkdir(parents=True, exist_ok=True); files = []
    for i, o in enumerate(g.get("output") or []):
        ext = ".mp4" if o["type"] == "video" else ".png"
        p = Path(outdir) / f"{g['id'][:8]}-{i}{ext}"
        with session.get(o["url"], stream=True, timeout=300) as r:
            r.raise_for_status()
            with open(p, "wb") as f:
                for b in r.iter_content(1 << 16): f.write(b)
        files.append(str(p))
    return files

def run(key, body, outdir="outputs", *, session=requests, sleep=None, resume=None, log=sys.stderr):
    gid = resume or create(key, body, session=session)["id"]
    print(f"任务 {gid}", file=log)
    Path(outdir).mkdir(parents=True, exist_ok=True)
    with open(Path(outdir) / "jobs.jsonl", "a", encoding="utf-8") as f:  # 先记下 id：进程中断后还能 --resume
        f.write(json.dumps({"id": gid, "body": body, "t": time.strftime("%F %T")}, ensure_ascii=False) + "\n")
    g = wait(key, gid, session=session, sleep=sleep, log=log)
    if g["state"] == "failed":
        code = g.get("failure_code", "")
        hint = FIX_FAIL.get(code) or ("可以重新提交" if code in RETRYABLE_FAIL else "")
        raise LumaError(f"失败 {code}: {g.get('failure_reason', '')}  {hint}")
    return download(g, outdir, session=session)

def main(argv=None):
    ap = argparse.ArgumentParser(); ap.add_argument("prompt")
    ap.add_argument("--type", default="image", choices=["image", "video"]); ap.add_argument("--model")
    ap.add_argument("--aspect", default="16:9"); ap.add_argument("--duration", choices=["5s", "10s"])
    ap.add_argument("--resolution", choices=["360p", "540p", "720p", "1080p"]); ap.add_argument("--resume")
    a = ap.parse_args(argv)
    key = os.environ.get("LUMA_AGENTS_API_KEY") or sys.exit("请先 export LUMA_AGENTS_API_KEY=…")
    body = {"prompt": a.prompt, "type": a.type, "aspect_ratio": a.aspect}
    if a.model: body["model"] = a.model
    if a.type == "video":
        body["model"] = a.model or "ray-3.2"
        v = {k: val for k, val in (("duration", a.duration), ("resolution", a.resolution)) if val}
        if v: body["video"] = v
    try:
        for f in run(key, body, resume=a.resume): print("已下载", f)
    except LumaError as e:
        sys.exit(f"错误：{e}")

if __name__ == "__main__":
    main()

第 2 步：本地替身

"""本地替身（我写的）：字段、状态名和 failure_code 取自官方 Agents API 文档；
返回的“图片/视频”是本地文件。401 的响应体照抄真实接口（2026-10-21 实测）。"""
import json, threading, uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

JOBS = {}; POLLS = {}
class H(BaseHTTPRequestHandler):
    def log_message(self, *a): pass
    def _send(self, code, obj, ctype="application/json"):
        b = obj if isinstance(obj, bytes) else json.dumps(obj).encode()
        self.send_response(code); self.send_header("Content-Type", ctype); self.send_header("Content-Length", str(len(b))); self.end_headers(); self.wfile.write(b)
    def _auth(self):
        a = self.headers.get("Authorization", "")
        if a != "Bearer luma-api-demo":
            self._send(401, {"detail": "Invalid API key" if a.startswith("Bearer luma-api-") else "Missing or invalid API key"}); return False
        return True
    def do_POST(self):
        if not self._auth(): return
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        gid = str(uuid.uuid4()); typ = body.get("type", "image")
        fail = "content_moderated" if "blocked" in body["prompt"] else None
        JOBS[gid] = {"id": gid, "created_at": "2026-10-21T12:45:00Z", "model": body.get("model", "uni-1"), "state": "queued", "type": typ, "_fail": fail}
        POLLS[gid] = 0
        self._send(201, {k: v for k, v in JOBS[gid].items() if not k.startswith("_")})  # 官方：创建成功是 201
    def do_GET(self):
        if self.path.startswith("/files/"):
            return self._send(200, b"\x89PNG fake bytes", "image/png")
        if not self._auth(): return
        gid = self.path.rsplit("/", 1)[-1]; j = JOBS[gid]; POLLS[gid] += 1
        j["state"] = ["queued", "processing", "processing", "completed"][min(POLLS[gid], 3)]
        out = {k: v for k, v in j.items() if not k.startswith("_")}
        if j["state"] == "completed":
            if j["_fail"]:
                out.update(state="failed", failure_code=j["_fail"], failure_reason="Prompt rejected by moderation")
            else:
                host = self.headers["Host"]
                out["output"] = [{"type": j["type"], "url": f"http://{host}/files/{gid}.png"}]
        self._send(200, out)

def start(port=0):
    s = ThreadingHTTPServer(("127.0.0.1", port), H); threading.Thread(target=s.serve_forever, daemon=True).start()
    return s

第 3 步：演示脚本

import os, sys
import fake_luma
srv = fake_luma.start(); os.environ["LUMA_AGENTS_BASE"] = f"http://127.0.0.1:{srv.server_port}"
import importlib, luma_job; importlib.reload(luma_job)
luma_job.time.sleep = lambda s: None  # 替身里不用真的等
sys.argv = ["luma_job.py"] + sys.argv[1:]; luma_job.main()

第 4 步：测试

import os, pytest
import fake_luma

@pytest.fixture
def lj(monkeypatch, tmp_path):
    srv = fake_luma.start(); monkeypatch.setenv("LUMA_AGENTS_BASE", f"http://127.0.0.1:{srv.server_port}")
    import importlib, luma_job; importlib.reload(luma_job); monkeypatch.chdir(tmp_path)
    yield luma_job; srv.shutdown()

def test_happy_path(lj, tmp_path):
    files = lj.run("luma-api-demo", {"prompt": "iced coffee"}, "out", sleep=lambda s: None)
    assert len(files) == 1 and (tmp_path / files[0]).read_bytes().startswith(b"\x89PNG")
    assert "iced coffee" in (tmp_path / "out/jobs.jsonl").read_text()

def test_moderated_not_retryable(lj):
    with pytest.raises(lj.LumaError, match="content_moderated.*不要原样重试"):
        lj.run("luma-api-demo", {"prompt": "blocked thing"}, "out", sleep=lambda s: None)

def test_401_hint(lj):
    with pytest.raises(lj.LumaError, match="401.*luma-api-"):
        lj.create("wrong", {"prompt": "x"})

def test_timeout_mentions_resume(lj, monkeypatch):
    t = iter([0, 0, 1000, 1000, 1000])
    monkeypatch.setattr(lj.time, "monotonic", lambda: next(t))
    gid = lj.create("luma-api-demo", {"prompt": "x"})["id"]
    with pytest.raises(lj.LumaError, match="--resume"):
        lj.wait("luma-api-demo", gid, timeout=10, sleep=lambda s: None)

def test_legacy_403_hint(lj):
    class R: status_code = 403; text = '{"detail":"Not authenticated"}'
    with pytest.raises(lj.LumaError, match="旧版 Dream Machine"):
        lj._check(R())

第 5 步：运行

$ export LUMA_AGENTS_API_KEY=luma-api-demo     # 替身认这个 key
$ python demo_local.py "A glass of iced coffee on a marble countertop"
任务 b1d103e9-4d2c-417c-883b-600dacd8a01a
  21:01:22 processing
  21:01:22 completed
已下载 outputs/b1d103e9-0.png
$ python demo_local.py "blocked prompt"          # 替身对含 blocked 的提示词返回审核失败
任务 155e38f1-8a2e-436c-a6ae-1258dde29ba8
  21:01:22 processing
  21:01:22 failed
错误：失败 content_moderated: Prompt rejected by moderation  提示词或参考图被审核拒绝：修改内容，不要原样重试
$ LUMA_AGENTS_API_KEY=bad python luma_job.py "A sunset"        # 真实接口
错误：401 {"detail":"Missing or invalid API key"}：检查 LUMA_AGENTS_API_KEY（Agents API 的 key 以 luma-api- 开头）
$ python -m pytest -q
5 passed

怎么检查结果？

• outputs/jobs.jsonl 中每次提交都有记录（id、请求体、时间）。
• 等待超时不等于失败：用 --resume <id> 继续；链接过期也一样，再 GET 一次就会拿到新链接。
• 失败为 content_moderated 时改提示词或参考图，不要原样重试。

调试示例

旧地址返回 403，提示不对路。 很多旧教程用的是旧版 Dream Machine API（https://api.lumalabs.ai/dream-machine/v1）。我把脚本的 BASE 指向旧地址、用无效 key 试了一次：

$ LUMA_AGENTS_BASE=https://api.lumalabs.ai/dream-machine/v1 LUMA_AGENTS_API_KEY=luma-api-bad python luma_job.py "A sunset"
错误：HTTP 403: {"detail":"Not authenticated"}

旧接口对无效 key 返回的是 403 Not authenticated，而不是 Agents API 的 401，于是落进了通用分支，看不出是 key 或地址的问题。修复：识别这个 403，提示“你可能在用旧版地址，本脚本面向 Agents API，两者 key 和请求格式不同”：

错误：403 {"detail":"Not authenticated"}：认证失败。你可能在用旧版 Dream Machine 地址（https://api.lumalabs.ai/dream-machine/v1）；本脚本面向 Agents API（https://agents.lumalabs.ai/v1），两者的 key 和请求格式不同

常见问题

模板规则

• 先记 id 再等待。
• 完成后立刻下载。
• 按 failure_code 决定改还是重试。
• key 只放在环境变量里。

项目资源

异步任务的关键：记下 id，别丢了结果。