Webhooks

// 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 会收到结果

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
  }'

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 会收到结果

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 会收到结果
}

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"))
  );
}

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