SPECTR Engine Docs

Showing the in-process Python wheel. Switch to API mode (sidebar) for the hosted HTTPS endpoints.

Start Here

UFM takes bytes, decomposes them into structural primitives, and gives you repeatable results you can use in software. The same engine runs in two ways. Pick one with the Local / API toggle in the sidebar, then everything below adapts.

# 1. Download the wheel for your Python version (Windows only today).
curl.exe -fSL -H "X-Api-Key: %UFM_KEY%" `
  -o ufm-3.0.0-cp313-cp313-win_amd64.whl `
  "https://api.spectrengine.com/v1/wheel?python=3.13&os=windows&arch=amd64"

# 2. Install + activate.
pip install ufm-3.0.0-cp313-cp313-win_amd64.whl
python -m ufm activate %UFM_KEY%

import ufm

eng = ufm.InvariantIdentityEngine(storage_path='ledger.bin')
seed, status = eng.process(b'Hello World')
print(seed, status)
print(eng.reconstruct(b'Hello World'))
eng.save()

Plain-English Terms

Local Python Wheel

The hosted API and the local wheel are two surfaces over the same native UFM engine. Use the wheel when you want the engine inside your own process, with local ledgers, no HTTP base64 wrapper, and no per-request network call after activation.

What you need first

• An account at spectrengine.com and an API key from the dashboard. Keys look like ufm_live_a1b2c3d4.your_secret_here.
• Python 3.12, 3.13, or 3.14 on 64-bit Windows. macOS and Linux wheels are not yet published.
• curl on the path. Modern Windows 10/11 ships curl.exe by default.

1. Detect your Python version

The wheel must match your Python minor version exactly. Run:

py -c "import sys; print(f'{sys.version_info.major}.{sys.version_info.minor}')"
# → 3.12  or  3.13  or  3.14

If you have multiple Pythons installed, decide now which one you want UFM in. Use the matching py -3.13 / py -3.12 launcher in the steps below.

2. Download the wheel with your API key

The wheel is gated by your API key. There is no public download URL. Pick the matching Python version and run one of the commands below from the folder where you want the wheel to land.

cmd.exe / Git Bash:

set UFM_KEY=ufm_live_a1b2c3d4.your_secret_here

curl -fSL ^
  -H "X-Api-Key: %UFM_KEY%" ^
  -o ufm-3.0.0-cp313-cp313-win_amd64.whl ^
  "https://api.spectrengine.com/v1/wheel?python=3.13&os=windows&arch=amd64"

PowerShell:

$env:UFM_KEY = "ufm_live_a1b2c3d4.your_secret_here"

