celestin 0.1.0a1

Official Python SDK for Celestin — zero-knowledge bank reconciliation API

celestin — Official Python SDK

Zero-knowledge bank reconciliation API — Python client library.

---

Features

• Full API coverage — Reconciliations and Tenants endpoints.
• Zero-knowledge anonymisation — Client-side PII tokenisation with HKDF + HMAC-SHA256. Your raw data never reaches the server.
• Automatic retries — Exponential backoff with jitter on 429 / 5xx. Honours Retry-After.
• Typed error hierarchy — isinstance-check on CelestinAuthError, CelestinValidationError, etc.
• Idempotent POSTs — Auto-generated Idempotency-Key (UUID v4) on every POST.
• Python 3.11+ — Uses native generics and from __future__ import annotations.

---

Installation

pip install celestin

With NER Layer 2 support (Phase 12+):

pip install "celestin[ner]"

---

Quickstart

import os
from celestin import Celestin

client = Celestin(
    api_key=os.environ["CELESTIN_API_KEY"],
    tenant_id="acme-sl",
)

# Submit a reconciliation job
job = client.reconciliations.create(body={
    "bank": [
        {"id": "b1", "date": "2026-03-01", "amount": -1200.50, "description": "PAGO PROVEEDOR"},
    ],
    "ledger": [
        {"id": "l1", "date": "2026-03-01", "debit": 0.0, "credit": 1200.50,
         "account": "400", "description": "Factura proveedor"},
    ],
})
print(f"Job id: {job['id']}, status: {job['status']}")

# Poll for completion
result = client.reconciliations.retrieve(job["id"])
print(f"Matches: {result['summary'].get('matches_total')}")

---

Zero-Knowledge Walkthrough

Celestin's anonymisation layer lets you run AI-powered reconciliation without sending plaintext PII (NIFs, IBANs, emails, phones, card numbers) to any external server.

How it works

1. You call Anonymizer.redact() on every text field before building the request body.
2. The SDK replaces each PII value with a deterministic opaque token such as <NIF:A3KM8P7ZNQWR5>.
3. The server processes the tokenised data and returns results that also contain the tokens.
4. You call Anonymizer.deanonymize() on any returned strings to recover the originals.

The token is computed as:

tenant_salt = HKDF-SHA256(master_key, tenant_id_utf8, "celestin-anon-v2", 32)
mac         = HMAC-SHA256(tenant_salt, normalised_value_utf8)
token       = "<TYPE:" + CrockfordBase32(mac[0:8]) + ">"

The master_key never leaves your process.

Code example

import os
from celestin import Celestin, Anonymizer

# 32-byte master key (keep secret, never log)
master_key = bytes.fromhex(os.environ["CELESTIN_MASTER_KEY"])

anon = Anonymizer(master_key=master_key, tenant_id="acme-sl")

raw_description = "Pago a 12345678Z IBAN ES9121000418450200051332"
safe = anon.redact(raw_description)
# safe -> "Pago a <NIF:...> IBAN <IBAN:...>"

client = Celestin(api_key=os.environ["CELESTIN_API_KEY"], tenant_id="acme-sl")
job = client.reconciliations.create(body={
    "bank": [{"id": "b1", "date": "2026-03-01", "amount": -1200.50,
              "description": safe}],
    "ledger": [...],
    "tokens": {
        "scheme": "hmac-sha256-base32-v2",
        "salt_fingerprint": anon.fingerprint(),
    },
})

# Recover original values in the returned match explanations
for match in job.get("matches", []):
    if "explanation" in match:
        match["explanation"] = anon.deanonymize(match["explanation"])

# Clean up when done with this reconciliation
anon.destroy()

---

Error Handling

from celestin import (
    Celestin,
    CelestinAuthError,
    CelestinValidationError,
    CelestinRateLimitError,
    CelestinConnectionError,
    CelestinError,
)
import time

client = Celestin(api_key="sk_test_...")

try:
    job = client.reconciliations.create(body={...})
except CelestinAuthError:
    # 401 — invalid or expired API key
    print("Check your CELESTIN_API_KEY environment variable.")
except CelestinValidationError as exc:
    # 422 — request body failed server-side validation
    for fe in exc.field_errors:
        print(f"  Field {fe['field']}: {fe['message']}")
except CelestinRateLimitError as exc:
    # 429 — rate limited
    wait = exc.retry_after_seconds or 60
    time.sleep(wait)
except CelestinConnectionError:
    # Network failure (all retries exhausted)
    print("Could not reach api.celestin.es — check your internet connection.")
except CelestinError as exc:
    # Catch-all for all Celestin errors
    print(f"Error {exc.code} (HTTP {exc.status}): {exc}")

---

Configuration

Parameter	Type	Default	Description
api_key	str	required	sk_test_... or sk_live_...
base_url	str	https://api.celestin.es	Override for sandbox
tenant_id	str	None	Default tenant for all requests
master_key	str	None	Hex-encoded 32-byte key for anonymisation
timeout_ms	int	30_000	Per-request timeout (milliseconds)
user_agent	str	None	Appended to the SDK's User-Agent

client = Celestin(
    api_key="sk_test_...",
    base_url="https://sandbox-api.celestin.es",  # sandbox
    tenant_id="my-tenant",
    timeout_ms=10_000,
    user_agent="my-app/1.0",
)

Context manager

with Celestin(api_key="sk_test_...") as client:
    jobs = client.reconciliations.list()
# HTTP connection closed automatically

---

Resources

client.reconciliations

Method	Description
create(body, *, tenant_id=None, idempotency_key=None, timeout_ms=None)	Submit a new reconciliation job
retrieve(job_id, *, tenant_id=None, timeout_ms=None)	Get a job by id
list(*, limit=None, cursor=None, status=None, tenant_id=None, timeout_ms=None)	List jobs
cancel(job_id, *, tenant_id=None, timeout_ms=None)	Cancel a queued/processing job

client.tenants

Method	Description
create(body, *, timeout_ms=None)	Create a tenant
retrieve(id, *, timeout_ms=None)	Get a tenant by id or external_id
list(*, limit=None, cursor=None, timeout_ms=None)	List tenants
update(id, body, *, timeout_ms=None)	Partially update a tenant
soft_delete(id, *, timeout_ms=None)	Soft-delete a tenant

---

SDK Parity

This SDK is byte-for-byte compatible with @celestin/sdk-node on the anonymisation tokenisation contract. A token emitted by this SDK for the triple (master_key, tenant_id, value) is identical to the token the Node SDK emits for the same triple. Phase 12 will ship a shared test-vector suite that pins this invariant across all official SDKs.

---

Development

# Clone and set up
git clone https://github.com/team-banzai/celestin
cd celestino/sdk-python
python -m venv .venv
.venv/bin/pip install -e ".[dev]"

# Run tests
.venv/bin/pytest tests/ -v

# Type check
.venv/bin/mypy --strict src/celestin/

---

License

MIT — see LICENSE.

Copyright (c) 2026 Team Banzai S.L.U.