快速入門

const BASE = 'https://api.spicyapi.ai/api/v1';
const KEY = process.env.SPICY_API_KEY;

const headers = {
  Authorization: `Bearer ${KEY}`,
  'Content-Type': 'application/json',
};
async function call(path, init) {
  const res = await fetch(`${BASE}${path}`, {
    ...init,
    headers: { ...headers, ...init?.headers },
    signal: AbortSignal.timeout(30_000),
  });
  const body = await res.json();
  // 判斷成功看 code，不要只看 res.ok —— 業務碼比 HTTP 狀態碼的表達力強得多
  if (body.code !== 200) {
    throw new Error(`${body.code} ${body.msg} (request_id=${body.request_id})`);
  }
  return body.data;
}

async function generate() {
  const catalog = await call(
    '/models?modality=image&task=text-to-image&includeSchema=1&includeExamples=1',
  );
  const selected = catalog.items.find(
    (item) => item.enabled && item.available && item.examples?.length,
  );
  if (!selected) throw new Error('No callable text-to-image model with a validated example');

  const idempotencyKey = crypto.randomUUID();
  const { taskId } = await call('/jobs/createTask', {
    method: 'POST',
    headers: { 'Idempotency-Key': idempotencyKey },
    body: JSON.stringify({
      model: selected.model,
      input: selected.examples[0].input,
    }),
  });

  // 2 秒起步，每次乘 1.5，封頂 15 秒。第一次查詢幾乎必然還是 queued，
  // 起步給短一點只是浪費一次往返。
  const deadline = Date.now() + 10 * 60_000;
  let wait = 2000;
  while (Date.now() < deadline) {
    await new Promise((r) => setTimeout(r, wait));
    wait = Math.min(wait * 1.5, 15000);

    const task = await call(`/jobs/recordInfo?taskId=${taskId}`);
    if (task.state === 'succeeded') {
      const asset = task.output?.assets?.find((item) => item.url);
      if (asset) return asset.url;
      if (task.output?.assets?.some((item) => item.pending)) continue;
      throw new Error('Result is unavailable');
    }
    // 除 queued / running 外均按終態處理，可相容歷史 canceled 與未來新增終態。
    if (task.state !== 'queued' && task.state !== 'running') {
      throw new Error(`${task.state}: ${task.errorMessage ?? '無更多資訊'}`);
    }
  }
  throw new Error(`任務 ${taskId} 等待逾時`);
}

generate().then(console.log);

import os, time, uuid, requests

BASE = "https://api.spicyapi.ai/api/v1"
HEADERS = {
    "Authorization": f"Bearer {os.environ['SPICY_API_KEY']}",
    "Content-Type": "application/json",
}


def call(method: str, path: str, headers=None, **kwargs) -> dict:
    res = requests.request(
        method, f"{BASE}{path}", headers={**HEADERS, **(headers or {})}, timeout=30, **kwargs
    )
    body = res.json()
    # 判斷成功看 code，不要只看 status_code
    if body["code"] != 200:
        raise RuntimeError(f'{body["code"]} {body["msg"]} (request_id={body.get("request_id")})')
    return body["data"]


def generate() -> str:
    catalog = call(
        "GET",
        "/models?modality=image&task=text-to-image&includeSchema=1&includeExamples=1",
    )
    selected = next(
        (
            item
            for item in catalog["items"]
            if item["enabled"] and item["available"] and item.get("examples")
        ),
        None,
    )
    if selected is None:
        raise RuntimeError("No callable text-to-image model with a validated example")

    idempotency_key = str(uuid.uuid4())
    task_id = call("POST", "/jobs/createTask", headers={"Idempotency-Key": idempotency_key}, json={
        "model": selected["model"],
        "input": selected["examples"][0]["input"],
    })["taskId"]

    # 2 秒起步，每次乘 1.5，封頂 15 秒
    deadline = time.monotonic() + 600
    wait = 2.0
    while time.monotonic() < deadline:
        time.sleep(wait)
        wait = min(wait * 1.5, 15.0)

        task = call("GET", "/jobs/recordInfo", params={"taskId": task_id})
        if task["state"] == "succeeded":
            assets = task.get("output", {}).get("assets", [])
            ready = next((asset for asset in assets if asset.get("url")), None)
            if ready:
                return ready["url"]
            if any(asset.get("pending") for asset in assets):
                continue
            raise RuntimeError("Result is unavailable")
        # 除 queued / running 外均按終態處理。
        if task["state"] not in {"queued", "running"}:
            raise RuntimeError(f'{task["state"]}: {task.get("errorMessage", "無更多資訊")}')
    raise TimeoutError(f"任務 {task_id} 等待逾時")