curl.exe -fSL `
  -H "X-Api-Key: $env:UFM_KEY" `
  -o ufm-3.0.0-cp313-cp313-win_amd64.whl `
  "https://api.spectrengine.com/v1/wheel?python=3.13&os=windows&arch=amd64"

Use curl.exe (not curl) in PowerShell. The alias maps to Invoke-WebRequest which has different flags.

List what's currently shipped:

curl.exe -fSL -H "X-Api-Key: %UFM_KEY%" https://api.spectrengine.com/v1/wheel/manifest
# → {"wheels":[{"filename":"ufm-3.0.0-cp313-cp313-win_amd64.whl","python":"3.13",...}, ...]}

3. Install and activate

py -3.13 -m venv .venv
.\.venv\Scripts\activate

pip install ufm-3.0.0-cp313-cp313-win_amd64.whl
python -m ufm activate %UFM_KEY%
python -m ufm status
python -m ufm version

# Expected: "active": true, your tier, and a non-zero days_remaining.
# After this, "import ufm" works in any script using this venv.

Pass the full key to activate (public_id, dot, and secret). Activation hits api.spectrengine.com once and caches a signed 30-day licence token under %LOCALAPPDATA%\\ufm\\licence.json. Subsequent import ufm calls use the cached token and do not need network.

For unattended jobs, you can also set UFM_LICENCE_KEY before the first engine call. Set UFM_LICENCE_CACHE_DIR when you want the licence cache to live in a controlled directory.

4. Troubleshooting

• 403 on the download: wrong or revoked API key. Re-issue in the dashboard.
• 404 on the download: the python/os/arch combo is not shipped. Hit /v1/wheel/manifest to see what is.
• pip says "not a supported wheel on this platform": the wheel's cpXYZ tag does not match the active Python. Run the detect command in step 1 inside the same venv.
• activate fails with network error: activation must reach api.spectrengine.com the first time. Once cached, the engine runs fully offline.
• status says active: false after a clean activate: remove %LOCALAPPDATA%\\ufm\\licence.json and re-run activate.

import ufm

status = ufm.licence_status()
print(status["active"])
print(status["tier"])
print(status["days_remaining"])

Mental model

• HTTP requests use base64 strings. Local calls take raw bytes.
• API persistence is your account ledger. Local persistence is the storage_path file you choose.
• Request-scoped API endpoints map to temporary local ledgers or stateless helpers.
• Local Ben history lives in a BenSession; persist it yourself if you need durable chat history.
• There are no hosted rate limits locally, but the 100 MB input limit and licence scopes still apply.

Encoding your data for ingestion

The engine ingests raw bytes. How you turn your input (strings, numbers, structured records, files) into bytes shapes what the analytics actually measure. Three rules of thumb:

• Text: encode with .encode("utf-8"). Each character becomes 1 to 4 bytes. Frequency analysis on UTF-8 text reflects character distribution at the byte level.
• Structured records (numbers, fields): pack each field as a fixed-width byte chunk so primitives line up with field boundaries. For example, six small integers fit in bytes([n1, n2, n3, n4, n5, n6]). With SymbolLengthMode.Fixed(8) each byte becomes its own primitive, so frequency_histogram tells you exactly how often each value appears.
• Files / blobs: read with Path.read_bytes() and pass straight in. Useful for binary diffing, structural similarity between revisions, or detecting partial reuse.

The other big choice: one ingest per record vs one ingest per corpus. Each process() call produces its own seed and records its own range in the timeline.

• Per record (recommended for analysis): call process(record) or process_batch([record1, record2, ...]). You get one seed per record, can replay(seed) any specific record later, and frequency / discovery analytics reflect cross-record statistics.
• One big blob: concatenate everything and call process(blob) once. The whole corpus has a single seed. Useful when you only care about the corpus as a whole (e.g. fingerprinting a backup file).

And the symbol_length choice:

• auto_curve (default): the engine picks the symbol length that maximises information density for your input. Right answer for unstructured text, code, and binary blobs.
• Fixed(8): one byte per primitive. Use when each byte has its own meaning (one-byte-per-field records, ASCII histograms, raw bytes you want counted as-is).
• Fixed(16) or higher: longer chunks. Use when meaningful units in your input are larger than one byte (UTF-16 text, packed structs).

Core ingest, persist, and replay

This is the local equivalent of POST /v1/process, GET /v1/replay/{seed}, and POST /v1/reconstruct.

import ufm

data = b"Hello World"

# storage_path MUST be a relative path inside your current working
# directory. tempfile.TemporaryDirectory() lives outside cwd on most
# systems and will be rejected. Pass a plain "ledger.bin" or a path
# under "./" instead.
with ufm.InvariantIdentityEngine(storage_path="customer-ledger.bin") as eng:
    seed, status = eng.process(data)
    print(seed, status)                  # NOVELTY on first ingest, e.g. (0, NOVELTY)

    seed2, status2 = eng.process(data)
    print(seed2 == seed, status2)        # True, REPLAY

    assert eng.reconstruct(data) is True

    replayed = [bytes(seq) for seq in eng.replay(seed)]
    print(replayed[0])                   # b"Hello World"

    print(eng.ledger_summary())

# save() is called automatically when the context exits.

About that small seed value: it is a primitive-set index, not a content hash, so seeds for early or simple inputs are commonly 0, 1, or other small integers. See the glossary for the full definition.

Full engine analysis

Use this pattern when you want the same layers as POST /v1/engine: core identity, universal pipeline quality, timeline, frequency, discovery, and optional structural comparison.

import tempfile
import ufm

def to_bits(data: bytes) -> list[int]:
    return [int(bit) for byte in data for bit in f"{byte:08b}"]

def mode_from_strategy(strategy: dict | None, fallback: str = "auto_curve") -> str:
    raw = str((strategy or {}).get("symbol_length_mode") or fallback).lower()
    if raw in ("autocurve", "auto_curve"):
        return "auto_curve"
    if raw == "entropy":
        return "entropy"
    if raw.startswith("fixed(") and raw.endswith(")"):
        return f"fixed{raw[6:-1]}"
    return raw if raw.startswith("fixed") else fallback

def compare_bytes(a: bytes, b: bytes, mode: str = "auto_curve") -> dict:
    la = ufm.ingest_raw(to_bits(a), symbol_length_mode=mode)
    lb = ufm.ingest_raw(to_bits(b), symbol_length_mode=mode)
    return ufm.ledger_compare(la, lb)

def run_local_engine(
    data: bytes,
    *,
    compare_with: bytes | None = None,
    verify: bool = True,
    max_lag: int = 50,
    top_n: int = 20,
) -> dict:
    with tempfile.TemporaryDirectory(prefix="ufm-engine-") as tmp:
        up = ufm.UniversalPipeline(
            storage_path=f"{tmp}/engine-request-ledger.bin",
            zero_point=True,
            verify=verify,
        )
        universal = up.run(data)
        mode = mode_from_strategy(universal.get("strategy"))
        sig = ufm.ufm_signature(data, symbol_length_mode=mode)
        ledger = ufm.ingest_raw(to_bits(data), symbol_length_mode=mode)

        result = {
            "core": {
                "seed": universal.get("seed", sig["seed"]),
                "status": (universal.get("execute") or {}).get("status"),
                "discovery_rate": sig["discovery_rate"],
                "symbol_length": sig["symbol_length"],
                "primitive_count": sig["primitive_count"],
                "timeline_length": sig["timeline_length"],
                "reuse_ratio": sig["reuse_ratio"],
                "replay_valid": sig["replay_valid"],
                "signature": sig["signature"],
            },
            "universal": universal,
            "timeline": {
                "acf": ledger.acf(max_lag),
                "segments": ledger.segments(100),
                "transitions": ledger.transitions(100, 0.1),
            },
            "frequency": {
                "histogram": ledger.frequency_histogram(),
                "top_n": ledger.top_n_primitives(top_n),
            },
            "discovery": {
                "discovery_sequence": ledger.discovery_sequence(),
                "discovery_rate": ledger.discovery_rate,
                "primitive_count": ledger.primitive_count,
                "timeline_length": ledger.timeline_length,
                "symbol_length": ledger.symbol_length,
            },
            "effective_symbol_length_mode": mode,
            "engine_version": ufm.VERSION,
        }
        if compare_with is not None:
            result["comparison"] = compare_bytes(data, compare_with, mode)
        return result

analysis = run_local_engine(b"Hello World", compare_with=b"\xef\xbb\xbfHello World")
print(analysis["core"]["seed"])
print(analysis["universal"]["quality"])
print(analysis["comparison"]["jaccard"])

Pair comparison and semantic noise

This is the local equivalent of POST /v1/pipeline, POST /v1/compare, POST /v1/noise/detect, POST /v1/noise/delta, and POST /v1/semantic/analyze.

import ufm

def to_bits(data: bytes) -> list[int]:
    return [int(bit) for byte in data for bit in f"{byte:08b}"]

source = b"line one\nline two\n"
target = b"line one\r\nline two\r\n"

la = ufm.ingest_raw(to_bits(source))
lb = ufm.ingest_raw(to_bits(target))
structural = ufm.ledger_compare(la, lb)

semantic = ufm.SemanticDecisionPipeline("semantic-ledger.jsonl")
noise = semantic.run_with_policy(
    source,
    target,
    enabled_noise_classes=["line_ending_crlf"],
    strict_allowlist=True,
)

print(structural["jaccard"])
print(noise["converges"])       # True for classified CRLF/LF noise
print(noise["noise_units"])
print(noise["validation_checks"])
print(noise["decision_hash"])

Analytics helpers

The granular analysis endpoints are direct ledger operations locally. These work on any FluidLedger object, whether you got it from ufm.ingest_raw(...) (one-shot, in memory) or from a persistent engine (see Reopening a saved ledger below).

import ufm

def to_bits(data: bytes) -> list[int]:
    return [int(bit) for byte in data for bit in f"{byte:08b}"]

data = b"abcabcabc"
ledger = ufm.ingest_raw(to_bits(data), symbol_length_mode="auto_curve")

timeline = {
    "acf": ledger.acf(50),
    "segments": ledger.segments(100),
    "transitions": ledger.transitions(100, 0.1),
}
frequency = {
    "histogram": ledger.frequency_histogram(),
    "top_n": ledger.top_n_primitives(20),
}
discovery = {
    "discovery_sequence": ledger.discovery_sequence(),
    "discovery_rate": ledger.discovery_rate,
}
symbol_length, selector_meta = ufm.find_optimal_symbol_length(to_bits(data))
profile = ufm.structural_profile(data, symbol_width=16)

print(timeline)
print(frequency)
print(discovery)
print(symbol_length, selector_meta)
print(profile)

What each call returns:

• acf(max_lag): list of floats. Autocorrelation at lags 1..max_lag. Peaks indicate periodic structure (e.g. repeating motifs).
• segments(window): list of {start, end, label} dicts identifying contiguous regions where one primitive dominates.
• transitions(window, threshold): list of timeline indices where the local primitive distribution shifts more than threshold.
• frequency_histogram(): dict mapping {primitive_id: count}. IDs are small integers assigned in discovery order.
• top_n_primitives(n): list of (primitive_id, count) tuples ordered by descending count.
• discovery_sequence(): list of timeline positions where a new primitive was first seen. Length equals primitive_count.
• discovery_rate (property): float between 0 and 1.
• primitive_count, timeline_length, symbol_length: integer scalars on the ledger.
• ufm.structural_profile(data, symbol_width=N): dict with bit-level features (anchor offsets, symbol-length curves) computed without modifying any ledger.

Reopening and analyzing a saved ledger

After a session of eng.process(...) calls saves to disk (automatic on context exit, or explicit eng.save()), you can reopen the same storage_path and run analytics on it without re-ingesting. The engine loads the ledger lazily on first use.

import ufm

# Same path you wrote to earlier. Loaded into the engine on first call below.
with ufm.InvariantIdentityEngine(storage_path="corpus-ledger.bin") as eng:
    summary = eng.ledger_summary()
    print(summary)
    # → {"primitive_count": 248, "timeline_length": 9100,
    #    "symbol_length": 8, "discovery_rate": 0.027, ...}

    # eng.ledger() exposes the underlying FluidLedger for analytics.
    fl = eng.ledger()
    if fl is not None:
        print("top primitives:", fl.top_n_primitives(10))
        print("histogram entries:", len(fl.frequency_histogram()))
        print("discovery rate:", fl.discovery_rate)
        print("acf 1..10:", fl.acf(10))

    # Replay any seed you remember from when you ingested.
    for seed in [0, 1, 2]:
        slices = list(eng.replay(seed))
        print(f"replay({seed}) -> {len(slices)} slice(s)")

Batch and corpus workflows

Use process_batch for persisted corpus ingestion and ufm_signature_batch for independent stateless signatures. Local corpus import/export is just controlled movement of your ledger.bin file.

import shutil
import ufm

def to_bits(data: bytes) -> list[int]:
    return [int(bit) for byte in data for bit in f"{byte:08b}"]

docs = [b"file one", b"file two", b"file three"]

with ufm.InvariantIdentityEngine(storage_path="corpus-ledger.bin") as eng:
    results = eng.process_batch(docs)
    seeds = [seed for seed, _status in results]
    summary = eng.ledger_summary()

ledgers = [ufm.ingest_raw(to_bits(doc)) for doc in docs]
jaccard_matrix = [
    [ufm.ledger_jaccard(a, b) for b in ledgers]
    for a in ledgers
]

print(seeds)
print(summary)
print(jaccard_matrix)

# Export/import the local ledger file.
shutil.copyfile("corpus-ledger.bin", "customer-export.ufmr")
shutil.copyfile("customer-export.ufmr", "restored-ledger.bin")

Universal and decision pipelines

UniversalPipeline is the governed data processing path. DecisionPipeline is the text decision/audit path with anti-drift gates and a persistent audit ledger.

import ufm

up = ufm.UniversalPipeline(
    storage_path="universal-ledger.bin",
    bit_depth=21,
    verify=True,
    zero_point=False,
)
run = up.run(b"payload for the governed pipeline")
print(run["success"], run["replay_valid"], run["quality"])
print(run["stages_completed"])

dp = ufm.DecisionPipeline("decision-ledger.jsonl")
decision = dp.run("What is structural identity in UFM?")
print(decision["status"])
print(decision["decision_hash"])
print(decision["gates"])

Call Ben locally

Backend options

• Ollama: local model server. Free, runs offline once a model is pulled. --backend ollama --model llama3.1 (or any model you have pulled).
• OpenAI: your OpenAI key. --backend openai --model gpt-4o --api-key sk-...
• Anthropic: your Anthropic key. --backend anthropic --model claude-sonnet-4-20250514 --api-key sk-ant-...
• Subprocess: pipe to any command. Use this when an AI agent (Claude Code, Cursor, etc.) is the LLM and you do not want to attach a separate API key. The agent writes a short bridge script that talks to its own LLM. Details below.

Python form:

import os
import ufm

# Pick one backend.
backend = ufm.backend_from_config(backend="ollama", model="llama3.1")
# backend = ufm.backend_from_config(backend="anthropic",
#     model="claude-sonnet-4-20250514", api_key=os.environ["ANTHROPIC_API_KEY"])
# backend = ufm.backend_from_config(backend="openai",
#     model="gpt-4o", api_key=os.environ["OPENAI_API_KEY"])

session = ufm.BenSession(backend=backend)

first = session.ask("What is UFM actually doing?")
print(first.text)
print(first.session_id, first.turn_count)

second = session.ask("How does replay relate to structural identity?")
print(second.text)
print(session.history())

# One-shot convenience wrapper with a provider API key.
answer = ufm.ask_ben(
    "Explain the replay invariant.",
    backend="anthropic",
    model="claude-sonnet-4-20250514",
    api_key=os.environ["ANTHROPIC_API_KEY"],
)
print(answer["text"])

CLI form:

python -m ufm ask-ben "What is UFM?" --backend ollama --model llama3.1
python -m ufm ask-ben "Explain replay" --backend openai --model gpt-4o --api-key YOUR_OPENAI_API_KEY
python -m ufm ask-ben "Explain replay" --backend anthropic --model claude-sonnet-4-20250514 --api-key YOUR_ANTHROPIC_API_KEY

Subprocess backend (for AI agents and custom LLM gateways)

The subprocess backend pipes the request to a command you specify and reads the response from its stdout. This lets an AI agent (or any environment that already has LLM access) answer Ben's prompts without Ben needing a separate API key.

Wire format:

• Stdin (JSON): {"messages":[{"role":"...","content":"..."}, ...], "temperature":0.0, "max_tokens":2048}
• Stdout: either plain text (returned as the answer) or JSON {"text":"...", "model":"..."}. Both content on the way in and text on the way out are UTF-8. Embed any unicode you like (e.g. Θ ∇Ψ ∆).
• Exit code: non-zero is treated as an error; stderr is surfaced to the caller.

Minimal bridge example. Read messages, call your LLM, print response:

# bridge.py: reference bridge for the subprocess backend.
import json
import sys

def call_llm(messages: list[dict]) -> str:
    # Replace this with your actual LLM call. The bridge is whatever you
    # already have access to: an in-house gateway, an SDK, an agent's own
    # LLM, etc.
    last_user = next((m["content"] for m in reversed(messages)
                      if m["role"] == "user"), "")
    return f"replace me with a real LLM call. you asked: {last_user}"

payload = json.loads(sys.stdin.read())
print(json.dumps({"text": call_llm(payload["messages"])}))

Windows reference bridge: Claude Code as the LLM

If your environment already has Claude Code installed, you can use it as the LLM with no extra API key. The script below is the battle-tested Windows form. It dodges three traps that the generic bridge above hits on Windows:

• Claude Code is shipped as a .cmd shim on Windows; subprocess.run(["claude", ...]) can't find it without help. Resolve via shutil.which("claude").
• Ben's sealed system prompt is > 32 KB, which blows past the Windows argv limit. Write the system prompt to a temp file and pass --system-prompt-file instead of --system-prompt.
• A spawned claude -p inherits its parent CWD's CLAUDE.md, which contaminates Ben's reply with project-specific context. Run the spawn with cwd= set to a clean tempdir.

# bridge-claude-code.py: Windows reference bridge for the subprocess backend.
#
# Three Windows-specific gotchas this handles for you:
#   1. "claude" is a .cmd shim, so resolve with shutil.which() before exec.
#   2. Ben's sealed system prompt > 32 KB blows the Windows argv limit;
#      write it to a temp file and pass --system-prompt-file.
#   3. Spawned claude -p inherits CWD's CLAUDE.md and gets contaminated;
#      cwd=tempdir keeps the spawn clean.
#
import json
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

CLAUDE_BIN = shutil.which("claude")
if CLAUDE_BIN is None:
    print("claude CLI not on PATH", file=sys.stderr)
    sys.exit(2)

def split_messages(messages: list[dict]) -> tuple[str, str]:
    """Concatenate system messages and the rest separately."""
    sys_parts = [m["content"] for m in messages if m.get("role") == "system"]
    rest = "\n\n".join(
        f"{m['role'].upper()}: {m['content']}"
        for m in messages
        if m.get("role") != "system"
    )
    return "\n\n".join(sys_parts), rest

def call_claude(system_prompt: str, user_text: str) -> str:
    with tempfile.TemporaryDirectory(prefix="ufm-bridge-") as tmp:
        tmp_path = Path(tmp)
        sys_file = tmp_path / "system.txt"
        sys_file.write_text(system_prompt, encoding="utf-8")
        # Lock the spawn down: no tools, clean cwd, file-based system prompt.
        result = subprocess.run(
            [
                CLAUDE_BIN, "-p",
                "--system-prompt-file", str(sys_file),
                "--allowedTools", "",   # Ben/Bob need an LLM, not tool access.
                "--output-format", "text",
            ],
            input=user_text,
            capture_output=True,
            text=True,
            encoding="utf-8",
            cwd=str(tmp_path),
            timeout=180,
            check=False,
        )
        if result.returncode != 0:
            stderr = (result.stderr or "").strip()[-2000:]
            raise RuntimeError(f"claude exited {result.returncode}: {stderr}")
        return (result.stdout or "").strip()

payload = json.loads(sys.stdin.read())
system_prompt, user_text = split_messages(payload["messages"])
answer = call_claude(system_prompt, user_text)
print(json.dumps({"text": answer, "model": "claude-code"}))

Then call Ben/Bob through whichever bridge you wrote:

python -m ufm ask-ben "What is UFM?" \
  --backend subprocess \
  --cmd "python bridge.py"

# Or with the Windows Claude Code bridge:
python -m ufm ask-ben "What is UFM?" \
  --backend subprocess \
  --cmd "python bridge-claude-code.py"

python -m ufm ask-bob "Explain C-CORE-002" \
  --backend subprocess \
  --cmd "python bridge.py"

Python form, useful when you want to keep a session across turns:

import ufm

backend = ufm.SubprocessBackend(cmd=["python", "bridge.py"], timeout=180)
session = ufm.BenSession(backend=backend)

print(session.ask("What is UFM actually doing?").text)
print(session.ask("How does replay relate to structural identity?").text)

Call Bob locally

Bob is also included in the wheel. The sealed corpus, claim-gate snapshot, OOV thresholds, and prompts ship with the package. Bob can run corpus retrieval and gate/OOV/audit output without an LLM backend; pass a backend when you want generated response wording.

import os
import ufm

# Corpus retrieval, gate status, OOV metrics, and audit output.
bob = ufm.BobPipeline()
result = bob.query("What is the replay invariant?", mode="advisory", max_anchors=5)
print(result.response)
print(result.gate_status)       # PASS, WARN, or BLOCK
print(result.evidence)
print(result.oov)
print(result.audit)
print(result.boundary_flags)

# Optional generated answer using an LLM backend.
backend = ufm.backend_from_config(
    backend="anthropic",
    model="claude-sonnet-4-20250514",
    api_key=os.environ["ANTHROPIC_API_KEY"],
)
bob_with_llm = ufm.BobPipeline(backend=backend)
print(bob_with_llm.query("Explain C-CORE-002.").response)

# One-shot convenience wrapper.
one_shot = ufm.ask_bob(
    "Is replay identity verified?",
    backend="ollama",
    model="llama3.1",
)
print(one_shot["response"])
print(one_shot["gate_status"])

Same four backend options as Ben. See backend options above. The subprocess backend works identically here for AI agents and custom LLM gateways.

CLI form:

python -m ufm ask-bob "Is the replay invariant verified?" --backend ollama --model llama3.1
python -m ufm ask-bob "Explain C-CORE-002" --backend anthropic --api-key YOUR_ANTHROPIC_API_KEY
python -m ufm ask-bob "Explain C-CORE-002" --backend subprocess --cmd "python bridge.py"

Local equivalents at a glance

Operational notes

• Keep one ledger file per project or tenant. Primitive reuse is scoped to that file.
• Do not run multiple writers against the same ledger path without your own lock.
• Ledger paths must stay inside the current working directory; paths containing .. are rejected.
• Call ufm.set_num_threads(n) before the first ingestion if you need to bound CPU use.
• Provider-backed Ben/Bob calls require the provider SDK, for example pip install openai or pip install anthropic.

Authentication

All API endpoints (except /v1/health) require authenticated context. Programmatic clients should pass an API key in theX-Api-Key header. Browser dashboard sessions can use an access_token cookie.

Getting your API key

Go to API Keys in the sidebar, click Create Key, and copy the full key immediately. It is only shown once.

Using your API key

curl -H "X-Api-Key: ufm_live_a1b2c3d4.your_secret_here" \
  -H "Content-Type: application/json" \
  -X POST https://api.spectrengine.com/v1/process \
  -d '{"data_b64": "..."}'

Key format

Keys follow the format ufm_live_{public_id}.{secret}. The prefix identifies the key; the secret authenticates it. Both parts are required.

Licences

Licences are signed 30-day tokens that unlock tier-scoped features such as ben.ask and bob.query. The local wheel caches the signed token during python -m ufm activate, then checks it locally on normal engine calls.

POST /v1/licences/verify

Mint or refresh a signed licence token for the caller's API key. Authenticated with X-Api-Key (not JWT). Rate-limited to 60 requests per hour.

Response:

• token.payload.public_id (string) Public id from the API key used for activation
• token.payload.scopes (string[]) Granted local feature scopes, for example ben.ask and bob.query
• token.payload.tier (string) Customer tier
• token.payload.expires_at (string (ISO 8601)) Token expiry timestamp
• token.signature (string) Base64 signature over the payload

GET /v1/licences/me

Return the caller's current licence status without minting a new token.

• active (boolean) True iff a non-revoked, non-expired licence exists
• tier (string) Subscription tier (e.g. standard)
• expires_at (string (ISO 8601) or null) Expiry timestamp of the active licence, or null if none exists
• revoked (boolean) True if the licence has been administratively revoked

POST /v1/licences/revoke/{licence_id}

Admin-only. Revoke a specific licence by its UUID. Returns { revoked: true, licence_id }.

Python example

import os
import requests

# X-Api-Key authentication, not JWT
headers = {"X-Api-Key": os.environ["SPECTR_API_KEY"]}
resp = requests.post(f"{BASE}/v1/licences/verify", headers=headers)
token = resp.json()["token"]

print(token["payload"]["tier"])
print(token["payload"]["scopes"])

status = requests.get(f"{BASE}/v1/licences/me", headers=headers).json()
print(status["active"], status["expires_at"])

Error Handling

All errors return a consistent JSON structure:

{
  "error": {
    "code": "ERROR_CODE",
    "message": "Human-readable description",
    "request_id": "correlation-uuid"
  }
}

Error codes

Handling errors in code

import requests

BASE    = "https://api.spectrengine.com"
API_KEY = "ufm_live_a1b2c3d4.your_secret_here"

resp = requests.post(f"{BASE}/v1/process", json=payload,
                      headers={"X-Api-Key": API_KEY})

if resp.status_code == 200:
    result = resp.json()
elif resp.status_code == 429:
    retry_after = resp.headers.get("Retry-After", 60)
    print(f"Rate limited. Retry in {retry_after}s")
else:
    error = resp.json().get("error", {})
    print(f"Error {resp.status_code}: {error.get('message')}")
    print(f"Request ID: {error.get('request_id')}")