Client examples

Minimal TypeScript and Python clients with explicit error and retry handling.

Use an assigned HTTPS origin without /v1. These helpers return the complete API envelope. For writes, persist one idempotency key per logical operation; pass the same idempotency_key to every attempt, including retries after a process restart. Automatic HTTP retries are limited to errors the server marks as retryable; transport exceptions propagate to the caller. Install the Python requests package before using the example.

TypeScript:

const origin = process.env.CITERA_API_BASE_URL;
if (!origin || !process.env.CITERA_API_KEY) throw new Error('Set CITERA_API_BASE_URL and CITERA_API_KEY');
const BASE = `${origin.replace(/\/$/, '')}/v1`;
export async function citera<T>(path: string, init: RequestInit = {}): Promise<T> {
  const headers = new Headers(init.headers);
  headers.set("Authorization", `Bearer ${process.env.CITERA_API_KEY}`);
  if (init.body && !headers.has("Content-Type")) headers.set("Content-Type", "application/json");
  const res = await fetch(`${BASE}${path}`, { ...init, headers });
  const body = await res.json();
  if (!res.ok) {
    const err = new Error(`${body.error.code}: ${body.error.message} (request ${body.request_id})`);
    Object.assign(err, { status: res.status, retryable: body.error.retryable, retryAfter: res.headers.get('Retry-After') });
    throw err;
  }
  return body as T;
}

const origin = process.env.CITERA_API_BASE_URL;
if (!origin || !process.env.CITERA_API_KEY) throw new Error('Set CITERA_API_BASE_URL and CITERA_API_KEY');
const BASE = `${origin.replace(/\/$/, '')}/v1`;
export async function citera<T>(path: string, init: RequestInit = {}): Promise<T> {
  const headers = new Headers(init.headers);
  headers.set("Authorization", `Bearer ${process.env.CITERA_API_KEY}`);
  if (init.body && !headers.has("Content-Type")) headers.set("Content-Type", "application/json");
  const res = await fetch(`${BASE}${path}`, { ...init, headers });
  const body = await res.json();
  if (!res.ok) {
    const err = new Error(`${body.error.code}: ${body.error.message} (request ${body.request_id})`);
    Object.assign(err, { status: res.status, retryable: body.error.retryable, retryAfter: res.headers.get('Retry-After') });
    throw err;
  }
  return body as T;
}

const origin = process.env.CITERA_API_BASE_URL;
if (!origin || !process.env.CITERA_API_KEY) throw new Error('Set CITERA_API_BASE_URL and CITERA_API_KEY');
const BASE = `${origin.replace(/\/$/, '')}/v1`;
export async function citera<T>(path: string, init: RequestInit = {}): Promise<T> {
  const headers = new Headers(init.headers);
  headers.set("Authorization", `Bearer ${process.env.CITERA_API_KEY}`);
  if (init.body && !headers.has("Content-Type")) headers.set("Content-Type", "application/json");
  const res = await fetch(`${BASE}${path}`, { ...init, headers });
  const body = await res.json();
  if (!res.ok) {
    const err = new Error(`${body.error.code}: ${body.error.message} (request ${body.request_id})`);
    Object.assign(err, { status: res.status, retryable: body.error.retryable, retryAfter: res.headers.get('Retry-After') });
    throw err;
  }
  return body as T;
}

const origin = process.env.CITERA_API_BASE_URL;
if (!origin || !process.env.CITERA_API_KEY) throw new Error('Set CITERA_API_BASE_URL and CITERA_API_KEY');
const BASE = `${origin.replace(/\/$/, '')}/v1`;
export async function citera<T>(path: string, init: RequestInit = {}): Promise<T> {
  const headers = new Headers(init.headers);
  headers.set("Authorization", `Bearer ${process.env.CITERA_API_KEY}`);
  if (init.body && !headers.has("Content-Type")) headers.set("Content-Type", "application/json");
  const res = await fetch(`${BASE}${path}`, { ...init, headers });
  const body = await res.json();
  if (!res.ok) {
    const err = new Error(`${body.error.code}: ${body.error.message} (request ${body.request_id})`);
    Object.assign(err, { status: res.status, retryable: body.error.retryable, retryAfter: res.headers.get('Retry-After') });
    throw err;
  }
  return body as T;
}

Python:

import os, time, random, requests

BASE = os.environ["CITERA_API_BASE_URL"].rstrip("/") + "/v1"
S = requests.Session()
S.headers["Authorization"] = f"Bearer {os.environ['CITERA_API_KEY']}"

def call(method, path, *, json=None, idempotency_key=None, attempts=5):
    # The caller persists one key per logical write, including across restarts.
    headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {}
    for i in range(attempts):
        r = S.request(method, BASE + path, json=json, headers=headers, timeout=30)
        body = r.json()
        if r.ok:
            return body
        if body["error"]["retryable"] and i < attempts - 1:
            time.sleep(float(r.headers.get("Retry-After", 2 ** i)) + random.uniform(0, 0.5))
            continue
        raise RuntimeError(f"{body['error']['code']}: {body['error']['message']} ({body['request_id']})")

import os, time, random, requests

BASE = os.environ["CITERA_API_BASE_URL"].rstrip("/") + "/v1"
S = requests.Session()
S.headers["Authorization"] = f"Bearer {os.environ['CITERA_API_KEY']}"

def call(method, path, *, json=None, idempotency_key=None, attempts=5):
    # The caller persists one key per logical write, including across restarts.
    headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {}
    for i in range(attempts):
        r = S.request(method, BASE + path, json=json, headers=headers, timeout=30)
        body = r.json()
        if r.ok:
            return body
        if body["error"]["retryable"] and i < attempts - 1:
            time.sleep(float(r.headers.get("Retry-After", 2 ** i)) + random.uniform(0, 0.5))
            continue
        raise RuntimeError(f"{body['error']['code']}: {body['error']['message']} ({body['request_id']})")

import os, time, random, requests

BASE = os.environ["CITERA_API_BASE_URL"].rstrip("/") + "/v1"
S = requests.Session()
S.headers["Authorization"] = f"Bearer {os.environ['CITERA_API_KEY']}"

def call(method, path, *, json=None, idempotency_key=None, attempts=5):
    # The caller persists one key per logical write, including across restarts.
    headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {}
    for i in range(attempts):
        r = S.request(method, BASE + path, json=json, headers=headers, timeout=30)
        body = r.json()
        if r.ok:
            return body
        if body["error"]["retryable"] and i < attempts - 1:
            time.sleep(float(r.headers.get("Retry-After", 2 ** i)) + random.uniform(0, 0.5))
            continue
        raise RuntimeError(f"{body['error']['code']}: {body['error']['message']} ({body['request_id']})")

import os, time, random, requests

BASE = os.environ["CITERA_API_BASE_URL"].rstrip("/") + "/v1"
S = requests.Session()
S.headers["Authorization"] = f"Bearer {os.environ['CITERA_API_KEY']}"

def call(method, path, *, json=None, idempotency_key=None, attempts=5):
    # The caller persists one key per logical write, including across restarts.
    headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {}
    for i in range(attempts):
        r = S.request(method, BASE + path, json=json, headers=headers, timeout=30)
        body = r.json()
        if r.ok:
            return body
        if body["error"]["retryable"] and i < attempts - 1:
            time.sleep(float(r.headers.get("Retry-After", 2 ** i)) + random.uniform(0, 0.5))
            continue
        raise RuntimeError(f"{body['error']['code']}: {body['error']['message']} ({body['request_id']})")

API & MCP overview

Get help