Adobe Firefly：一个可以直接放进后端的 Firefly 生成客户端

这篇做一个可以直接放进后端的 Firefly 生成客户端。它会把 OAuth 令牌缓存到文件里（权限 600，过期前 5 分钟自动续期），通过 generate-async 提交任务，轮询 statusUrl 等待结果，然后立刻把预签名图片下载到本地，同时保存 seed 和 altText。 令牌被吊销时客户端会自动刷新一次再重试，遇到 429 会按 Retry-After 等待。

先看清成品（在 box 上对着模拟服务器真实运行）

• pytest 3 项全部通过。
• 两次任务只换了 1 次令牌。
• 令牌被吊销后自动刷新，任务照样成功。
• 尺寸、seed 个数不合法时返回 400，报错信息原样显示出来。

合规提醒：Firefly Services API 面向企业，需要组织开通 Firefly Services，并在 Adobe Developer Console 创建 OAuth Server-to-Server 凭据。CLIENT_SECRET 和访问令牌只能放在服务器端，不要写进前端代码、仓库或截图。提示词和参考图要遵守 Adobe 的生成式 AI 使用准则，不要生成侵犯他人肖像、版权或商标的内容。在中国境内公开发布生成内容，要遵守《生成式人工智能服务管理暂行办法》，并按《人工智能生成合成内容标识办法》加上“AI 生成”标识。Firefly 按生成额度计费，额度取决于你的合同，本文不列价格。

运行说明：box 上没有 Firefly 凭据，所以没有真正生成过图片。文中的生成流程跑在我自己写的模拟服务器 fake_firefly.py 上，接口形状照官方文档实现，图片是程序画的色块。用错误凭据对真实的 IMS 和 Firefly 接口做的请求是真实运行的，返回结果见 C 篇。

做完之后你会得到什么？

需要准备哪些环境？

接口流程（来自官方文档）

1. POST https://ims-na1.adobelogin.com/ims/token/v3：表单字段 grant_type=client_credentials、client_id、client_secret，scope 用官方给出的 openid,AdobeID,session,additional_info,read_organizations,firefly_api,ff_apis。
2. POST https://firefly-api.adobe.io/v3/images/generate-async：请求头 x-api-key（填 Client ID）和 Authorization: Bearer 令牌，返回 jobId、statusUrl、cancelUrl。
3. 轮询 statusUrl，状态从 running 变成 succeeded 或 failed；成功后结果在 result.outputs[]，每项包含 seed 和 image.url（预签名链接）。

同步版 v3 接口已在 2025-10-03 下线，新代码只能用 *-async。变体数量的字段名是 numVariations（1–4），如果同时传 seeds，个数必须和 numVariations 相等。

文件结构

49-adobe-firefly-A-src/
├── ff_client.py
├── fake_firefly.py
└── test_client.py

第 1 步：客户端

"""Adobe Firefly Services 最小客户端：OAuth 令牌缓存 + 异步生成 + 轮询 + 下载。
接口依据官方文档：IMS client_credentials 取令牌，POST /v3/images/generate-async 提交，
轮询返回的 statusUrl 直到 succeeded / failed，结果在 result.outputs[].image.url（预签名，需尽快下载）。"""
import json, os, sys, time, pathlib, argparse
import requests

IMS_URL = os.environ.get("FIREFLY_IMS_URL") or "https://ims-na1.adobelogin.com/ims/token/v3"
API_BASE = os.environ.get("FIREFLY_API_BASE") or "https://firefly-api.adobe.io"
SCOPE = "openid,AdobeID,session,additional_info,read_organizations,firefly_api,ff_apis"
UA = "binbho-tutorial-firefly/1.0 (+https://github.com/binbho)"
CACHE = pathlib.Path(os.environ.get("FIREFLY_TOKEN_CACHE", pathlib.Path.home() / ".cache/firefly_token.json"))


class FireflyError(Exception):
    def __init__(self, status, body):
        super().__init__(f"HTTP {status}: {body}")
        self.status, self.body = status, body


