异步任务

统一异步任务接口使用说明，适用于视频生成、图片生成、音乐生成等耗时任务。

统一异步任务接口提供标准化的「提交 → 轮询 → 获取结果」流程，屏蔽不同厂商 API 差异。

概述

视频生成、图片生成、音乐生成等 AI 任务通常需要几十秒甚至数分钟才能完成。为保障用户体验与系统稳定性，这些任务统一通过异步任务接口处理：

提交任务 — 发送生成请求，立即获得任务 ID
查询状态 — 使用任务 ID 轮询进度，或通过回调地址接收通知
获取结果 — 任务完成后从响应中获取生成的资源 URL

提交任务

POST /v1/task/submit

请求参数

字段	类型	必填	说明
`model`	string	是	模型名称
`input`	object	是	模型所需的生成参数（因模型而异）
`callback_url`	string	否	任务状态变更时的回调地址

请求示例

cURLPythonNode.js

bash

curl -X POST https://api.calgine.com.cn/v1/task/submit \
  -H "Authorization: Bearer sk-your-token" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "kling-v3/text-to-video",
    "input": {
      "prompt": "一只金色的柴犬在樱花树下奔跑，慢镜头，电影质感",
      "duration": "5",
      "aspect_ratio": "16:9"
    },
    "callback_url": "https://your-server.com/webhook/task"
  }'

python

import requests

response = requests.post(
    "https://api.calgine.com.cn/v1/task/submit",
    headers={
        "Authorization": "Bearer sk-your-token",
        "Content-Type": "application/json",
    },
    json={
        "model": "kling-v3/text-to-video",
        "input": {
            "prompt": "一只金色的柴犬在樱花树下奔跑，慢镜头，电影质感",
            "duration": "5",
            "aspect_ratio": "16:9",
        },
        "callback_url": "https://your-server.com/webhook/task",
    },
)

data = response.json()
print(data)
# {"id": "task_abc123", "status": "queued", "created_at": 1711234567}

const response = await fetch("https://api.calgine.com.cn/v1/task/submit", {
  method: "POST",
  headers: {
    Authorization: "Bearer sk-your-token",
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    model: "kling-v3/text-to-video",
    input: {
      prompt: "一只金色的柴犬在樱花树下奔跑，慢镜头，电影质感",
      duration: "5",
      aspect_ratio: "16:9",
    },
    callback_url: "https://your-server.com/webhook/task",
  }),
});

const data = await response.json();
console.log(data);
// {"id": "task_abc123", "status": "queued", "created_at": 1711234567}

响应示例

json

{
  "id": "task_abc123",
  "status": "queued",
  "created_at": 1711234567
}

查询任务

GET /v1/task/{task_id}

请求示例

cURLPythonNode.js

bash

curl https://api.calgine.com.cn/v1/task/task_abc123 \
  -H "Authorization: Bearer sk-your-token"

python

import requests

response = requests.get(
    "https://api.calgine.com.cn/v1/task/task_abc123",
    headers={"Authorization": "Bearer sk-your-token"},
)

data = response.json()
print(data["status"])   # "queued" | "in_progress" | "completed" | "failed"
print(data["outputs"])  # 完成后包含资源 URL 列表

const response = await fetch("https://api.calgine.com.cn/v1/task/task_abc123", {
  headers: { Authorization: "Bearer sk-your-token" },
});

const data = await response.json();
console.log(data.status);  // "queued" | "in_progress" | "completed" | "failed"
console.log(data.outputs); // 完成后包含资源 URL 列表

响应示例

任务进行中：

json

{
  "id": "task_abc123",
  "status": "in_progress",
  "created_at": 1711234567
}

任务完成（按 token 计费）：

json

{
  "id": "task_abc123",
  "status": "completed",
  "created_at": 1711234567,
  "completed_at": 1711234620,
  "outputs": [
    "https://cdn.example.com/video/result.mp4"
  ],
  "usage": {
    "type": "tokens",
    "input_tokens": 14,
    "input_token_details": {
      "audio_tokens": 14
    },
    "output_tokens": 45,
    "total_tokens": 59
  }
}

任务完成（按时长计费）：

json

