异步任务模型

{
  "success": true,
  "task_id": "27a1128f-14b5-440e-a82d-510c81e6fcd4"
}

import time, requests

API = "https://api.acedata.cloud"
headers = {"Authorization": "Bearer YOUR_API_TOKEN"}

# 1. 创建任务
r = requests.post(
    f"{API}/midjourney/imagine",
    headers=headers,
    json={"prompt": "A cat sitting on a table"},
).json()
task_id = r["task_id"]

# 2. 轮询
while True:
    info = requests.post(
        f"{API}/midjourney/tasks",
        headers=headers,
        json={"action": "retrieve", "task_id": task_id},
    ).json()

    # finished_at 字段一旦出现，任务即完成
    if info.get("finished_at"):
        print(info["response"])
        break

    time.sleep(5)

{
  "prompt": "A cat sitting on a table",
  "callback_url": "https://your-server.com/midjourney/callback"
}

{
  "success": true,
  "task_id": "...",
  "trace_id": "...",
  "image_url": "...",
  "...": "...（其余业务字段，与同步成功响应一致）"
}

{
  "success": false,
  "task_id": "...",
  "trace_id": "...",
  "error": {
    "code": "...",
    "message": "..."
  }
}

from flask import Flask, request

app = Flask(__name__)

# 自己维护的「我提交过的 task_id」集合
my_tasks = set()

@app.post("/midjourney/callback")
def callback():
    payload = request.json
    task_id = payload.get("task_id")

    if task_id not in my_tasks:
        return "", 204  # 不是我的任务，忽略

    if payload.get("success"):
        # 处理成功
        save_image(payload["image_url"])
    else:
        # 处理失败
        log_error(payload.get("error"))

    return "", 200