> ## Documentation Index
> Fetch the complete documentation index at: https://www.krea.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Webhooks

> 在 Krea API 任务完成时接收带签名的 POST 请求，验证其 Ed25519 签名，并处理重试。

## 概述

与其轮询 `GET /jobs/{id}`，不如向 Krea 提供一个 webhook URL。任务结束时，Krea 会将任务发送到该 URL。

## 使用 Webhooks

在任意生成请求中添加 `X-Webhook-URL` 请求头。当任务完成、失败或被取消时，Krea 会向该 URL 发送 POST 请求。

```javascript Node.js theme={"theme":{"light":"github-light-default","dark":"vesper"}}
// npm install @krea-ai/sdk
import { Krea } from "@krea-ai/sdk";

const krea = new Krea({ apiKey: process.env.KREA_API_KEY });

const job = await krea.image(
  "bfl/flux-1-dev",
  {
    prompt: "a serene mountain landscape at sunset",
    width: 1024,
    height: 576
  },
  { webhookUrl: "https://your-server.com/webhook" }
);

console.log(`Job ID: ${job.job_id}`);
// 任务完成时，你的 webhook 会收到结果
```

```bash cURL theme={"theme":{"light":"github-light-default","dark":"vesper"}}
curl -X POST https://api.krea.ai/generate/image/bfl/flux-1-dev \
  -H "Authorization: Bearer $KREA_API_KEY" \
  -H "Content-Type: application/json" \
  -H "X-Webhook-URL: https://your-server.com/webhook" \
  -d '{
    "prompt": "a serene mountain landscape at sunset",
    "width": 1024,
    "height": 576
  }'
```

```python Python theme={"theme":{"light":"github-light-default","dark":"vesper"}}
import os

import requests

API_BASE = "https://api.krea.ai"
API_TOKEN = os.environ["KREA_API_KEY"]

response = requests.post(
    f"{API_BASE}/generate/image/bfl/flux-1-dev",
    headers={
        "Authorization": f"Bearer {API_TOKEN}",
        "Content-Type": "application/json",
        "X-Webhook-URL": "https://your-server.com/webhook"
    },
    json={
        "prompt": "a serene mountain landscape at sunset",
        "width": 1024,
        "height": 576
    }
)

job = response.json()
print(f"Job ID: {job['job_id']}")
# 任务完成时，你的 webhook 会收到结果
```

```go Go theme={"theme":{"light":"github-light-default","dark":"vesper"}}
package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "net/http"
    "os"
)

func main() {
    apiBase := "https://api.krea.ai"
    apiToken := os.Getenv("KREA_API_KEY")

    payload := map[string]interface{}{
        "prompt": "a serene mountain landscape at sunset",
        "width":  1024,
        "height": 576,
    }

    jsonData, _ := json.Marshal(payload)
    req, _ := http.NewRequest("POST", apiBase+"/generate/image/bfl/flux-1-dev", bytes.NewBuffer(jsonData))
    req.Header.Set("Authorization", "Bearer "+apiToken)
    req.Header.Set("Content-Type", "application/json")
    req.Header.Set("X-Webhook-URL", "https://your-server.com/webhook")

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

    var job map[string]interface{}
    json.NewDecoder(resp.Body).Decode(&job)
    fmt.Printf("Job ID: %s\n", job["job_id"])
    // 任务完成时，你的 webhook 会收到结果
}
```

## Webhook 载荷

请求体的结构与 `GET /jobs/{id}` 的响应相同：

```json theme={"theme":{"light":"github-light-default","dark":"vesper"}}
{
  "created_at": "2025-01-15T10:30:00.000Z",
  "completed_at": "2025-01-15T10:30:05.000Z",
  "job_id": "550e8400-e29b-41d4-a716-446655440000",
  "status": "completed",
  "type": "flux",
  "result": {
    "urls": ["https://..."]
  },
  "progress": null
}
```

## 验证签名

Krea 使用 Ed25519 为每个 webhook 签名。在信任载荷之前请先验证签名，并拒绝没有签名的请求。

| 请求头 | 值 |
| - | - |
| `X-Krea-Webhook-Timestamp` | Krea 发送请求的时间，以 Unix 秒表示 |
| `X-Krea-Webhook-Url` | 你随任务发送的 `X-Webhook-URL` |
| `X-Krea-Webhook-Metadata` | 保留字段，目前为空 |
| `X-Krea-Webhook-Signature` | 十六进制编码的 Ed25519 签名 |

Krea 将时间戳、URL、元数据和原始请求体用换行符连接，计算其 SHA-256 并以小写十六进制表示，然后对该值签名：

```text theme={"theme":{"light":"github-light-default","dark":"vesper"}}
hex(sha256(timestamp + "\n" + url + "\n" + metadata + "\n" + raw_body))
```

验证请求的步骤：