{
  "id": "task_xyz789",
  "status": "completed",
  "created_at": 1711234567,
  "completed_at": 1711234620,
  "outputs": [
    "https://cdn.example.com/video/result.mp4"
  ],
  "usage": {
    "type": "duration",
    "seconds": 27
  }
}

任务失败：

json

{
  "id": "task_abc123",
  "status": "failed",
  "created_at": 1711234567,
  "completed_at": 1711234590,
  "error": "Content policy violation detected"
}

任务状态

状态	说明
`queued`	任务已提交，等待处理
`in_progress`	任务正在生成中
`completed`	任务完成，`outputs` 中包含结果
`failed`	任务失败，`error` 中包含原因

回调通知

提交任务时传入 callback_url，当任务状态发生变更时，系统会向该地址发送 POST 请求，请求体与查询接口响应格式一致：

json

{
  "id": "task_abc123",
  "status": "completed",
  "created_at": 1711234567,
  "completed_at": 1711234620,
  "outputs": [
    "https://cdn.example.com/video/result.mp4"
  ],
  "usage": {
    "type": "tokens",
    "input_tokens": 14,
    "output_tokens": 45,
    "total_tokens": 59
  }
}

回调地址必须是公网可访问的 HTTPS 地址
回调请求超时时间为 10 秒
回调地址不能指向 localhost、127.0.0.1 或平台自身域名
即使设置了回调，仍建议实现轮询作为兜底，避免因网络原因错过通知

完整轮询示例

PythonNode.js

python

import time
import requests

API_KEY = "sk-your-token"
BASE_URL = "https://api.calgine.com.cn/v1"

# 1. 提交任务
submit_resp = requests.post(
    f"{BASE_URL}/task/submit",
    headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    },
    json={
        "model": "kling-v3/text-to-video",
        "input": {
            "prompt": "一只金色的柴犬在樱花树下奔跑",
            "duration": "5",
            "aspect_ratio": "16:9",
        },
    },
)
task = submit_resp.json()
task_id = task["id"]
print(f"任务已提交: {task_id}")

# 2. 轮询等待完成
while True:
    resp = requests.get(
        f"{BASE_URL}/task/{task_id}",
        headers={"Authorization": f"Bearer {API_KEY}"},
    )
    result = resp.json()
    status = result["status"]
    print(f"状态: {status}")

    if status == "completed":
        print("生成结果:", result["outputs"])
        break
    elif status == "failed":
        print("任务失败:", result.get("error", "Unknown error"))
        break

    time.sleep(5)

const API_KEY = "sk-your-token";
const BASE_URL = "https://api.calgine.com.cn/v1";

