跳转到主要内容

​
路径参数

​
id
string
必填
从创建视频请求返回的任务 ID。将 id 和 task_id 视为相同的异步标识。
如果创建响应返回 poll_url，请调用该精确 URL。当它指向 /v1/tasks/{id} 时，将其视为规范的固定状态端点。如果该任务已不存在或无法再从公开任务记录中查询到，TokenLab 会返回 async_task_not_found，并附带消息 Task not found or no longer available.

​
响应

​
id
string
规范的异步任务标识符。
​
task_id
string
异步任务标识符别名。
​
poll_url
string
当创建响应提供时的首选轮询 URL。
​
billing_transaction_id
string
当结算已经完成时返回 TokenLab 账单交易 ID。它对应 dashboard / 对账使用的交易标识，与异步 id / task_id 不同。
​
status
string
任务状态：pending、processing、completed、failed。
​
progress
number
可选进度值。仅在任务提供真实进度时返回；请使用 status 判断任务是否完成。
​
video_url
string
生成视频的 URL（完成时）。
​
video
object
单个视频负载，在可用时包含 url、duration、width 和 height。
​
videos
array
当生成任务返回多个输出时的多个视频负载。
​
error
string
错误信息（如果失败）。
​
created
integer
创建时间戳。
​
updated
integer
最后更新时间戳。
​
model
string
用于该任务的模型。
curl "https://api.tokenlab.sh/v1/tasks/ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" \
  -H "Authorization: Bearer sk-your-api-key"

import requests
import time

task_id = "ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"

while True:
    response = requests.get(
        f"https://api.tokenlab.sh/v1/tasks/{task_id}",
        headers={"Authorization": "Bearer sk-your-api-key"}
    )

    data = response.json()
    print(f"Status: {data['status']}")

    if data["status"] == "completed":
        print(f"Video URL: {data['video_url']}")
        break
    elif data["status"] == "failed":
        print(f"Error: {data['error']}")
        break

    time.sleep(5)  # Poll every 5 seconds

const taskId = 'ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa';

async function pollForVideo() {
  while (true) {
    const response = await fetch(
      `https://api.tokenlab.sh/v1/tasks/${taskId}`,
      { headers: { 'Authorization': 'Bearer sk-your-api-key' } }
    );

    const data = await response.json();
    console.log(`Status: ${data.status}`);

    if (data.status === 'completed') {
      console.log(`Video URL: ${data.video_url}`);
      return data.video_url;
    } else if (data.status === 'failed') {
      throw new Error(data.error);
    }

    await new Promise(r => setTimeout(r, 5000));
  }
}

package main

import (
    "encoding/json"
    "fmt"
    "net/http"
    "time"
)

func main() {
    taskID := "ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"

    for {
        req, _ := http.NewRequest("GET",
            fmt.Sprintf("https://api.tokenlab.sh/v1/tasks/%s", taskID), nil)
        req.Header.Set("Authorization", "Bearer sk-your-api-key")

        client := &http.Client{}
        resp, _ := client.Do(req)

        var result map[string]interface{}
        json.NewDecoder(resp.Body).Decode(&result)
        resp.Body.Close()

        fmt.Printf("Status: %s\n", result["status"])

        if result["status"] == "completed" {
            fmt.Printf("Video URL: %s\n", result["video_url"])
            break
        } else if result["status"] == "failed" {
            fmt.Printf("Error: %s\n", result["error"])
            break
        }

        time.Sleep(5 * time.Second)
    }
}

<?php
$taskId = 'ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa';

while (true) {
    $ch = curl_init("https://api.tokenlab.sh/v1/tasks/{$taskId}");

    curl_setopt_array($ch, [
        CURLOPT_RETURNTRANSFER => true,
        CURLOPT_HTTPHEADER => [
            'Authorization: Bearer sk-your-api-key'
        ]
    ]);

    $response = curl_exec($ch);
    curl_close($ch);

    $data = json_decode($response, true);
    echo "Status: {$data['status']}\n";

    if ($data['status'] === 'completed') {
        echo "Video URL: {$data['video_url']}\n";
        break;
    } elseif ($data['status'] === 'failed') {
        echo "Error: {$data['error']}\n";
        break;
    }

    sleep(5);
}

{
  "id": "ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "task_id": "ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "poll_url": "/v1/tasks/ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "status": "pending",
  "model": "veo3.1",
  "created": 1706000000,
  "updated": 1706000000
}

{
  "id": "ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "task_id": "ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "poll_url": "/v1/tasks/ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "status": "processing",
  "model": "veo3.1",
  "created": 1706000000,
  "updated": 1706000030
}

{
  "id": "ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "task_id": "ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "poll_url": "/v1/tasks/ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "billing_transaction_id": "cmo164zyc0s48jpgegi80vqv4",
  "status": "completed",
  "video_url": "https://assets.tokenlab.sh/videos/abc123.mp4",
  "model": "veo3.1",
  "created": 1706000000,
  "updated": 1706000060
}

{
  "id": "ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "task_id": "ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "poll_url": "/v1/tasks/ldtask_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "status": "failed",
  "error": "Content policy violation",
  "model": "veo3.1",
  "created": 1706000000,
  "updated": 1706000060
}

​
轮询最佳实践

  • 每 5-10 秒轮询一次
  • 为长期任务实现指数退避
  • 设置最大超时（例如 10 分钟）
  • 优雅地处理 failed 状态
import time

def wait_for_video(task_id, max_wait=600, interval=5):
    """Wait for video with timeout."""
    start = time.time()

    while time.time() - start < max_wait:
        response = requests.get(
            f"https://api.tokenlab.sh/v1/tasks/{task_id}",
            headers={"Authorization": "Bearer sk-your-api-key"}
        )
        data = response.json()

        if data["status"] == "completed":
            return data["video_url"]
        elif data["status"] == "failed":
            raise Exception(data.get("error", "Video generation failed"))

        time.sleep(interval)

    raise TimeoutError("Video generation timed out")