class Firefly:
    def __init__(self, client_id=None, client_secret=None, poll_interval=1.0, timeout=300):
        self.cid = client_id or os.environ["FIREFLY_SERVICES_CLIENT_ID"]
        self.secret = client_secret or os.environ["FIREFLY_SERVICES_CLIENT_SECRET"]
        self.poll_interval, self.timeout = poll_interval, timeout
        self.s = requests.Session()
        self.s.headers["User-Agent"] = UA

    # ---- 令牌：缓存到文件，提前 5 分钟过期，避免每次请求都去 IMS 换令牌 ----
    def token(self, force=False):
        if not force and CACHE.exists():
            c = json.loads(CACHE.read_text())
            if c.get("client_id") == self.cid and c["expires_at"] - 300 > time.time():
                return c["access_token"]
        r = self.s.post(IMS_URL, data={"grant_type": "client_credentials", "client_id": self.cid,
                                       "client_secret": self.secret, "scope": SCOPE}, timeout=30)
        if r.status_code != 200:
            raise FireflyError(r.status_code, r.text)
        d = r.json()
        CACHE.parent.mkdir(parents=True, exist_ok=True)
        CACHE.write_text(json.dumps({"client_id": self.cid, "access_token": d["access_token"],
                                     "expires_at": time.time() + int(d.get("expires_in", 3600))}))
        os.chmod(CACHE, 0o600)
        return d["access_token"]

    def _call(self, method, url, **kw):
        for attempt in range(2):
            h = {"x-api-key": self.cid, "Authorization": f"Bearer {self.token(force=attempt > 0)}",
                 "Accept": "application/json"}
            r = self.s.request(method, url, headers=h, timeout=60, **kw)
            if r.status_code == 401 and attempt == 0:
                continue                      # 令牌过期/被吊销：强制刷新一次再试
            if r.status_code == 429:
                time.sleep(int(r.headers.get("Retry-After", "5")))
                continue
            if r.status_code >= 400:
                raise FireflyError(r.status_code, r.text)
            return r.json()
        raise FireflyError(r.status_code, r.text)

    def submit(self, prompt, n=1, size=None, seeds=None, content_class=None, negative=None):
        body = {"prompt": prompt, "numVariations": n}
        if size: body["size"] = {"width": size[0], "height": size[1]}
        if seeds: body["seeds"] = seeds
        if content_class: body["contentClass"] = content_class
        if negative: body["negativePrompt"] = negative
        return self._call("POST", f"{API_BASE}/v3/images/generate-async", json=body)

    def wait(self, job):
        t0 = time.time()
        while time.time() - t0 < self.timeout:
            d = self._call("GET", job["statusUrl"])
            if d.get("status") in ("succeeded", "failed", "cancelled"):
                return d
            time.sleep(self.poll_interval)
        raise TimeoutError(f"job {job['jobId']} 超时")

    def download(self, result, outdir, stem):
        outdir = pathlib.Path(outdir); outdir.mkdir(parents=True, exist_ok=True)
        files = []
        for i, o in enumerate(result["outputs"]):
            r = requests.get(o["image"]["url"], timeout=120)   # 预签名 URL，不要带 Authorization
            r.raise_for_status()
            p = outdir / f"{stem}_{i}_seed{o['seed']}.png"
            p.write_bytes(r.content); files.append((p, o["seed"]))
        return files


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("prompt"); ap.add_argument("-n", type=int, default=1)
    ap.add_argument("--size", default=None, help="如 2048x2048")
    ap.add_argument("--seeds", default=None, help="逗号分隔，个数须等于 -n")
    ap.add_argument("--out", default="outputs")
    a = ap.parse_args()
    ff = Firefly()
    size = tuple(map(int, a.size.split("x"))) if a.size else None
    seeds = [int(x) for x in a.seeds.split(",")] if a.seeds else None
    job = ff.submit(a.prompt, a.n, size, seeds)
    print("已提交", job["jobId"])
    d = ff.wait(job)
    if d["status"] != "succeeded":
        print("任务失败：", json.dumps(d, ensure_ascii=False)); sys.exit(1)
    alt = d["result"].get("altText", "")
    for p, seed in ff.download(d["result"], a.out, "img"):
        print(f"✓ {p}  seed={seed}")
    pathlib.Path(a.out, "alt.txt").write_text(alt + "\n")
    print("altText:", alt)


if __name__ == "__main__":
    main()

第 2 步：模拟服务器

模拟服务器只放行两种尺寸，用来演示 400 报错。真实支持的尺寸列表以官方 API 参考为准：尺寸不对时，真实接口会在 400 的报错信息里列出所有合法尺寸。

"""本地模拟服务器（自己写的，不是 Adobe 的）：模拟 IMS 令牌、generate-async、status、预签名图片。"""
import io, os, uuid, random
from flask import Flask, request, jsonify, send_file
from PIL import Image, ImageDraw

