异步生图 API 详细教程

export GPT88_API_KEY="YOUR_GPT88_API_KEY"
export BASE_URL="https://img.gpt88.cc"

curl -sS -X POST "$BASE_URL/v1/images/generations" \
  -H "Authorization: Bearer $GPT88_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-image-2",
    "prompt": "高级电商护肤品主图，玻璃瓶身，白色石材台面，柔和晨光，干净高级构图，无文字，无水印",
    "size": "1024x1024",
    "quality": "high",
    "n": 1,
    "async": true
  }' | tee create.json

# 兼容顶层 task_id / id 和嵌套 data.task_id / data.id。
TASK_ID=$(jq -r '.task_id // .id // .data.task_id // .data.id // empty' create.json)
test -n "$TASK_ID" || { echo "没有返回 task ID："; cat create.json; exit 1; }
echo "created task: $TASK_ID"

MAX_ATTEMPTS=60
INTERVAL_SECONDS=5

for attempt in $(seq 1 "$MAX_ATTEMPTS"); do
  curl -sS "$BASE_URL/v1/images/generations/$TASK_ID" \
    -H "Authorization: Bearer $GPT88_API_KEY" > status.json

  STATUS=$(jq -r '.data.status // .status // .task.status // empty' status.json | tr '[:upper:]' '[:lower:]')
  echo "attempt=$attempt status=$STATUS"

  case "$STATUS" in
    succeeded|success|completed)
      RESULT_URL=$(jq -r '.data.result_url // .result_url // .data.result.data[0].url // .result.data[0].url // .data[0].url // empty' status.json)
      if [ -n "$RESULT_URL" ]; then
        curl -L "$RESULT_URL" -o async-image.png
        echo "已保存 async-image.png"
        exit 0
      fi
      echo "任务成功但没有找到 URL；请检查 status.json 中是否返回 b64_json 或模型专属结果。" >&2
      exit 2
      ;;
    failed|failure|cancelled|canceled)
      echo "异步生图失败：" >&2
      cat status.json >&2
      exit 3
      ;;
  esac

  sleep "$INTERVAL_SECONDS"
done

echo "轮询超时。保留 TASK_ID=$TASK_ID，稍后继续查询。" >&2
exit 4

RESULT_URL=$(jq -r '.data.result_url // .result_url // .data.result.data[0].url // .result.data[0].url // .data[0].url // empty' status.json)
if [ -n "$RESULT_URL" ]; then
  curl -L "$RESULT_URL" -o async-image.png
else
  B64=$(jq -r '.data.result.data[0].b64_json // .data[0].b64_json // empty' status.json)
  test -n "$B64" || { echo "没有找到 URL 或 base64 图片" >&2; exit 1; }
  printf '%s' "$B64" | base64 -d > async-image.png
fi
file async-image.png

const BASE_URL = "https://img.gpt88.cc";
const API_KEY = process.env.GPT88_API_KEY;

const sleep = (ms) => new Promise(resolve => setTimeout(resolve, ms));
const first = (...values) => values.find(value => value !== undefined && value !== null && value !== "");

function taskIdOf(payload) {
  return first(payload.task_id, payload.id, payload.data?.task_id, payload.data?.id);
}

function statusOf(payload) {
  return String(first(payload.data?.status, payload.status, payload.task?.status, "")).toLowerCase();
}

function resultOf(payload) {
  return first(
    payload.data?.result_url,
    payload.result_url,
    payload.data?.result?.data?.[0]?.url,
    payload.result?.data?.[0]?.url,
    payload.data?.[0]?.url,
    payload.data?.result?.data?.[0]?.b64_json,
    payload.data?.[0]?.b64_json,
  );
}

