Python SDK — Sovereign Frontier

Installation

shell

$ pip install sovereign-router
# or with optional PyNaCl for manifest signing helpers
$ pip install sovereign-router pynacl

Requires requests ≥ 2.31. No other runtime dependencies.

Quickstart

python — AIEM registry

from sovereign_router import AiemRegistryClient

sf = AiemRegistryClient(api_key="sk-sf-...")

# list top sealed models
models = sf.list_artifacts("models", tier="sealed", limit=10)
for m in models:
    print(m["id"], m["tier"])

# cryptographically verify an artifact
result = sf.verify("models", "qwen/qwen3-235b-a22b")
assert result["valid"]

python — inference gateway

from sovereign_router import SovereignRouterClient

sr = SovereignRouterClient(
    base_url="https://router.example.com",
    api_key="sk-router-...",
    policy="clinical-strict",
)
reply = sr.chat(
    model="qwen/qwen3-235b-a22b",
    messages=[{"role": "user", "content": "Summarise the study"}],
)
print(reply["choices"][0]["message"]["content"])
print("rekor_index:", reply["attestation"]["rekor_index"])

AiemRegistryClient

Interacts with the Sovereign Frontier AIEM artifact registry — publishing, verifying, pulling, and listing attested AI artifacts. Connects to https://api.sovereignfrontier.ai/api/v1 by default.

AiemRegistryClient(api_key=None, base_url=None, timeout=30.0, session=None)

Parameter	Type	Description
api_key	str \| None	Registry key sent as `X-SF-Key` header. Omit for public read-only calls.
base_url	str \| None	Override the default API base URL.
timeout	float	Per-request timeout in seconds. Default `30.0`.
session	requests.Session \| None	Provide a custom `requests.Session` (e.g. with mTLS or retries).

publish() POST

sf.publish(manifest: dict, signature_b64: str) → dict

Publish an AIEM-1 manifest to the registry. The manifest must be signed with an Ed25519 key whose public key is registered with Sovereign Frontier.

Parameter	Type	Description
manifest	dict	Full AIEM-1 manifest. Build one with `build_manifest()`.
signature_b64	str	Ed25519 signature over canonical JSON of `manifest`, base64-encoded.

Returns

dict — artifact_id, category, rekor_index, inclusion_proof, tier_assigned, message

example

import json, base64
from nacl.signing import SigningKey

key = SigningKey.generate()
manifest = sf.build_manifest(
    category="models",
    artifact_id="my-org/my-model-v1",
    name="My Model v1",
    content_hash="sha256:abc123...",
    publisher_id="my-org",
)
sig_b64 = base64.b64encode(
    key.sign(json.dumps(manifest, sort_keys=True).encode()).signature
).decode()
result = sf.publish(manifest, sig_b64)
print("rekor_index:", result["rekor_index"])

build_manifest() static

AiemRegistryClient.build_manifest(*, category, artifact_id, name, content_hash, publisher_id, ...) → dict

Static helper that builds a minimal valid AIEM-1 manifest dict ready for signing.

Parameter	Type	Description
category	str	Artifact category: `models`, `datasets`, `architectures`, `optimizers`, `tools`, `protocols`, `systems`, `processors`.
artifact_id	str	Canonical artifact identifier (e.g. `org/name-version`).
name	str	Human-readable display name.
content_hash	str	SHA-256 hash of the artifact binary (`sha256:<hex>`).
publisher_id	str	Registered publisher identifier.
description	str \| None	Optional short description.
size_bytes	int \| None	Artifact file size.
format	str \| None	Format string (e.g. `safetensors`, `gguf`).
licenses	list[str] \| None	SPDX license identifiers.
source_url	str \| None	Canonical upstream URL.
commit	str \| None	Git commit SHA if built from source.
sbom_ref	str \| None	CycloneDX / SPDX SBOM reference.
tier	str	Initial trust tier. Default `quarantine`.

verify()

sf.verify(category: str, artifact_id: str) → dict

Replay the full cryptographic attestation chain for an artifact. Checks Ed25519 signatures and Merkle inclusion proofs.

Returns

dict — artifact_id, category, valid, tier, chain_length, errors, warnings, manifest

example

result = sf.verify("models", "qwen/qwen3-235b-a22b")
if not result["valid"]:
    print("chain errors:", result["errors"])

pull()

sf.pull(category: str, artifact_id: str, require_policy: str | None = None) → dict

Fetch artifact metadata and a presigned download URL, optionally gated by a policy.

Parameter	Type	Description
require_policy	str \| None	Policy ID that must pass before a download URL is issued.

Returns

dict — artifact_id, category, download_url, content_hash, manifest, policy_gate

list_artifacts()

sf.list_artifacts(category: str, *, tier=None, publisher=None, q=None, page=0, limit=50) → list

List artifacts in a registry category with optional filters.

Parameter	Type	Description
category	str	One of `models`, `datasets`, `architectures`, `optimizers`, `tools`, `protocols`, `systems`, `processors`.
tier	str \| None	Filter by trust tier: `sealed`, `hardened`, `quarantine`, `interdicted`.
publisher	str \| None	Filter by publisher ID.
q	str \| None	Free-text search query.
page	int	Zero-based page number.
limit	int	Items per page (max 200).

get_manifest()

sf.get_manifest(category: str, artifact_id: str) → dict

Fetch the full AIEM-1 manifest for an artifact, including all attestation fields.

status() · list_publishers()

sf.status() → dict