1. 检查 `X-Krea-Webhook-Url` 是否与你随任务发送的 `X-Webhook-URL` 完全一致。
2. 检查时间戳与服务器当前时间的差值是否在 5 分钟以内。
3. 如果签名能通过 `https://api.krea.ai/.well-known/jwks.json` 中任一密钥的验证，则接受该请求。密钥最多缓存 24 小时。Krea 会在开始使用新密钥签名的 24 小时前发布该密钥。

请对原始请求体字节进行哈希，而不是重新序列化后的 JSON。例如，在 Express 中使用 `express.raw()`，在基于 Fetch API 的框架中使用 `await request.arrayBuffer()`，在 Flask 中使用 `request.get_data()`。如果缺少 `X-Krea-Webhook-Metadata` 请求头，请将其视为空字符串。

```javascript Node.js theme={"theme":{"light":"github-light-default","dark":"vesper"}}
import { createHash, createPublicKey, verify } from "node:crypto";

// 你随任务发送的 X-Webhook-URL
const WEBHOOK_URL = "https://your-server.com/webhook";

let publicKeys = [];
let publicKeysFetchedAt = 0;

// headers 是 Fetch API 的 Headers 对象，rawBody 是 Buffer
export async function verifyKreaWebhook(headers, rawBody) {
  const timestamp = headers.get("X-Krea-Webhook-Timestamp");
  const url = headers.get("X-Krea-Webhook-Url");
  const signature = headers.get("X-Krea-Webhook-Signature") ?? "";
  const metadata = headers.get("X-Krea-Webhook-Metadata") ?? "";

  if (url !== WEBHOOK_URL) return false;
  if (!(Math.abs(Date.now() / 1000 - Number(timestamp)) <= 300)) return false;

  if (Date.now() - publicKeysFetchedAt > 60 * 60 * 1000) {
    const { keys } = await (await fetch("https://api.krea.ai/.well-known/jwks.json")).json();
    publicKeys = keys.map((jwk) => createPublicKey({ key: jwk, format: "jwk" }));
    publicKeysFetchedAt = Date.now();
  }

  const message = createHash("sha256")
    .update(`${timestamp}\n${url}\n${metadata}\n`)
    .update(rawBody)
    .digest("hex");

  return publicKeys.some((key) =>
    verify(null, Buffer.from(message), key, Buffer.from(signature, "hex"))
  );
}
```

```python Python theme={"theme":{"light":"github-light-default","dark":"vesper"}}
import base64
import hashlib
import time

import requests
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

# 你随任务发送的 X-Webhook-URL
WEBHOOK_URL = "https://your-server.com/webhook"

_public_keys = []
_public_keys_fetched_at = 0.0


# headers 是不区分大小写的映射，例如 Flask 的 request.headers
def verify_krea_webhook(headers, raw_body: bytes) -> bool:
    global _public_keys, _public_keys_fetched_at
    timestamp = headers.get("X-Krea-Webhook-Timestamp", "")
    url = headers.get("X-Krea-Webhook-Url")
    metadata = headers.get("X-Krea-Webhook-Metadata", "")

    try:
        if url != WEBHOOK_URL or abs(time.time() - int(timestamp)) > 300:
            return False
        signature = bytes.fromhex(headers.get("X-Krea-Webhook-Signature", ""))
    except ValueError:
        return False

    if time.time() - _public_keys_fetched_at > 3600:
        jwks = requests.get("https://api.krea.ai/.well-known/jwks.json", timeout=10).json()
        _public_keys = [
            Ed25519PublicKey.from_public_bytes(base64.urlsafe_b64decode(jwk["x"] + "="))
            for jwk in jwks["keys"]
        ]
        _public_keys_fetched_at = time.time()

    signed = f"{timestamp}\n{url}\n{metadata}\n".encode() + raw_body
    message = hashlib.sha256(signed).hexdigest().encode()
    for key in _public_keys:
        try:
            key.verify(signature, message)
            return True
        except InvalidSignature:
            pass

    return False
```

## 投递与重试

请在 15 秒内返回 `2xx`。如果请求超时、无法连接，或收到 `5xx`、`408`、`423`、`425` 或 `429`，Krea 会在大约 10 分钟内持续重试，每次都使用新的时间戳和签名。对于其他 `4xx`，Krea 不会重试。

同一任务可能会多次送达，因此请根据 `job_id` 和 `status` 去重。

## Webhooks 与轮询

| 方式 | 优点 | 缺点 |
| - | - | - |
| **Webhooks** | 任务结束后立即收到结果，API 调用更少 | 需要公开可访问的端点 |
| **轮询** | 任何地方都能用，无需服务器 | API 调用更多，结果最多会延迟一个轮询间隔 |

如果你有可以接收 HTTP 请求的服务器，建议使用 webhooks。对于客户端应用或无法公开端点的场景，则使用轮询。

## 下一步

> [card] [任务生命周期](https://www.krea.ai/docs/cn/developers/job-lifecycle.md)
>
> 了解任务状态和状态轮询

> [card] [代码示例](https://www.krea.ai/docs/cn/developers/examples/text-to-image.md)
>
> 使用 API 从头到尾生成一张图像


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.