async function generateAsyncImage(prompt) {
  const createResponse = await fetch(BASE_URL + "/v1/images/generations", {
    method: "POST",
    headers: {
      Authorization: "Bearer " + API_KEY,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      model: "gpt-image-2",
      prompt,
      size: "1024x1024",
      quality: "high",
      n: 1,
      async: true,
    }),
  });

  const created = await createResponse.json();
  if (!createResponse.ok) throw new Error("提交失败：" + JSON.stringify(created));

  const taskId = taskIdOf(created);
  if (!taskId) throw new Error("没有返回 task ID：" + JSON.stringify(created));

  for (let attempt = 1; attempt <= 60; attempt += 1) {
    await sleep(5000);
    const pollResponse = await fetch(BASE_URL + "/v1/images/generations/" + encodeURIComponent(taskId), {
      headers: { Authorization: "Bearer " + API_KEY },
    });
    const payload = await pollResponse.json();
    if (!pollResponse.ok) throw new Error("轮询失败：" + JSON.stringify(payload));

    const status = statusOf(payload);
    if (["succeeded", "success", "completed"].includes(status)) {
      const result = resultOf(payload);
      if (!result) throw new Error("任务成功，但没有 URL 或 base64 结果");
      return { taskId, result, raw: payload };
    }
    if (["failed", "failure", "cancelled", "canceled"].includes(status)) {
      throw new Error("异步生图失败：" + JSON.stringify(payload));
    }
  }

  throw new Error("轮询超时，请保留 task ID 后继续查询：" + taskId);
}

const output = await generateAsyncImage("一盏银色台灯的极简编辑风产品图，暖侧光，无文字");
console.log(output);

import base64
import os
import time
from pathlib import Path

import requests

BASE_URL = "https://img.gpt88.cc"
API_KEY = os.environ["GPT88_API_KEY"]
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

def first(*values):
    return next((value for value in values if value not in (None, "")), None)

def task_id_of(payload):
    data = payload.get("data") or {}
    return first(payload.get("task_id"), payload.get("id"), data.get("task_id"), data.get("id"))

def status_of(payload):
    data = payload.get("data") or {}
    task = payload.get("task") or {}
    return str(first(data.get("status"), payload.get("status"), task.get("status"), "")).lower()

def result_of(payload):
    data = payload.get("data") or {}
    result = data.get("result") or payload.get("result") or {}
    result_data = result.get("data") or data.get("data") or []
    first_item = result_data[0] if result_data else (data[0] if isinstance(data, list) and data else {})
    return first(data.get("result_url"), payload.get("result_url"), first_item.get("url"), first_item.get("b64_json"))

create = requests.post(
    BASE_URL + "/v1/images/generations",
    headers={**HEADERS, "Content-Type": "application/json"},
    json={
        "model": "gpt-image-2",
        "prompt": "高级电商护肤品主图，玻璃瓶身，柔和晨光，无文字",
        "size": "1024x1024",
        "quality": "high",
        "n": 1,
        "async": True,
    },
    timeout=30,
)
create.raise_for_status()
created = create.json()
task_id = task_id_of(created)
if not task_id:
    raise RuntimeError(f"没有返回 task ID：{created}")

for _ in range(60):
    time.sleep(5)
    response = requests.get(
        BASE_URL + "/v1/images/generations/" + task_id,
        headers=HEADERS,
        timeout=30,
    )
    response.raise_for_status()
    payload = response.json()
    status = status_of(payload)
    if status in {"succeeded", "success", "completed"}:
        result = result_of(payload)
        if not result:
            raise RuntimeError(f"任务成功但没有结果：{payload}")
        if result.startswith("http"):
            image = requests.get(result, timeout=60)
            image.raise_for_status()
            Path("async-image.png").write_bytes(image.content)
        else:
            Path("async-image.png").write_bytes(base64.b64decode(result))
        break
    if status in {"failed", "failure", "cancelled", "canceled"}:
        raise RuntimeError(f"异步生图失败：{payload}")
else:
    raise TimeoutError(f"轮询超时，请保留 task ID：{task_id}")

{
  "task_id": "imgtask_123",
  "model": "gpt-image-2",
  "prompt_hash": "sha256:...",
  "status": "processing",
  "submitted_at": "2026-08-05T10:00:00Z",
  "last_polled_at": "2026-08-05T10:00:15Z",
  "attempts": 3,
  "result_url": null,
  "error": null
}