Registry health summary — tree size, federation state, active alerts.

sf.list_publishers() → list

All registered publishers with tier/revocation metadata.

SovereignRouterClient

Drop-in replacement for the OpenAI client targeting a SovereignRouter gateway. Every response includes an attestation field with a Rekor index and trust metadata.

SovereignRouterClient(base_url, api_key=None, policy=None, admin_token=None, timeout=60.0)

Parameter	Type	Description
base_url	str	Router base URL (e.g. `https://router.example.com`).
api_key	str \| None	Tenant key sent as `X-Router-Key`.
policy	str \| None	Default policy ID applied to all requests. Can be overridden per call.
admin_token	str \| None	Admin token for tenant/provider management endpoints.
timeout	float	Per-request timeout in seconds.

chat()

sr.chat(model, messages, max_tokens=None, temperature=None, policy=None, provider=None) → dict

Non-streaming chat completion. Returns the full OpenAI-compatible response plus an attestation key extracted from _sf_attestation.

example

reply = sr.chat(
    model="qwen/qwen3-235b-a22b",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user",   "content": "Explain Merkle trees."},
    ],
    max_tokens=512,
    policy="default",
)
print(reply["choices"][0]["message"]["content"])
att = reply["attestation"]
print(f"tier={att['tier_observed']}  rekor={att['rekor_index']}")

chat_stream()

sr.chat_stream(model, messages, ...) → Iterator[dict]

Server-sent-event streaming. Yields event dicts. The final event is always an attestation event followed by a done event.

Event `kind`	Keys	Description
`role`	role	Initial role announcement from the model.
`delta`	content	Incremental content chunk.
`finish`	finish_reason	Completion reason (`stop`, `length`, etc.).
`attestation`	payload	Full SF attestation object (same as non-streaming).
`done`	—	Stream sentinel.

example

for event in sr.chat_stream(
    model="qwen/qwen3-235b-a22b",
    messages=[{"role": "user", "content": "Count to 5."}],
):
    if event["kind"] == "delta":
        print(event["content"], end="", flush=True)
    elif event["kind"] == "attestation":
        print("\nrekor:", event["payload"]["rekor_index"])

Provider management

sr.list_providers() → list[dict]

List all registered inference providers.

sr.register_provider(p: dict) → dict admin

Register a new provider (requires admin_token). Fields: id, kind, display_name, endpoint, serves_models.

sr.provider_health() → dict

Current health snapshot — successes, failures, latency p50/p95, score for each provider.

Policy management

sr.list_policies() → list[dict]

All defined routing policies.

sr.get_policy(pid: str) → dict

Policy details: require_tier, allowed_origins, require_drift_clean, cost/latency caps.

Tenant management

sr.create_tenant(**opts) → dict admin

Mint a tenant and return a fresh sk-router-... API key.

Option	Type	Description
id	str	Unique tenant identifier.
display_name	str	Human-readable label.
usd_cap	float \| None	Monthly spend cap in USD.
requests_per_minute	int	Rate limit (default 60).
allowed_policies	list[str]	Policy IDs this tenant may use.

sr.list_tenants() → list[dict]

sr.get_tenant(tid: str) → dict

Audit feed

sr.recent_calls(limit: int = 50) → dict

Recent inference call log with tenant, model, policy, and attestation metadata.

sr.rekor_entries(kind=None, limit=50) → dict

Raw Rekor transparency-log entries, optionally filtered by entry kind.

Supply-chain endpoints

sr.get_bom(category: str, artifact_id: str) → dict

SBOM + risk score for one artifact (GET /api/v1/supply-chain/:category/*id).

sr.list_suppliers() → dict

Ranked supplier roster with revocation and risk metadata.

sr.supply_chain_risk_summary() → dict

Fleet rollup by tier, risk band, and sanctioned-entity flags.

sr.firmware_receipt(category: str, artifact_id: str, build=None) → dict

Offline-verifiable firmware receipt bundle. Pass build to pin to a specific build ID.

sr.verify_firmware_receipt(receipt: dict, binary_sha256_hex: str) → dict

Local-only verifier (no network). Replays each attestation's RFC 6962 Merkle audit path, checks the binary content hash.

Returns

dict — ok: bool, errors: list[str], warnings: list[str]

Error handling

Both clients raise typed exceptions on non-2xx responses:

AiemRegistryError(message)

Raised by AiemRegistryClient on HTTP errors. Message includes status code and response body.

SovereignRouterError(message)

Raised by SovereignRouterClient on HTTP errors.

example

from sovereign_router import AiemRegistryClient, AiemRegistryError

try:
    sf.verify("models", "nonexistent/artifact")
except AiemRegistryError as e:
    print("registry error:", e)

Offline / local verification

Air-gapped environments: use verify_firmware_receipt() to confirm an artifact's provenance without any network access. Only the receipt bundle (pre-fetched) and the artifact binary SHA-256 are needed.

example

import hashlib, json

# pre-fetch the receipt once (can be done online)
receipt = sr.firmware_receipt("models", "qwen/qwen3-235b-a22b")
with open("receipt.json", "w") as f:
    json.dump(receipt, f)

# ── later, fully offline ──────────────────────────────────────────
with open("receipt.json") as f:
    receipt = json.load(f)
with open("model.safetensors", "rb") as f:
    sha256 = hashlib.sha256(f.read()).hexdigest()

result = sr.verify_firmware_receipt(receipt, sha256)
if not result["ok"]:
    raise RuntimeError(result["errors"])