通过 HTTP 回调接收任务结果，无需轮询。
使用 callback 模式时，任务完成后 APIXO 会向指定 URL 发送 POST 请求。可消除轮询开销并提供即时通知，适合高并发的生产应用。
配置步骤
创建 webhook 接口
你的接口必须接受 POST 请求，在 30 秒内返回 HTTP 200，并通过 HTTPS 可公开访问。若接口未在 30 秒内返回 HTTP 200，APIXO 将最多重试 3 次，每次间隔递增。
Express.js
FastAPI
Next.js
const express = require('express');
const app = express();
app.use(express.json());
app.post('/webhook/apixo', (req, res) => {
const { taskId, state, resultJson, failCode, failMsg } = req.body.data;
if (state === 'success') {
const urls = JSON.parse(resultJson).resultUrls;
console.log('Generated:', urls);
} else if (state === 'failed') {
console.error(`Task ${taskId} failed: ${failCode} - ${failMsg}`);
}
res.status(200).send('OK');
});
app.listen(3000);
from fastapi import FastAPI, Request
import json
app = FastAPI()
@app.post('/webhook/apixo')
async def handle_webhook(request: Request):
body = await request.json()
data = body['data']
if data['state'] == 'success':
urls = json.loads(data['resultJson'])['resultUrls']
print(f"Generated: {urls}")
elif data['state'] == 'failed':
print(f"Task {data['taskId']} failed: {data['failCode']} - {data['failMsg']}")
return {'status': 'ok'}
// app/api/webhook/apixo/route.ts
import { NextRequest, NextResponse } from 'next/server';
export async function POST(request: NextRequest) {
const body = await request.json();
const { taskId, state, resultJson, failCode, failMsg } = body.data;
if (state === 'success') {
const urls = JSON.parse(resultJson).resultUrls;
console.log('Generated:', urls);
} else if (state === 'failed') {
console.error(`Task ${taskId} failed: ${failCode} - ${failMsg}`);
}
return NextResponse.json({ status: 'ok' });
}
提交带回调的任务
请求中需包含 request_type: "callback" 和你的 callback_url：curl -X POST https://api.apixo.ai/api/v1/generateTask/nano-banana \
-H "Authorization: Bearer YOUR_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"request_type": "callback",
"callback_url": "https://your-domain.com/webhook/apixo",
"input": {
"mode": "text-to-image",
"prompt": "A beautiful landscape"
}
}'
const response = await fetch('https://api.apixo.ai/api/v1/generateTask/nano-banana', {
method: 'POST',
headers: {
'Authorization': 'Bearer YOUR_API_KEY',
'Content-Type': 'application/json',
},
body: JSON.stringify({
request_type: 'callback',
callback_url: 'https://your-domain.com/webhook/apixo',
input: {
mode: 'text-to-image',
prompt: 'A beautiful landscape',
},
}),
});
response = requests.post(
'https://api.apixo.ai/api/v1/generateTask/nano-banana',
headers={
'Authorization': 'Bearer YOUR_API_KEY',
'Content-Type': 'application/json',
},
json={
'request_type': 'callback',
'callback_url': 'https://your-domain.com/webhook/apixo',
'input': {
'mode': 'text-to-image',
'prompt': 'A beautiful landscape',
},
}
)
Webhook 请求体
{
"code": 200,
"message": "success",
"data": {
"taskId": "task_abc123xyz",
"state": "success",
"resultJson": "{\"resultUrls\":[\"https://cdn.apixo.ai/output/abc.jpg\"]}",
"costTime": 12500,
"createTime": 1704067200000,
"completeTime": 1704067212500
}
}
{
"code": 200,
"message": "success",
"data": {
"taskId": "task_abc123xyz",
"state": "failed",
"failCode": "CONTENT_VIOLATION",
"failMsg": "Content violates usage policy",
"createTime": 1704067200000,
"completeTime": 1704067205000
}
}
字段说明
包含 resultUrls 数组的 JSON 字符串。仅在成功时存在。
重试策略
若 webhook 接口未返回 HTTP 200：
| Attempt | Delay |
|---|
| 1st retry | 30 seconds |
| 2nd retry | 2 minutes |
| 3rd retry | 10 minutes |
安全建议
实现幂等性
Webhook 可能会多次投递。使用 taskId 去重，避免同一结果被处理多次。
const processedTasks = new Set();
app.post('/webhook/apixo', (req, res) => {
const { taskId } = req.body.data;
if (processedTasks.has(taskId)) {
return res.status(200).send('Already processed');
}
processedTasks.add(taskId);
// Process the task...
res.status(200).send('OK');
});
仅使用 HTTPS
webhook 接口务必使用 HTTPS，确保传输加密。
测试 Webhooks
开发时可用 ngrok 暴露本地服务：
ngrok http 3000
# 返回: https://abc123.ngrok.io — 将其作为 callback_url 使用
使用 callback 模式时，仍可通过 API 查询任务状态，便于调试。