app = Flask(__name__)
BASE = os.environ.get("FAKE_BASE", "http://127.0.0.1:18491")
SIZES = {(1024, 1024), (2048, 2048)}      # 模拟器只放行这两个尺寸，演示 400
JOBS, STATS = {}, {"tokens": 0}
TOKENS = set()


@app.post("/ims/token/v3")
def token():
    f = request.form
    if f.get("client_id") != "demo-client" or f.get("client_secret") != "demo-secret":
        return jsonify(error="invalid_client", error_description="invalid client_id parameter"), 400
    t = uuid.uuid4().hex; TOKENS.add(t); STATS["tokens"] += 1
    return jsonify(access_token=t, token_type="bearer", expires_in=int(os.environ.get("FAKE_TTL", 86399)))


def authed():
    a = request.headers.get("Authorization", "")
    return request.headers.get("x-api-key") == "demo-client" and a.startswith("Bearer ") and a[7:] in TOKENS


@app.post("/v3/images/generate-async")
def gen():
    if not authed():
        return jsonify(error_code="401013", message="Oauth token is not valid"), 401
    b = request.get_json()
    n = b.get("numVariations", 1)
    if not 1 <= len(b.get("prompt", "")) <= 1024 or not 1 <= n <= 4:
        return jsonify(error_code="validation_error", message="prompt 1-1024 chars, numVariations 1-4"), 400
    if "seeds" in b and len(b["seeds"]) != n:
        return jsonify(error_code="validation_error", message="seeds count must equal numVariations"), 400
    s = b.get("size", {"width": 2048, "height": 2048}); size = (s["width"], s["height"])
    if size not in SIZES:
        return jsonify(error_code="bad_request", message=f"Size must be one of {sorted(SIZES)}"), 400
    jid = "urn:ff:jobs:fake:" + uuid.uuid4().hex[:12]
    seeds = b.get("seeds") or [random.randint(1, 2**31) for _ in range(n)]
    JOBS[jid] = dict(polls=0, prompt=b["prompt"], seeds=seeds, size=size)
    return jsonify(jobId=jid, statusUrl=f"{BASE}/v3/status/{jid}", cancelUrl=f"{BASE}/v3/cancel/{jid}"), 202


@app.get("/v3/status/<path:jid>")
def status(jid):
    if not authed():
        return jsonify(error_code="401013", message="Oauth token is not valid"), 401
    j = JOBS[jid]; j["polls"] += 1
    if j["polls"] < 3:
        return jsonify(status="running", jobId=jid)
    if "FAIL" in j["prompt"]:
        return jsonify(status="failed", jobId=jid, error_code="content_policy", message="(模拟) 提示词未通过审核")
    w, h = j["size"]
    outs = [{"seed": sd, "image": {"url": f"{BASE}/img/{jid}/{sd}/{w}x{h}.png"}} for sd in j["seeds"]]
    return jsonify(status="succeeded", jobId=jid, result={"size": {"width": w, "height": h}, "outputs": outs,
                   "contentClass": "art", "altText": "(模拟) " + j["prompt"][:60]})


