# Recipes

Each recipe is a single runnable script that solves one real integration problem end to end. Read them here, or download the whole kit:

**[Download the integration kit](https://docs.genomicintelligence.ai/integration-kit.tar.gz)** (`.tar.gz`) — every file below plus the docs, bundled.

## recipes/01_health_probe.py

The minimum check before any other work: API reachable, key valid, per-task model lists loadable. Exits 0 on green.

[Download raw](https://docs.genomicintelligence.ai/recipes/01_health_probe.py)

```python title="recipes/01_health_probe.py"
"""Recipe 01 — health probe.

The minimum check before any other work: is the API reachable, is your
key valid, are the per-task model lists loadable. Exits 0 on green, 1
on any failure.

    GI_API_KEY=gi_… python3 01_health_probe.py
"""

from __future__ import annotations

import os
import sys

import requests

BASE_URL = os.environ.get("GI_BASE_URL", "https://api.genomicintelligence.ai")
API_KEY = os.environ.get("GI_API_KEY")

def fail(msg: str) -> None:
    print(f"FAIL: {msg}", file=sys.stderr)
    sys.exit(1)

def main() -> int:
    if not API_KEY:
        fail("set GI_API_KEY")

    # 1. Public liveness — no auth required.
    r = requests.get(f"{BASE_URL}/health", timeout=10)
    if not r.ok:
        fail(f"/health returned {r.status_code}: {r.text[:200]}")
    health = r.json()
    print(f"health   : {health.get('status')} {health.get('version')}")

    # 2. Authenticated round-trip — proves the bearer key is registered.
    r = requests.get(
        f"{BASE_URL}/v1/tasks/promoter/models",
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=10,
    )
    if r.status_code == 401:
        fail("401 unauthorized — key missing, malformed, or unrecognised. "
             f"Check Authorization header. Body: {r.text[:200]}")
    if not r.ok:
        body = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
        err = body.get("error") or {}
        fail(f"{r.status_code} {err.get('code', 'unknown')} request_id={err.get('request_id')}")
    # Note: model-listing endpoints (`/v1/tasks/{task}/models`,
    # `/v1/tasks/{task}/models/{id}/status`) return flat objects, NOT the
    # `{data, meta}` envelope used by predict endpoints. Live OpenAPI at
    # /v1/openapi.json is the source of truth for shapes per route.
    body = r.json()
    models = body.get("models", [])
    print(f"models   : {len(models)} registered for task=promoter "
          f"(default={body.get('default_model')})")

    # 3. Read RateLimit headers — confirms your tier and your headroom.
    print(f"ratelimit: limit={r.headers.get('RateLimit-Limit')} "
          f"remaining={r.headers.get('RateLimit-Remaining')} "
          f"reset={r.headers.get('RateLimit-Reset')}s "
          f"policy={r.headers.get('RateLimit-Policy')}")

    print("ok")
    return 0

if __name__ == "__main__":
    raise SystemExit(main())
```

## recipes/02_promoters_for_gene_list.py

Realistic batch shape: fetch each gene from Ensembl (public, no auth) and run promoter prediction, one BED-style row per detected region.

[Download raw](https://docs.genomicintelligence.ai/recipes/02_promoters_for_gene_list.py)

```python title="recipes/02_promoters_for_gene_list.py"
"""Recipe 02 — promoter prediction across a list of human gene symbols.

Realistic batch shape: you have N gene symbols, you want promoters for
each. We fetch the genomic sequence for each gene from Ensembl REST
(public, no auth) and run `POST /v1/tasks/promoter/predict` against
each. Output is one BED-style row per detected promoter region.

Demonstrates:
  - sequence acquisition from a public bioinformatics source (Ensembl)
  - serial pacing against a per-key concurrency cap
  - extraction of the typed `data.regions` field per the contract

    GI_API_KEY=gi_… python3 02_promoters_for_gene_list.py TP53 MYC GAPDH
"""

from __future__ import annotations

import os
import sys
import time
from typing import Iterable

import requests

BASE_URL = os.environ.get("GI_BASE_URL", "https://api.genomicintelligence.ai")
API_KEY = os.environ.get("GI_API_KEY")
ENSEMBL = "https://rest.ensembl.org"

# Default genes to demo if the user gives no args. Sized for sub-2s per call.
DEFAULT_GENES = ["TP53", "MYC", "GAPDH"]

def fetch_gene_sequence(symbol: str, species: str = "human") -> tuple[str, str]:
    """Resolve `symbol` to an Ensembl ID and fetch its genomic sequence.

    Returns (display_name, sequence). Raises on any HTTP failure.

    Fetches on the gene's native strand so that minus-strand genes
    (e.g. TP53) are returned in 5'→3' coding orientation — the
    orientation the promoter model was trained on.
    """
    # Lookup Ensembl ID + strand
    r = requests.get(
        f"{ENSEMBL}/lookup/symbol/{species}/{symbol}",
        headers={"Accept": "application/json"},
        timeout=15,
    )
    r.raise_for_status()
    info = r.json()
    chrom, start, end = info["seq_region_name"], info["start"], info["end"]
    strand = info.get("strand", 1)

    # Fetch genomic sequence on the gene's strand
    region = f"{chrom}:{start}..{end}:{strand}"
    r = requests.get(
        f"{ENSEMBL}/sequence/region/{species}/{region}",
        headers={"Accept": "text/plain"},
        timeout=30,
    )
    r.raise_for_status()
    return f"{symbol}|chr{chrom}:{start}-{end}|strand:{strand}", r.text.strip().upper()

def predict_promoter(sequence: str, sequence_name: str) -> tuple[dict, dict]:
    """One sync call to /v1/tasks/promoter/predict.

    Returns ``(body, headers)`` — the parsed ``{data, meta}`` body plus the
    response headers, so the caller can pace off the ``RateLimit-*`` spine.
    """
    r = requests.post(
        f"{BASE_URL}/v1/tasks/promoter/predict",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        json={"sequence": sequence, "sequence_name": sequence_name,
              "options": {"threshold": 0.5}},
        timeout=120,
    )
    if not r.ok:
        body = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
        err = body.get("error") or {}
        raise RuntimeError(
            f"{r.status_code} {err.get('code', 'unknown')}: "
            f"{err.get('message', r.text[:200])} (request_id={err.get('request_id')})"
        )
    return r.json(), r.headers

def pace(headers) -> float:
    """Return how many seconds to sleep before the next call to stay at ~80% of capacity.

    Reads IETF RateLimit-* headers. If headroom is low, slow down; if
    plenty, no wait. Conservative pacing — beats reactive 429 retries.
    """
    try:
        remaining = int(headers.get("RateLimit-Remaining", "999"))
        limit = int(headers.get("RateLimit-Limit", "10"))
        reset = float(headers.get("RateLimit-Reset", "0"))
    except (TypeError, ValueError):
        return 0.0
    threshold = max(1, int(limit * 0.2))  # aim to stay above 20% headroom
    if remaining <= threshold and reset > 0:
        # Spread remaining work across the reset window
        return reset / max(1, remaining)
    return 0.0

def run(symbols: Iterable[str]) -> int:
    if not API_KEY:
        print("ERROR: set GI_API_KEY", file=sys.stderr)
        return 2

    print("# track  start  end  name  score")
    last_headers = {}
    for symbol in symbols:
        wait = pace(last_headers)
        if wait > 0:
            print(f"# pacing: sleep {wait:.2f}s", file=sys.stderr)
            time.sleep(wait)

        try:
            display, sequence = fetch_gene_sequence(symbol)
        except requests.HTTPError as exc:
            print(f"# {symbol}: ensembl lookup failed — {exc}", file=sys.stderr)
            continue

        try:
            body, last_headers = predict_promoter(sequence, display)
        except RuntimeError as exc:
            print(f"# {symbol}: predict failed — {exc}", file=sys.stderr)
            continue

        # `last_headers` now carries this call's RateLimit-* state, so the
        # pace() at the top of the next iteration slows down before we run low.
        regions = body.get("data", {}).get("regions") or []
        for r in regions:
            print(f"{symbol}\t{r['start']}\t{r['end']}\t{r.get('name', '.')}\t{r['score']:.4f}")
        if not regions:
            print(f"# {symbol}: 0 regions above threshold", file=sys.stderr)

    return 0

if __name__ == "__main__":
    args = sys.argv[1:] or DEFAULT_GENES
    raise SystemExit(run(args))
```

## recipes/03_async_annotation_polling.py

Async dispatch with progress polling for large inputs — HTTP status discriminates progress (`202`) from terminal (`200`/`4xx`/`5xx`).

[Download raw](https://docs.genomicintelligence.ai/recipes/03_async_annotation_polling.py)

```python title="recipes/03_async_annotation_polling.py"
"""Recipe 03 — async annotation with progress polling.

Annotation is the slowest atomic task. For inputs above ~30 kbp, opt
into async to avoid the 300 s upstream proxy timeout. The polling shape
is uniform: HTTP status discriminates progress (`202`) vs terminal
(`200` success, `4xx`/`5xx` failure). Body is always `{data, meta}`
while running and on success; only terminal failure switches to the
error envelope.

Demonstrates:
  - `Prefer: respond-async` opt-in
  - 202-progress polling shape
  - exponential backoff on the poll interval (don't hammer)
  - terminal-state discrimination

    GI_API_KEY=gi_… python3 03_async_annotation_polling.py path/to/sequence.fa
"""

from __future__ import annotations

import os
import sys
import time

import requests

BASE_URL = os.environ.get("GI_BASE_URL", "https://api.genomicintelligence.ai")
API_KEY = os.environ.get("GI_API_KEY")
HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}

def load_fasta(path: str) -> tuple[str, str]:
    text = open(path).read().splitlines()
    name = text[0].lstrip(">").strip()
    sequence = "".join(line.strip() for line in text[1:] if line.strip()).upper()
    return name, sequence

def submit_async(sequence: str, sequence_name: str) -> str:
    """POST with `Prefer: respond-async`; return the job_id.

    Annotation accepts 1,000–500,000 bp (whitespace-stripped); shorter is
    `422 validation_failed`. `options` is closed — annotation accepts only
    `shift_coordinates` and `reverse_complement`, and
    a model is chosen with the TOP-LEVEL `model` field, never inside
    `options`.
    """
    r = requests.post(
        f"{BASE_URL}/v1/tasks/annotation/predict",
        headers={**HEADERS, "Prefer": "respond-async"},
        json={"sequence": sequence, "sequence_name": sequence_name},
        timeout=30,
    )
    r.raise_for_status()
    return r.json()["data"]["job_id"]

def poll_until_terminal(job_id: str, max_wait_s: float = 30 * 60) -> dict:
    """Poll `/v1/tasks/jobs/{id}` until 200 or terminal 4xx/5xx.

    Returns the final body. Raises on terminal failure or timeout.
    """
    deadline = time.monotonic() + max_wait_s
    interval = 2.0  # seconds; doubled up to a cap on each 202
    while True:
        r = requests.get(
            f"{BASE_URL}/v1/tasks/jobs/{job_id}",
            headers={"Authorization": f"Bearer {API_KEY}"},
            timeout=30,
        )
        if r.status_code == 200:
            return r.json()  # done — {data, meta}
        if r.status_code == 202:
            # In flight. Body is also {data, meta}; data.progress carries
            # current_percent / message / elapsed_seconds.
            body = r.json()
            progress = (body.get("data") or {}).get("progress") or {}
            pct = progress.get("current_percent")
            msg = progress.get("message", "")
            print(f"  [{pct:>3}%] {msg}", file=sys.stderr)

            if time.monotonic() > deadline:
                raise TimeoutError(f"job {job_id} did not terminate within {max_wait_s}s")
            time.sleep(interval)
            interval = min(interval * 1.5, 15.0)  # back off, cap at 15s
            continue

        # Terminal failure — unified error envelope.
        body = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
        err = body.get("error") or {}
        raise RuntimeError(
            f"job {job_id} terminal: {r.status_code} {err.get('code', 'unknown')} "
            f"{err.get('message', '')} (request_id={err.get('request_id')})"
        )

def main() -> int:
    if not API_KEY:
        print("ERROR: set GI_API_KEY", file=sys.stderr); return 2
    if len(sys.argv) < 2:
        print("usage: 03_async_annotation_polling.py <fasta-path>", file=sys.stderr); return 2

    name, sequence = load_fasta(sys.argv[1])
    print(f"submitting {len(sequence):,} bp as async annotation job…", file=sys.stderr)
    job_id = submit_async(sequence, name)
    print(f"job_id={job_id}", file=sys.stderr)

    body = poll_until_terminal(job_id)
    transcripts = body.get("data", {}).get("transcripts") or []
    meta = body.get("meta", {})
    print(f"done — transcripts={len(transcripts)} "
          f"inference_ms={meta.get('inference_time_ms')} "
          f"counts={meta.get('task_specific_counts')}")
    for t in transcripts:
        print(f"  {t.get('name', '.')}\t{t.get('start')}\t{t.get('end')}\t"
              f"{t.get('strand')}\t{t.get('score'):.4f}")
    return 0

if __name__ == "__main__":
    raise SystemExit(main())
```

## recipes/04_ratelimit_aware_retry.py

A `call()` wrapper that paces itself off the IETF `RateLimit-*` headers, honours `Retry-After` on 429, and backs off on 5xx.

[Download raw](https://docs.genomicintelligence.ai/recipes/04_ratelimit_aware_retry.py)

```python title="recipes/04_ratelimit_aware_retry.py"
"""Recipe 04 — rate-limit-aware request wrapper.

A `call()` helper that:
  - reads IETF `RateLimit-*` headers from every response and paces the
    next call to stay at ~80% of capacity
  - on `429`, honours `Retry-After`; on `5xx`, exponential backoff
    capped at 30 s
  - never retries permanent 4xx errors (catalogue:
    https://docs.genomicintelligence.ai/reference/errors.md)
  - returns the parsed `{data, meta}` body on success or raises a
    typed exception on permanent failure

Use this as a drop-in for `requests.post(...)` when you need pacing
without managing it by hand.

    GI_API_KEY=gi_… python3 04_ratelimit_aware_retry.py
"""

from __future__ import annotations

import os
import time
from dataclasses import dataclass, field
from typing import Any, Optional

import requests

BASE_URL = os.environ.get("GI_BASE_URL", "https://api.genomicintelligence.ai")
API_KEY = os.environ.get("GI_API_KEY")

@dataclass
class GIError(Exception):
    status: int
    code: str
    message: str
    request_id: Optional[str] = None
    details: Any = None

    def __str__(self) -> str:
        return f"[{self.status} {self.code}] {self.message} (request_id={self.request_id})"

@dataclass
class RateLimitState:
    """Sticky pacing state between calls. Threadsafe? No — per-worker."""
    limit: int = 0
    remaining: int = 0
    reset_seconds: float = 0.0
    last_response_at: float = field(default_factory=time.monotonic)

PERMANENT_4XX = {
    "bad_request", "unauthorized", "forbidden", "not_found", "model_not_found",
    "validation_failed", "task_not_supported_by_model", "payload_too_large",
    "sync_too_large", "unsupported_format", "conflict", "job_expired",
}
RETRYABLE = {
    "too_many_requests",   # 429: rate or concurrency cap
    "rate_limited",        # 429: same treatment
    "model_loading",       # 503: warmup race
    "service_unavailable", # 503: startup/shutdown
    "timeout",             # inference deadline — resubmit async
    "insufficient_memory", # GPU pressure — retry, or shorten the sequence
}

def _wait_before_next(state: RateLimitState) -> float:
    """Pacing: stay above 20% headroom of the token bucket."""
    if state.limit <= 0:
        return 0.0
    threshold = max(1, int(state.limit * 0.2))
    if state.remaining > threshold:
        return 0.0
    elapsed = time.monotonic() - state.last_response_at
    remaining_window = max(0.0, state.reset_seconds - elapsed)
    if remaining_window <= 0 or state.remaining <= 0:
        return 0.0
    return remaining_window / max(1, state.remaining)

def _read_headers(state: RateLimitState, headers) -> None:
    try:
        state.limit = int(headers.get("RateLimit-Limit", state.limit))
        state.remaining = int(headers.get("RateLimit-Remaining", state.remaining))
        state.reset_seconds = float(headers.get("RateLimit-Reset", state.reset_seconds))
    except (TypeError, ValueError):
        pass
    state.last_response_at = time.monotonic()

def call(
    method: str,
    path: str,
    *,
    json_body: Optional[dict] = None,
    headers: Optional[dict] = None,
    state: RateLimitState,
    max_attempts: int = 5,
) -> dict:
    """Issue one logical request with retries on retryable errors."""
    url = f"{BASE_URL}{path}"
    base_headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
    if headers:
        base_headers.update(headers)

    backoff = 1.0
    for attempt in range(1, max_attempts + 1):
        wait = _wait_before_next(state)
        if wait > 0:
            time.sleep(wait)

        r = requests.request(method, url, json=json_body, headers=base_headers, timeout=120)
        _read_headers(state, r.headers)

        if r.ok:
            return r.json()

        body = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
        err = (body or {}).get("error") or {}
        code = err.get("code", "unknown")

        if code in PERMANENT_4XX:
            # Don't retry; surface to caller.
            raise GIError(r.status_code, code, err.get("message", ""),
                          err.get("request_id"), err.get("details"))

        if code in RETRYABLE and attempt < max_attempts:
            retry_after = r.headers.get("Retry-After")
            sleep = float(retry_after) if retry_after else min(backoff, 30.0)
            time.sleep(sleep)
            backoff *= 2
            continue

        # 5xx server bug — one retry then give up.
        if r.status_code >= 500 and attempt < max_attempts:
            time.sleep(min(backoff, 30.0))
            backoff *= 2
            continue

        raise GIError(r.status_code, code, err.get("message", ""),
                      err.get("request_id"), err.get("details"))

    raise GIError(0, "exhausted", f"max_attempts={max_attempts} reached", None, None)

def demo() -> int:
    if not API_KEY:
        print("ERROR: set GI_API_KEY"); return 2

    state = RateLimitState()
    # A burst of small calls — the wrapper paces automatically.
    for i in range(15):
        try:
            body = call(
                "POST", "/v1/tasks/promoter/predict",
                json_body={"sequence": "ACGT" * 500, "sequence_name": f"demo-{i}",
                           "options": {"threshold": 0.5}},
                state=state,
            )
            ms = body.get("meta", {}).get("inference_time_ms")
            print(f"call {i:>2}: {ms} ms  (limit={state.limit} remaining={state.remaining})")
        except GIError as exc:
            print(f"call {i:>2}: FAILED — {exc}")
    return 0

if __name__ == "__main__":
    raise SystemExit(demo())
```

## recipes/05_typed_error_handling.py

Switch on `error.code`, never on `error.message`. Maps every error code to retryable / permanent / server-bug plus the action to take.

[Download raw](https://docs.genomicintelligence.ai/recipes/05_typed_error_handling.py)

```python title="recipes/05_typed_error_handling.py"
"""Recipe 05 — typed error handling.

Switch on `error.code`, never on `error.message`. Every code from
https://docs.genomicintelligence.ai/reference/errors.md is mapped here
to one of four buckets — retryable, permanent, server-bug, or
proxy-timeout (the enveloped-less 504 special case) — plus the action
your code should take.

Use this as the template for your own error handler. Copy `handle()`
into your codebase, register it as your one place to translate API
errors to your domain's exception types.

    GI_API_KEY=gi_… python3 05_typed_error_handling.py
"""

from __future__ import annotations

import os
import sys
from dataclasses import dataclass
from typing import Any, Optional

import requests

BASE_URL = os.environ.get("GI_BASE_URL", "https://api.genomicintelligence.ai")
API_KEY = os.environ.get("GI_API_KEY")

@dataclass
class APIError(Exception):
    status: int
    code: str
    message: str
    request_id: Optional[str]
    details: Any
    bucket: str       # "retryable" | "permanent" | "server_bug" | "proxy_timeout"
    action: str       # human-readable action hint for the caller

def per_field(details: Any) -> list:
    """Normalise `error.details` for a `validation_failed` into a list.

    The service usually emits `{"errors": [...]}`, matching the
    `ValidationFailedDetails` schema; older releases emitted the bare FastAPI
    array, and splice's response-size cap emits a different object entirely
    (`{task, record_count, maximum_records, sequence_length, threshold}`, no
    `errors` key). This helper accepts either list shape and treats anything
    else — including that one — as "no structured detail"; read
    `error.message` there.
    """
    if isinstance(details, list):
        return details
    if isinstance(details, dict):
        errors = details.get("errors")
        return errors if isinstance(errors, list) else []
    return []

def handle(response: requests.Response) -> dict:
    """Convert an API response into either a parsed body or a raised APIError.

    Single source of truth for every code in the catalogue. If we ever add
    a new error code, the `unknown_code` branch surfaces it cleanly.
    """
    if response.ok:
        return response.json()

    # 504 from the edge proxy is the one response on the contract that
    # is NOT the unified error envelope. Treat as a special case.
    if response.status_code == 504:
        raise APIError(
            status=504, code="gateway_timeout",
            message="Upstream proxy read timeout (300 s).",
            request_id=None, details=None,
            bucket="proxy_timeout",
            action="Retry the same request with header `Prefer: respond-async`.",
        )

    body = response.json() if response.headers.get("content-type", "").startswith("application/json") else {}
    err = (body or {}).get("error") or {}
    code = err.get("code", "unknown_code")
    msg = err.get("message", "")
    request_id = err.get("request_id")
    details = err.get("details")

    # Retryable transient — the API or upstream is asking you to back off.
    if code == "too_many_requests":
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="retryable",
                       action="Honour Retry-After. Both rate-bucket and concurrency 429s use this code.")
    if code == "rate_limited":
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="retryable",
                       action="Same treatment as too_many_requests — honour Retry-After.")
    if code == "model_loading":
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="retryable",
                       action="Wait ~30 s and retry. The model is warming up.")
    if code == "service_unavailable":
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="retryable",
                       action="Backoff and retry. Server is starting up or shutting down.")

    # Permanent client errors — surface to the caller, don't retry.
    if code == "bad_request":
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="permanent",
                       action="Fix the HTTP-level precondition and re-issue.")
    if code == "unauthorized":
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="permanent",
                       action="Check Authorization header. WWW-Authenticate splits invalid_request vs invalid_token.")
    if code == "forbidden":
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="permanent",
                       action="Key is registered but disabled. Contact the API operator.")
    if code == "not_found":
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="permanent",
                       action="Unknown route, unknown job, or cross-tenant access (returned as 404 not 403).")
    if code == "model_not_found":
        # details.available_models is the typed list of alternatives.
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="permanent",
                       action=f"Pick a model from details.available_models: {details}.")
    if code == "validation_failed":
        # `details` is usually per-field validation output. Read it
        # DEFENSIVELY: the service wraps the FastAPI errors in an object
        # ({"errors": [...]}), matching the schema; the bare array
        # ([{loc, msg, type}, ...]) is the legacy shape this helper still
        # tolerates. Splice's response-size cap is a validation_failed with
        # neither shape — {task, record_count, maximum_records,
        # sequence_length, threshold} — and neither is expression's refusal of
        # a window with too little real (non-N) sequence around the TSS,
        # whose details are {invalid_characters: [], invalid_positions: [],
        # valid_characters: [...]} with empty lists. So never index
        # details["errors"], and never infer the cause from details.
        # Never make control flow depend on its shape — or on `loc`. Sequence
        # bound violations report at ["body","sequence"], an unknown options
        # key at ["body","options","<key>"], but expression's tss_index checks
        # come from a whole-model validator and report at ["body"] — there is
        # no ["body","tss_index"] to match on. Switch on `code` only.
        raise APIError(response.status_code, code, msg, request_id, per_field(details),
                       bucket="permanent",
                       action="Surface message verbatim — it's deterministic. Never branch on details[].loc.")
    if code == "task_not_supported_by_model":
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="permanent",
                       action="Pick a different model — see details.supported_tasks.")
    if code == "payload_too_large":
        # 413 #1: the 16 MiB raw-body cap, enforced before parsing. NOT the
        # sequence-length cap — an over-500,000 bp sequence is a 422.
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="permanent",
                       action="Body exceeded 16 MiB. Split the request or use sequence references if available.")
    if code == "sync_too_large":
        # 413 #2: an operation refusing synchronous JSON delivery above its
        # published cap — the composite find-genes-and-predict-expression
        # workflow above 50,000 bp, or annotation above 200,000 bp. details
        # carries {sequence_length, threshold}. Resend the identical body async.
        # (BED/GFF3 annotation responses are uncapped and never land here.)
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="permanent",
                       action="Resend the same body with `Prefer: respond-async`.")
    if code == "unsupported_format":
        # 415. Per-task format sets differ (promoter json|bed|bedgraph,
        # splice json|bed|gff3, enhancer json|bedgraph, chromatin json|bed,
        # annotation json|bed|gff3, expression json only) and there is no
        # silent fallback to JSON. The message names the supported values.
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="permanent",
                       action="The Accept / ?format= value is not in this task's allowed set.")
    if code == "conflict":
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="permanent",
                       action="Reserved (e.g. busy-model unload conflict). Retry once after delay.")
    if code == "job_expired":
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="permanent",
                       action="Async result evicted (24 h TTL or process restart). Resubmit the original POST.")

    # Server-side capacity / timing — retry once, then shrink the request.
    if code == "timeout":
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="retryable",
                       action="Inference exceeded the server deadline. Resubmit with `Prefer: respond-async`.")
    if code == "insufficient_memory":
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="retryable",
                       action="GPU memory exhausted. Retry after a delay, or submit a shorter sequence.")

    # Server bug — capture and escalate.
    if code == "internal_error":
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="server_bug",
                       action=f"Email contact@genomicintelligence.ai with request_id={request_id}.")
    if code in ("http_error", "unknown"):
        raise APIError(response.status_code, code, msg, request_id, details,
                       bucket="server_bug",
                       action="Generic failure the service could not classify. Capture request_id.")

    # Unknown code — the enum is closed today (21 values) but a new code can
    # ship before your client is regenerated, so treat an unlisted value as a
    # generic failure rather than a parse error. Log loudly.
    raise APIError(response.status_code, code, msg, request_id, details,
                   bucket="server_bug",
                   action=f"Unknown error.code={code!r}. Treat as server bug; capture and escalate.")

def demo() -> int:
    if not API_KEY:
        print("ERROR: set GI_API_KEY"); return 2

    # Trigger 422 validation_failed. "ZZZZ" is both out-of-alphabet and below
    # promoter's 300 bp floor (each task publishes its own minLength: splice
    # 100, enhancer 50, chromatin 200, annotation 1000, expression 9198 with the TSS at least 4599 from each end).
    # Either way the code is validation_failed — which is exactly why you
    # switch on `code` and not on the shape of `details`.
    print("# trigger validation_failed (422)")
    r = requests.post(
        f"{BASE_URL}/v1/tasks/promoter/predict",
        headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
        json={"sequence": "ZZZZ", "sequence_name": "bad-input"},
        timeout=30,
    )
    try:
        handle(r)
    except APIError as exc:
        print(f"  caught: {exc.code} bucket={exc.bucket}")
        print(f"  action: {exc.action}")
        print(f"  message: {exc.message}")

    # Trigger 401 unauthorized with a clearly-bad key.
    print("\n# trigger unauthorized (401)")
    r = requests.get(
        f"{BASE_URL}/v1/tasks/promoter/models",
        headers={"Authorization": "Bearer gi_definitely_not_a_real_key"},
        timeout=10,
    )
    try:
        handle(r)
    except APIError as exc:
        print(f"  caught: {exc.code} bucket={exc.bucket}")
        print(f"  action: {exc.action}")
        print(f"  www-authenticate: {r.headers.get('WWW-Authenticate')}")

    return 0

if __name__ == "__main__":
    raise SystemExit(demo())
```