任务管理
提交任务
POST /v1/tasks — 提交异步生成任务
POST
/
v1
/
tasks
提交任务
curl --request POST \
--url https://www.qingbo.dev/v1/tasks \
--header 'Content-Type: application/json' \
--data '
{
"model": "<string>",
"action": "<string>",
"prompt": "<string>",
"callback_url": "<string>",
"extra": {}
}
'import requests
url = "https://www.qingbo.dev/v1/tasks"
payload = {
"model": "<string>",
"action": "<string>",
"prompt": "<string>",
"callback_url": "<string>",
"extra": {}
}
headers = {"Content-Type": "application/json"}
response = requests.post(url, json=payload, headers=headers)
print(response.text)const options = {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({
model: '<string>',
action: '<string>',
prompt: '<string>',
callback_url: '<string>',
extra: {}
})
};
fetch('https://www.qingbo.dev/v1/tasks', options)
.then(res => res.json())
.then(res => console.log(res))
.catch(err => console.error(err));package main
import (
"fmt"
"strings"
"net/http"
"io"
)
func main() {
url := "https://www.qingbo.dev/v1/tasks"
payload := strings.NewReader("{\n \"model\": \"<string>\",\n \"action\": \"<string>\",\n \"prompt\": \"<string>\",\n \"callback_url\": \"<string>\",\n \"extra\": {}\n}")
req, _ := http.NewRequest("POST", url, payload)
req.Header.Add("Content-Type", "application/json")
res, _ := http.DefaultClient.Do(req)
defer res.Body.Close()
body, _ := io.ReadAll(res.Body)
fmt.Println(string(body))
}HttpResponse<String> response = Unirest.post("https://www.qingbo.dev/v1/tasks")
.header("Content-Type", "application/json")
.body("{\n \"model\": \"<string>\",\n \"action\": \"<string>\",\n \"prompt\": \"<string>\",\n \"callback_url\": \"<string>\",\n \"extra\": {}\n}")
.asString();{
"pre_consumed_cost": 123
}提交图像、视频或音频生成任务,立即返回
更多参数请参考 图像生成概述 或 视频生成概述。
适合图像生成等较快完成的任务、原型开发和测试。视频等长耗时任务建议轮询或 Webhook。
task_id。
请求参数
string
必填
模型名称,如
"veo3", "gpt-4o-image", "sora2"string
默认值:"generate"
操作类型:
generate, edit, upscale, variation, blendstring
生成描述文本
string
任务完成时的 Webhook 回调地址
object
模型特定参数的透传字段,直接传递给上游模型
响应
{
"task_id": "task_01K8SGYNNNVBQTXNR4MM964S7K",
"model": "veo3",
"action": "generate",
"status": "queued",
"created_at": 1720000000,
"progress": 0,
"pre_consumed_cost": 5500
}
integer
提交时预扣的额度(quota)。任务完成后以查询接口返回的
cost 为准结算:
成功 = 实际扣费;失败/取消 = 自动全额退款(cost: 0)。请求内等待(Prefer: wait)
不想自己轮询时,加Prefer: wait=N 请求头(N ≤ 60 秒)即可在一次调用里等待结果:
- 窗口内完成 → 直接返回终态完整结果(含
result与cost) - 超窗未完成 → 返回当前状态 +
pre_consumed_cost,继续用GET /v1/tasks/{task_id}轮询
curl -X POST https://www.qingbo.dev/v1/tasks \
-H "Authorization: Bearer YOUR_API_KEY" \
-H "Prefer: wait=60" \
-H "Content-Type: application/json" \
-d '{"model": "gpt-4o-image", "prompt": "A tiny blue circle"}'
幂等提交(Idempotency-Key)
带Idempotency-Key: <唯一字符串> 请求头提交时,24 小时内同一 key 的重复请求会
直接复用首次创建的任务(同一 task_id,不重复扣费)。适合网络重试场景。
示例
curl -X POST https://www.qingbo.dev/v1/tasks \
-H "Authorization: Bearer YOUR_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"model": "veo3",
"action": "generate",
"prompt": "A cat walking on the beach at sunset",
"duration": 5
}'
import requests
response = requests.post(
"https://www.qingbo.dev/v1/tasks",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"model": "veo3",
"action": "generate",
"prompt": "A cat walking on the beach at sunset",
"duration": 5
}
)
task_id = response.json()["task_id"]
⌘I
提交任务
curl --request POST \
--url https://www.qingbo.dev/v1/tasks \
--header 'Content-Type: application/json' \
--data '
{
"model": "<string>",
"action": "<string>",
"prompt": "<string>",
"callback_url": "<string>",
"extra": {}
}
'import requests
url = "https://www.qingbo.dev/v1/tasks"
payload = {
"model": "<string>",
"action": "<string>",
"prompt": "<string>",
"callback_url": "<string>",
"extra": {}
}
headers = {"Content-Type": "application/json"}
response = requests.post(url, json=payload, headers=headers)
print(response.text)const options = {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({
model: '<string>',
action: '<string>',
prompt: '<string>',
callback_url: '<string>',
extra: {}
})
};
fetch('https://www.qingbo.dev/v1/tasks', options)
.then(res => res.json())
.then(res => console.log(res))
.catch(err => console.error(err));package main
import (
"fmt"
"strings"
"net/http"
"io"
)
func main() {
url := "https://www.qingbo.dev/v1/tasks"
payload := strings.NewReader("{\n \"model\": \"<string>\",\n \"action\": \"<string>\",\n \"prompt\": \"<string>\",\n \"callback_url\": \"<string>\",\n \"extra\": {}\n}")
req, _ := http.NewRequest("POST", url, payload)
req.Header.Add("Content-Type", "application/json")
res, _ := http.DefaultClient.Do(req)
defer res.Body.Close()
body, _ := io.ReadAll(res.Body)
fmt.Println(string(body))
}HttpResponse<String> response = Unirest.post("https://www.qingbo.dev/v1/tasks")
.header("Content-Type", "application/json")
.body("{\n \"model\": \"<string>\",\n \"action\": \"<string>\",\n \"prompt\": \"<string>\",\n \"callback_url\": \"<string>\",\n \"extra\": {}\n}")
.asString();{
"pre_consumed_cost": 123
}