// 1. 提交任务
const submitResp = await fetch(`${BASE_URL}/task/submit`, {
  method: "POST",
  headers: {
    Authorization: `Bearer ${API_KEY}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    model: "kling-v3/text-to-video",
    input: {
      prompt: "一只金色的柴犬在樱花树下奔跑",
      duration: "5",
      aspect_ratio: "16:9",
    },
  }),
});

const task = await submitResp.json();
const taskId = task.id;
console.log(`任务已提交: ${taskId}`);

// 2. 轮询等待完成
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

while (true) {
  const resp = await fetch(`${BASE_URL}/task/${taskId}`, {
    headers: { Authorization: `Bearer ${API_KEY}` },
  });
  const result = await resp.json();
  console.log(`状态: ${result.status}`);

  if (result.status === "completed") {
    console.log("生成结果:", result.outputs);
    break;
  } else if (result.status === "failed") {
    console.log("任务失败:", result.error);
    break;
  }

  await sleep(5000);
}

响应字段说明

字段	类型	说明
`id`	string	任务唯一标识
`status`	string	任务状态：`queued` / `in_progress` / `completed` / `failed`
`created_at`	integer	任务创建时间（Unix 时间戳，秒）
`completed_at`	integer	任务完成时间（Unix 时间戳，秒），仅在完成/失败时返回
`outputs`	string[]	生成结果的 URL 列表（视频/图片/音频），仅在 `completed` 时返回
`usage`	object	任务用量信息，详见下方「用量信息」章节，仅在终态且模型支持时返回
`error`	string	错误信息，仅在 `failed` 时返回

用量信息（Usage）

任务进入终态（completed / failed）后，部分模型会在响应中返回 usage 字段，反映本次任务的实际资源消耗。usage 是一个标签联合（tagged union）：通过 type 字段区分计量单位。

type=tokens（按 token 计费）

适用于 Doubao / Vidu / Ali / Kling 文本类、Gemini 等按 token 计费的任务：

json

{
  "type": "tokens",
  "input_tokens": 14,
  "input_token_details": {
    "audio_tokens": 14
  },
  "output_tokens": 45,
  "total_tokens": 59
}

字段	类型	说明
`type`	string	固定为 `"tokens"`
`input_tokens`	integer	输入 token 总数
`input_token_details`	object	按模态拆分的输入 token，可选；其内部子字段为 0 时省略，全部为 0 时该对象整体省略
`input_token_details.text_tokens`	integer	文本输入 token
`input_token_details.audio_tokens`	integer	音频输入 token
`input_token_details.image_tokens`	integer	图片输入 token
`input_token_details.cached_tokens`	integer	命中提示词缓存的 token（兼容 OpenAI prompt cache）
`output_tokens`	integer	输出 token 总数
`output_token_details`	object	按模态拆分的输出 token，可选；省略规则同 `input_token_details`
`total_tokens`	integer	`input_tokens + output_tokens` 总计

type=duration（按时长计费）

适用于 Sora / Xai-video 等按生成视频时长计费的任务：

json

{
  "type": "duration",
  "seconds": 27
}

字段	类型	说明
`type`	string	固定为 `"duration"`
`seconds`	number	任务用量时长（秒，支持小数）

TIP

纯按调用次数计费的任务（如 Flux / Replicate / Midjourney）不返回 usage 字段
任务在 queued / in_progress 状态时不会返回 usage
历史任务或上游未返回用量数据的任务也会省略此字段
字段值为 0 的子项（如 text_tokens: 0）会被省略

Flux 模型的完成通知

想在任务结束时收到通知，请在请求体顶层传 callback_url，所有 Flux 模型都支持。
input 里的 webhook_url / webhook_secret 不会发给上游。flux-3-image、flux-3-video、flux-tools/video-edit-v1、flux-tools/outpainting-v1 的官方接口不接受回调字段，带了也会被忽略，请用 callback_url 或轮询。
不传回调时，提交后用 GET /v1/task/{task_id} 轮询（建议间隔 3~10 秒；视频任务通常需要几分钟）。

bash

curl -X POST "https://api.calgine.com.cn/v1/task/submit" \
  -H "Authorization: Bearer sk-your-token" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "flux-tools/outpainting-v1",
    "callback_url": "https://your-server.com/hook",
    "input": {
      "images": ["https://example.com/base.png"],
      "width": 1200,
      "height": 800
    }
  }'

# 轮询任务状态
curl "https://api.calgine.com.cn/v1/task/{task_id}" \
  -H "Authorization: Bearer sk-your-token"

注意事项

使用标准 Authorization: Bearer <API_KEY> 鉴权
input 中的参数因模型而异，请在 Calgine Tokens 控制台的模型列表中点击对应模型查看详情文档
建议轮询间隔 3~10 秒，避免过于频繁的请求
任务结果中的资源 URL 可能有有效期限制，建议及时下载保存
Flux 模型要接收完成通知请用顶层 callback_url，见上文「Flux 模型的完成通知」

异步任务 ​

概述 ​

提交任务 ​

请求参数 ​

请求示例 ​

响应示例 ​

查询任务 ​

请求示例 ​

响应示例 ​

任务状态 ​

回调通知 ​

完整轮询示例 ​

响应字段说明 ​

用量信息（Usage） ​

type=tokens（按 token 计费） ​

type=duration（按时长计费） ​

Flux 模型的完成通知 ​

注意事项 ​

异步任务

概述

提交任务

请求参数

请求示例

响应示例

查询任务

请求示例

响应示例

任务状态

回调通知

完整轮询示例

响应字段说明

用量信息（Usage）

type=tokens（按 token 计费）

type=duration（按时长计费）

Flux 模型的完成通知

注意事项