@app.get("/img/<path:jid>/<int:seed>/<int:w>x<int:h>.png")
def img(jid, seed, w, h):
    rnd = random.Random(seed)
    im = Image.new("RGB", (w // 8, h // 8), tuple(rnd.randint(40, 220) for _ in range(3)))
    d = ImageDraw.Draw(im)
    for _ in range(12):
        x, y = rnd.randint(0, im.width), rnd.randint(0, im.height)
        d.ellipse([x, y, x + rnd.randint(10, 60), y + rnd.randint(10, 60)], fill=tuple(rnd.randint(0, 255) for _ in range(3)))
    d.text((6, 6), f"FAKE seed={seed}", fill="white")
    buf = io.BytesIO(); im.save(buf, "PNG"); buf.seek(0)
    return send_file(buf, mimetype="image/png")


@app.get("/_stats")
def stats():
    return jsonify(STATS)


if __name__ == "__main__":
    app.run("127.0.0.1", 18491)

第 3 步：测试

import threading, json, time, pathlib, importlib, os
import pytest
from werkzeug.serving import make_server


@pytest.fixture(scope="module")
def srv(tmp_path_factory):
    os.environ["FAKE_BASE"] = "http://127.0.0.1:18499"
    import fake_firefly; importlib.reload(fake_firefly)
    s = make_server("127.0.0.1", 18499, fake_firefly.app)
    threading.Thread(target=s.serve_forever, daemon=True).start()
    os.environ.update(FIREFLY_IMS_URL="http://127.0.0.1:18499/ims/token/v3",
                      FIREFLY_API_BASE="http://127.0.0.1:18499",
                      FIREFLY_TOKEN_CACHE=str(tmp_path_factory.mktemp("c") / "t.json"))
    import ff_client; importlib.reload(ff_client)
    yield ff_client, fake_firefly
    s.shutdown()


def test_full_flow_and_token_cache(srv, tmp_path):
    ff_client, fake = srv
    ff = ff_client.Firefly("demo-client", "demo-secret", poll_interval=0.01)
    for i in range(2):
        d = ff.wait(ff.submit("一只在写代码的猫", n=2, size=(1024, 1024), seeds=[7, 8]))
        assert d["status"] == "succeeded"
        files = ff.download(d["result"], tmp_path, f"r{i}")
        assert [s for _, s in files] == [7, 8] and files[0][0].stat().st_size > 100
    assert fake.STATS["tokens"] == 1          # 两次任务只换了一次令牌


def test_revoked_token_refreshes_once(srv):
    ff_client, fake = srv
    ff = ff_client.Firefly("demo-client", "demo-secret", poll_interval=0.01)
    fake.TOKENS.clear()                       # 模拟令牌被吊销
    before = fake.STATS["tokens"]
    assert ff.wait(ff.submit("猫"))["status"] == "succeeded"
    assert fake.STATS["tokens"] == before + 1


def test_errors(srv):
    ff_client, _ = srv
    ff = ff_client.Firefly("demo-client", "demo-secret", poll_interval=0.01)
    with pytest.raises(ff_client.FireflyError) as e:
        ff.submit("猫", n=2, seeds=[1])
    assert e.value.status == 400
    assert ff.wait(ff.submit("FAIL 测试"))["status"] == "failed"
    with pytest.raises(ff_client.FireflyError) as e:
        ff_client.Firefly("wrong", "x").token(force=True)
    assert "invalid_client" in e.value.body

$ python -m pytest -q
...                                                                      [100%]
3 passed in 0.26s

第 4 步：命令行运行（对着模拟服务器）

$ python fake_firefly.py &      # 另开终端也可以
$ export FIREFLY_IMS_URL=http://127.0.0.1:18491/ims/token/v3 FIREFLY_API_BASE=http://127.0.0.1:18491
$ export FIREFLY_SERVICES_CLIENT_ID=demo-client FIREFLY_SERVICES_CLIENT_SECRET=demo-secret
$ python ff_client.py "一只在写代码的橘猫，扁平插画" -n 2 --size 1024x1024 --seeds 101,202
已提交 urn:ff:jobs:fake:2a60f4413f96
✓ outputs/img_0_seed101.png  seed=101
✓ outputs/img_1_seed202.png  seed=202
altText: (模拟) 一只在写代码的橘猫，扁平插画

$ python ff_client.py "海边灯塔" --size 1000x1000
FireflyError: HTTP 400: {"error_code":"bad_request","message":"Size must be one of [(1024, 1024), (2048, 2048)]"}

检查点：

• 令牌缓存文件权限是 -rw-------。
• 模拟服务器的 /_stats 显示 {"tokens":1}。

切换到真实环境时，删掉 FIREFLY_IMS_URL 和 FIREFLY_API_BASE 两个变量，填上真实凭据即可。

调试示例

任务跑到一半，令牌失效了。 令牌缓存在文件里，可能被人在控制台吊销，或者被另一台机器的凭据轮换作废。这时缓存看起来还没过期，请求却返回 401。test_revoked_token_refreshes_once 复现了这个场景：清空模拟服务器里的有效令牌，再提交任务。_call() 收到第一次 401 后，会以 force=True 重新换令牌并重试一次，任务照样成功。模拟服务器上的令牌计数也正好多了 1。

重试只做一次，这是有意的：如果重新换来的令牌还是 401，说明凭据本身有问题，应该直接报错，交给 C 篇的自检脚本排查，不能无限重试。

常见问题

模板规则

• 令牌要缓存，并且提前续期，不要每次请求都去换令牌。
• 只用异步接口，轮询时设置总超时。
• 任务一成功就下载预签名图片，seed 和 altText 一起保存。

项目资源

有了这个客户端，后面的批量生成和接入自检都可以直接复用。