print(generate())

package main

import (
	"bytes"
	"crypto/rand"
	"encoding/json"
	"encoding/hex"
	"fmt"
	"math"
	"net/http"
	"os"
	"time"
)

const base = "https://api.spicyapi.ai/api/v1"

// envelope 是全站統一的回應外殼。Data 留成 RawMessage，
// 由各介面自己解成需要的形狀。
type envelope struct {
	Code      int             `json:"code"`
	Msg       string          `json:"msg"`
	Data      json.RawMessage `json:"data"`
	RequestID string          `json:"request_id"`
}

var client = &http.Client{Timeout: 30 * time.Second}

func newIdempotencyKey() string {
	b := make([]byte, 16)
	if _, err := rand.Read(b); err != nil {
		panic(err)
	}
	return "job-" + hex.EncodeToString(b)
}

func call(method, path string, body any, idempotencyKey string, out any) error {
	var rdr *bytes.Reader
	if body != nil {
		raw, err := json.Marshal(body)
		if err != nil {
			return err
		}
		rdr = bytes.NewReader(raw)
	} else {
		rdr = bytes.NewReader(nil)
	}

	req, err := http.NewRequest(method, base+path, rdr)
	if err != nil {
		return err
	}
	req.Header.Set("Authorization", "Bearer "+os.Getenv("SPICY_API_KEY"))
	req.Header.Set("Content-Type", "application/json")
	if idempotencyKey != "" {
		req.Header.Set("Idempotency-Key", idempotencyKey)
	}

	res, err := client.Do(req)
	if err != nil {
		return err
	}
	defer res.Body.Close()

	var env envelope
	if err := json.NewDecoder(res.Body).Decode(&env); err != nil {
		return err
	}
	// 判斷成功看 code，不要只看 res.StatusCode
	if env.Code != 200 {
		return fmt.Errorf("%d %s (request_id=%s)", env.Code, env.Msg, env.RequestID)
	}
	return json.Unmarshal(env.Data, out)
}

func generate() (string, error) {
	var catalog struct {
		Items []struct {
			Model     string `json:"model"`
			Enabled   bool   `json:"enabled"`
			Available bool   `json:"available"`
			Examples  []struct {
				Input map[string]any `json:"input"`
			} `json:"examples"`
		} `json:"items"`
	}
	if err := call(http.MethodGet,
		"/models?modality=image&task=text-to-image&includeSchema=1&includeExamples=1",
		nil, "", &catalog); err != nil {
		return "", err
	}
	var modelID string
	var modelInput map[string]any
	for _, item := range catalog.Items {
		if item.Enabled && item.Available && len(item.Examples) > 0 {
			modelID, modelInput = item.Model, item.Examples[0].Input
			break
		}
	}
	if modelID == "" {
		return "", fmt.Errorf("no callable text-to-image model with a validated example")
	}

	var created struct {
		TaskID string `json:"taskId"`
	}
	err := call(http.MethodPost, "/jobs/createTask", map[string]any{
		"model": modelID,
		"input": modelInput,
	}, newIdempotencyKey(), &created)
	if err != nil {
		return "", err
	}

	// 2 秒起步，每次乘 1.5，封頂 15 秒
	wait := 2 * time.Second
	deadline := time.Now().Add(10 * time.Minute)
	for time.Now().Before(deadline) {
		time.Sleep(wait)
		wait = time.Duration(math.Min(float64(wait)*1.5, float64(15*time.Second)))

		var task struct {
			State    string `json:"state"`
			ErrorMsg string `json:"errorMessage"`
			Output struct { Assets []struct { URL string `json:"url"`; Pending bool `json:"pending"` } `json:"assets"` } `json:"output"`
		}
		if err := call(http.MethodGet, "/jobs/recordInfo?taskId="+created.TaskID, nil, "", &task); err != nil {
			return "", err
		}

		if task.State == "succeeded" {
			pending := false
			for _, asset := range task.Output.Assets {
				if asset.URL != "" { return asset.URL, nil }
				pending = pending || asset.Pending
			}
			if pending { continue }
			return "", fmt.Errorf("result is unavailable")
		}
		// 判終態：不在這兩個中間態裡就是終態，不要逐個列舉終態
		if task.State != "queued" && task.State != "running" {
			return "", fmt.Errorf("%s: %s", task.State, task.ErrorMsg)
		}
	}
	return "", fmt.Errorf("任務 %s 等待逾時", created.TaskID)
}

func main() {
	url, err := generate()
	if err != nil {
		panic(err)
	}
	fmt.Println(url)
}

{ "code": 200, "msg": "success", "data": { }, "request_id": "req_…" }