Attesto Python SDK

Log every AI decision to a verifiable, on-chain audit trail with one call.

pip install attesto

Quick start

from datetime import UTC, datetime
from attesto import AttestoClient

attesto = AttestoClient(api_key="atto_live_...")  # issued when you register a system
ack = attesto.log_event(
    type="inference",
    status="verified",
    ts=datetime.now(UTC),
    latency_ms=42,
    input_hash="sha256:deadbeef...",
    output_hash="sha256:cafebabe...",
    payload={"score": 0.87, "model": "gpt-4o"},
)
print(ack.id, ack.system_id, ack.ts)

Runnable examples

examples/ — first attested event, offline receipt verification, completeness check. Each runs against real Attesto with ATTESTO_API_KEY set, or fully offline against the bundled emulator without it (CI runs them that way so they can't rot).

Canonicalization is specified normatively in ATTESTO-CANONICAL-JSON-001; the parity corpus golden-vectors/sdk-parity/ is its conformance set.

Committed payload number rule

When events are committed to a Proofstream, payload and metadata numbers must serialize identically across Python, Go, and JavaScript. Non-integer numbers and integers beyond ±(2^53−1) are rejected at ingestion (HTTP 422); encode decimals and large integers as strings (e.g. {"score": "0.87"}). This keeps cross-language commitment recomputation byte-exact.

The SDK enforces the same rule locally before sending, so you see it at dev time rather than as a production 422. log_event / log_events raise AttestoUnsafeNumberError (with .path, the JSON path to the offending value). Pass preflight=False to defer entirely to the server.

from attesto import payload_commitment, verify_payload_commitment

# Compute the commitment a Proofstream stores for a payload, byte-identical to
# the server (and to the Go / TypeScript SDKs):
payload_commitment({"decision": "approve", "score_bp": 8700})
# -> {"hash_alg": "sha256", "canonical_payload_hash": "..."}

# Recompute it from your own copy of the payload and compare to a fetched event:
verify_payload_commitment(my_payload, event)  # -> True / False

Verify receipts offline

verify_receipt checks a receipt entirely on your machine — it recomputes the domain-separated hash and verifies the Ed25519 signature with no call back to Attesto. This is the point of the SDK: you do not ask the party you are distrusting whether to trust it. (Requires cryptography>=42, a dependency of this package.)

from attesto import verify_receipt

report = verify_receipt(receipt, public_key_hex=signer_public_key_hex)
report.ok        # True only when no problems were found
report.problems  # () or e.g. ("receipt signature mismatch",)

AttestoV2Client.verify_receipt(...) is the server-assisted (remote) check and is kept for compatibility; prefer the offline function above when you have the signer's public key.

Verify inclusion, checkpoints, and completeness

The same offline trust model extends across the whole proof chain — all client-side, no calls to Attesto:

from attesto import (
    verify_inclusion_proof,     # an event is in a window root
    verify_checkpoint_root,     # window hashes fold to the checkpoint root
    verify_checkpoint_extension,  # one checkpoint continues the previous
    verify_completeness,        # no events were omitted in a range
)

verify_inclusion_proof(leaf_hash=leaf, proof=proof, root_hash=window_root)  # bool
verify_checkpoint_root(window_hashes, checkpoint_root)                       # bool
verify_checkpoint_extension(previous_checkpoint, current_checkpoint).ok      # bool
verify_completeness(events, from_seq_no=5, to_seq_no=8).ok                   # bool

verify_completeness proves no events were omitted in [from_seq_no, to_seq_no]: the sequence numbers must be gap-free and each event's prev_event_hash must chain to the previous event's event_hash. Omission matters to auditors as much as tampering, and the per-stream hash chain gives it for free.

Provenance verification (Attesto 3)

attesto.provenance is the verification client for the Attesto 3 provenance lane: commitment-only provenance streams fed by a customer-controlled Local Vault. The vault's pinned Rust edge core builds capsules, commitments and signatures; this module re-derives and checks them. It deliberately cannot construct a capsule or a randomizer — outside the Local Vault there is nothing legitimate to commit. Every function below runs offline: no network, no call to Attesto. Protocols: ATTESTO-PROVENANCE-001, ATTESTO-DISCLOSURE-001, ATTESTO-ZK-RANGE-001, and the bundle provenance binding of ATTESTO-PROOFSTREAM-001 (ADR-0014). Conformance is pinned by golden-vectors/provenance-v0.1-dev/ and golden-vectors/zk-range-v0.1-dev/.

Every report carries a not_claimed list. Print it next to the result: a verifier that shows only what it checked leaves the reader to assume the picture is complete.

Verify a disclosure presentation offline

A holder's Local Vault issues a selective-disclosure presentation: the revealed leaves with their randomizers, two-hop inclusion proofs to the capsule root, and an Ed25519 signature by the issuing installation. verify_disclosure opens each revealed commitment and replays each proof to the presented capsule_root.

import json
from attesto.provenance import verify_disclosure

presentation = json.load(open("presentation.json"))
report = verify_disclosure(
    presentation,
    expected_nonce=challenge,            # the nonce you issued; omit for bounded-lifetime mode
    subject_commitment=asset_commitment,  # the asset you hold; omit if you hold none
)
report.ok               # True only when every leaf opened and every proof folded to capsule_root
report.capsule_root     # str | None
report.verified_leaves  # ({"subtree": "claims", "leaf_role": "c2pa_manifest_valid", "value": ...}, ...)
report.freshness        # "challenge" when expected_nonce was given, else "bounded_lifetime"
report.subject_checked  # True only when subject_commitment matched the presentation's binding
report.problems         # () or every problem found, e.g. ("disclosure has expired",)
report.not_claimed      # ({"id": "undisclosed_facts_absent", "statement": ...},)

Signature: verify_disclosure(presentation, *, expected_nonce=None, subject_commitment=None, now=None) -> DisclosureReport. Problems are collected, not raised, so a caller sees everything wrong with a presentation. Without expected_nonce the presentation is bounded only by its expires_at, and the report says bounded_lifetime rather than implying a freshness it did not check. Non-claim: the disclosure proves the revealed leaves are in the capsule; it is not a statement that the capsule holds nothing else.

Verify bundle provenance inclusion and key revocation offline

A verifier bundle over a provenance stream carries provenance_root, provenance_event_count and vault_key_lifecycle inside its hashed payload (ADR-0014). An inclusion object — from GET /v2/streams/{stream_id}/provenance-events/{source_ref}/bundle-inclusion?from_checkpoint_id=...&to_checkpoint_id=..., or carried as provenance_inclusions next to the bundle — proves one capsule root under that root. verify_bundle_provenance recomputes the bundle hash, re-hashes the leaf from its fields, replays the proof, takes the receipt time for the leaf's seq_no from the bundle's own receipts, and applies the frozen revocation rule to the installation the leaf names.

import json
from attesto.provenance import verify_bundle_provenance

bundle = json.load(open("bundle.json"))
inclusion = json.load(open("inclusion.json"))   # the "inclusion" field of the bundle-inclusion response
report = verify_bundle_provenance(bundle, inclusion)
report.ok               # bundle hash holds, inclusion VALID, and the key was live at receipt
report.inclusion        # "VALID" | "INVALID"
report.key_status       # "valid" | "revoked_at_receipt" | "unknown_installation" | "not_evaluated"
report.flags            # e.g. ("revoked_at_receipt", "suspect_backdated")
report.seq_no, report.installation_id, report.capsule_root
report.receipt_time     # platform issued_at of that seq_no, from the bundle's receipts
report.revoked_at       # from vault_key_lifecycle, or None
report.problems         # () or every problem found
report.not_claimed      # three statements, see below

inclusion and key_status are separate on purpose: a capsule root can be provably under the bundle while its key was revoked before receipt. not_evaluated means the bundle did not give enough to say (no receipt for that seq_no, or a malformed lifecycle entry); it is never a pass.

The revocation rule is available on its own, and mirrors the platform's ingest path exactly:

from attesto.provenance import evaluate_key_revocation

verdict = evaluate_key_revocation(
    revoked_at="2026-08-18T12:40:00.000Z",        # from GET .../key-status or vault_key_lifecycle; None = never revoked
    receipt_time=receipt["payload"]["issued_at"],  # platform receipt time, never the vault's occurred_at
    claimed_occurred_at=envelope["occurred_at"],
    reason="key_compromise",
)
verdict.status      # "valid" | "revoked_at_receipt"
verdict.flags       # ("revoked_at_receipt", "suspect_backdated") when the claim predates the revocation
verdict.accepted    # status == "valid"
verdict.instant_reconstructed  # True when reason == "unrecorded": the instant was reconstructed by migration

The boundary is inclusive — an event receipted exactly at the revocation instant is revoked. Non-claims carried by BundleProvenanceReport.not_claimed: the root proves the capsule root existed under the bundle and nothing about the capsule's contents (the platform never opens one); revocation is evaluated against platform receipt time, never the vault-claimed occurred_at; the lifecycle is as of bundle build, so a revocation recorded later is not in the bundle.

Derive effective assurance (L3 is derived, never signed)

A vault signs L0, L1 or L2 in its envelope. L3 has no on-wire representation: it is derived by the verifier from L2 plus a met witness quorum on the containing checkpoint, and an envelope claiming L3 is refused.

from attesto.provenance import effective_assurance

report = effective_assurance("L2", witness_quorum_met=True, anchor_confirmed=True)
report.effective           # "L3"
report.derived             # True: derived here, signed by nobody
report.vault_assurance     # "L2"
report.witness_quorum_met  # True | False | None (None = not evaluated)
report.anchor_confirmed    # True | False | None
report.reasons             # ("anchor confirmed; anchoring does not promote assurance", "L3 derived from L2 plus a met witness quorum")

effective_assurance("L1", witness_quorum_met=True).effective   # "L1": only L2 can become L3
effective_assurance("L2").effective                            # "L2": quorum not evaluated withholds L3
effective_assurance("L3")                                      # raises AttestoProvenanceError

Levels: L0 software-held key; L1 hardware-held key (PKCS#11 token, non-extractable); L2 hardware-held key plus a TPM 2.0 quote over the vault's measurement; L3 = L2 and the containing checkpoint met the witness quorum. Anchoring is reported alongside and never promotes a level. The platform refuses an L1/L2 envelope it cannot substantiate from a registered attestation (provenance_assurance_not_substantiated), so a level that reaches a verifier was checked at receipt; the four facts stay separate in the report because ADR-0010 forbids collapsing them into one badge.

Verify an exact private-numeric opening (optional curve extra)

A private numeric claim commits its encoded value as a Pedersen commitment C = v·B + r·H over ristretto255. When a holder reveals v and the blinding r, verify_pedersen_opening recomputes C and compares it byte for byte. Curve arithmetic is an optional extra because the rest of verification is SHA-256 and Merkle work:

pip install 'attesto[zk]'   # pysodium / libsodium

from attesto.provenance import pedersen_available, verify_pedersen_opening

if pedersen_available():
    opened = verify_pedersen_opening(descriptor, encoded_value, blinding_scalar)  # bool
else:
    opened = None  # report "not_checked": this installation did not check, not "nobody can"

descriptor is the claim's private-numeric descriptor (its pedersen.commitment and encoding bounds); encoded_value is the canonical encoded integer (a detector score of exactly 0.0 encodes as 0, which is a legal opening); the blinding is 32 bytes hex. The descriptor must already have opened its claim leaf — verify it under the capsule root first, otherwise a matching pair can be fabricated whole. A value outside semantic_min_encoded..semantic_max_encoded returns False. Without the extra the function raises AttestoProvenanceError; check pedersen_available() first. This does not verify a range proof.

ZK range results: inspect, never pass through

Selective disclosure v2 proves that a named detector's measurement fell inside an interval (ATTESTO-ZK-RANGE-001) without revealing it. No SDK verifies the bulletproof; that remains the Rust core's job. What the SDK does is refuse to launder the issuer's word as its own:

from attesto.provenance import inspect_predicate_result, validate_range_statement

report = inspect_predicate_result(result, capsule_inclusion=None)
report["verified_here"]       # {"zk_predicate": "not_checked", "capsule_inclusion": "not_checked" | "verified" | "failed"}
report["reported_by_issuer"]  # whatever the issuer claimed, kept apart from what was checked here
report["not_claimed"]         # detector_correctness_not_proven, content_truth_not_proven, ai_generation_not_proven

width = validate_range_statement(statement)   # 8 | 16 | 32 | 64, derived from the public bounds

inspect_predicate_result raises on a result that drops one of the three required non-claims or carries a verdict-shaped field (ai_generated, synthetic, is_fake, authentic, confidence, score, probability): a proven bound says nothing about whether the content is machine-generated.

Provider results and content marks

Capsule evidence from an AttestoMark Image/Audio/Video provider is an ATTESTO-PROVIDER-RESULT-001/0.2 object carrying presented_matches_record (ADR-0015): whether the bytes presented to detection are the exact asset that was marked. A mismatch is an observation an honest transcode also produces, never a refusal. A detected mark identifies a registration; it is not evidence of authorship, AI origin, ownership or truth. See attesto-edge-provider-001.

What is not in this SDK

• No capsule construction, randomizer generation or envelope signing — those live in the Local Vault's pinned edge core.
• No range-proof verification (bulletproofs); only the Rust core verifies one.
• No client wrapper yet for POST /v2/provenance/streams; create provenance streams over plain HTTP with a system API key.
• The attesto CLI does not verify disclosures or bundle provenance in 0.5.0.

Typed compliance events, one-line attestation, and the evidence report

from attesto import ModelDecision, attest, session, article12, payload_commitment

@attest(client, stream_id="str_...")          # commitments over args + result;
def approve_loan(application: dict) -> dict:  # failures log-and-continue
    ...

with session(client, stream_id="str_...", actor_ref="op_jerome") as s:
    s.log(ModelDecision(model="credit-v1", decision="approve",
                        input_commitment=payload_commitment(app_data),
                        confidence_bp=8700, human_in_loop=True))

print(article12(client, "str_..."))   # deterministic Markdown evidence report

Typed events (ModelDecision, HumanOverride, IncidentReport, DataAccess) carry regulation_refs (EU AI Act / NIS2 / GDPR) and self-validate against the number policy. The report states what is recorded and independently verifiable — it never asserts conformity.

Testing without Attesto: MockAttesto

attesto.testing.MockAttesto is a local in-memory emulator with real hash-chain semantics (same canonical functions) and receipts signed by a per-instance throwaway key — run your full ingest-and-verify pipeline in CI with zero network and zero account:

from attesto import AttestoV2Client, MemoryHeadStore, verify_receipt
from attesto.testing import MockAttesto

with MockAttesto() as mock:
    client = AttestoV2Client(mock.api_key, base_url=mock.base_url,
                             head_store=MemoryHeadStore())
    stream = client.create_stream(use_case="ci", policy_id="mock-policy")
    receipt = client.log_event(stream_id=stream.stream_id,
                               source_ref="e1", payload={"n": 1})
    stored = client.get_receipt(receipt.stream_event_id)
    assert verify_receipt(stored.receipt.model_dump(by_alias=True),
                          public_key_hex=mock.public_key_hex).ok

Mock evidence can never pass as real: every object carries "mock": true, the signer kid is attesto-mock-ed25519, and verification against any real witness key fails.

Built-in self-test and doctor

On the first hashing operation per process the SDK verifies itself against a vendored ~1.7 KB copy of the cross-language parity vectors and raises AttestoSelfTestError on any divergence — a corrupted install can never silently produce wrong evidence. attesto.doctor() returns a deterministic report dict (self-test, Ed25519 availability, number-policy dry-run on your sample payload, head-store writability, and — with an API key — reachability, protocol acceptance, and clock skew).

Iterating long listings

Every list_* method has an iter_* twin that walks limit/offset pages transparently and stops on the first short page:

for event in attesto.iter_tenant_stream_events("str_...", page_size=200):
    process(event)

Verify anchors on-chain

verify_anchor_onchain checks an anchor epoch against the chain itself — one raw JSON-RPC eth_call to the anchoring contract's getCommitment(batchId) (comparing the on-chain merkle root) plus a transaction-receipt check (status, block). No web3 dependency; the RPC endpoint is yours, so this never asks Attesto to confirm Attesto.

from attesto import verify_anchor_onchain

anchor = attesto.get_anchor_epoch("aep_...")
report = verify_anchor_onchain(
    anchor_epoch=anchor.model_dump(by_alias=True),
    rpc_url="https://polygon-rpc.example",  # your own RPC endpoint
)
assert report.ok, report.problems

Your SDK is a witness

The client remembers the last accepted (seq_no, event_hash) per stream and checks every new receipt links forward. If the server ever rewinds a sequence number or presents a divergent lineage, your own machine catches it — log_event / log_events raise AttestoForkDetected and the stored head is not advanced. By default this persists to ~/.attesto/heads.json (atomic, 0600), so fork detection survives restarts.

from attesto import AttestoV2Client, MemoryHeadStore, FileHeadStore

# Default: file-backed, survives restarts.
attesto = AttestoV2Client.with_bearer_token(token)

# Or choose a store explicitly; head_store=None disables fork detection.
attesto = AttestoV2Client(token, head_store=FileHeadStore("/var/lib/app/heads.json"), _validate_key=False)
attesto = AttestoV2Client(token, head_store=MemoryHeadStore(), _validate_key=False)

Async

from datetime import UTC, datetime
from attesto import AsyncAttestoClient

async with AsyncAttestoClient(api_key="atto_live_...") as attesto:
    ack = await attesto.log_event(
        type="inference",
        ts=datetime.now(UTC),
        payload={"score": 0.87},
    )

Proofstream v2

import os
from datetime import UTC, datetime
from attesto import AttestoV2Client

with AttestoV2Client(api_key="atto_live_...") as attesto:
    receipt_signer_public_key_hex = os.environ["ATTESTO_RECEIPT_SIGNER_PUBLIC_KEY_HEX"]
    stream = attesto.create_stream(
        use_case="ai-decision-history",
        policy_id="policy-2026-01",
    )
    receipt = attesto.log_event(
        stream_id=stream.stream_id,
        source_ref="upstream-event-123",
        event_type="decision",
        occurred_at=datetime.now(UTC),
        payload={"decision": "approve", "score": 91},
    )
    batch = attesto.log_events(
        stream.stream_id,
        [
            {
                "source_ref": "upstream-event-124",
                "occurred_at": datetime.now(UTC),
                "payload": {"score": 88},
            },
            {
                "source_ref": "upstream-event-125",
                "event_type": "decision",
                "occurred_at": datetime.now(UTC),
                "payload": {"decision": "review"},
            },
        ],
    )
    assert batch.accepted == 2
    stored = attesto.get_receipt(receipt.stream_event_id)
    report = attesto.verify_receipt(
        receipt=stored.receipt,
        public_key_hex=receipt_signer_public_key_hex,
        stream_event_id=receipt.stream_event_id,
    )
    assert report.ok

    consistency = attesto.get_checkpoint_consistency(
        "chk_current",
        from_checkpoint_id="chk_previous",
    )
    assert consistency.step_count >= 1

    policy = attesto.get_witness_policy("policy-ai-credit-v1")
    assert policy.policy_hash

    # Bundle export is intentionally stricter than receipt ingest: every
    # checkpoint in the selected range must already have witness quorum
    # evidence and a confirmed anchor epoch.
    bundle = attesto.build_verifier_bundle(
        from_checkpoint_id="chk_previous",
        to_checkpoint_id="chk_current",
    )
    assert bundle.bundle_hash

    # Present only after the checkpoint has confirmed on-chain.
    anchor = attesto.get_anchor_epoch("aep_...")
    assert anchor.status == "confirmed"

    offline = attesto.verify_object(kind="bundle", proof_object=bundle.bundle)
    assert offline.ok

AttestoV2Client talks to the production /v2/streams, /v2/streams/{stream_id}/events, /v2/streams/{stream_id}/events/batch, /v2/receipts, /v2/windows, /v2/checkpoints, /v2/checkpoints/{checkpoint_id}/consistency, /v2/witness/policies/{policy_id}, /v2/anchors/{anchor_epoch_id}, /v2/ivc/epochs/{ivc_epoch_id}, /v2/audit/packs, and /v2/verify APIs. Single and batch writes both return signed receipts. It exposes witness policy and review-gated IVC epoch visibility. Receipt ingest can run before enforced rollout gates; verifier-bundle export requires witnessed and confirmed anchored checkpoints. Nova proof production remains review-gated until that rollout gate is enabled.

Receiving Attesto webhooks

from attesto import verify_webhook

@app.post("/attesto-webhook")
def handle(request):
    if not verify_webhook(body=request.body, headers=request.headers, secret=WEBHOOK_SECRET):
        return Response(status=401)
    process(request.json())

Verification recomputes hmac_sha256(secret, f"{timestamp}.{body}") from the X-Attesto-Timestamp / X-Attesto-Signature headers, rejects timestamps more than 300 s from now (replay protection), and compares in constant time.

Signed webhook connectors

Use the connector helper when an external source posts to a signed-webhook connector endpoint:

import json
from attesto import signed_connector_webhook_headers

body = json.dumps({"sourceRef": "evt_123"}, separators=(",", ":")).encode()
headers = signed_connector_webhook_headers(connector_secret, body)

The helper signs timestamp + "." + raw_body_bytes and returns the exact X-Attesto-Connector-* headers expected by /v2/connectors/signed-webhooks/{connectorId}/events.

Batching

attesto.log_events([
    {"type": "inference", "latency_ms": 40},
    {"type": "inference", "latency_ms": 33},
    {"type": "decision", "status": "pending", "payload": {"threshold": 0.7}},
])

Up to 1000 events per batch. The Attesto worker then groups them into a Merkle tree and commits the root on Polygon mainnet within your tenant's configured cadence (6h / 1h / per-event).

Source time

Attesto stores source-system time separately from backend ingest time. ts and Proofstream occurred_at must be timezone-aware. The Python SDK defaults them to datetime.now(UTC) from the running source process when omitted, but production integrations should pass the real upstream event timestamp whenever the source system provides one.

Configuration

arg	default	purpose
api_key	—	Required. Must match atto_live_<32 lowercase hex chars> or atto_test_<32 lowercase hex chars>.
base_url	https://verify.attesto.eu	Public Attesto API origin. Override only for private/staging deployments.
timeout_s	10.0	Per-request timeout.
max_retries	3	Retries on 5xx / 429 / transport errors, with jittered exponential backoff.
user_agent	attesto-python/0.5.0	Sent as the UA header.

Error handling

from attesto import (
    AttestoClient,
    AuthError,
    RateLimitError,
    ServerError,
    ValidationError,
)

try:
    attesto.log_event(type="inference")
except AuthError as exc:        # 401 / 403 — bad key
    ...
except RateLimitError as exc:   # 429 — exceeded tenant rate limit
    ...
except ValidationError as exc:  # 4xx payload problem
    ...
except ServerError as exc:      # 5xx after all retries exhausted
    ...

All Attesto SDK exceptions expose status and detail when the server returned an HTTP response. Transport failures keep both as None.

What you get

Every event:

1. Canonicalised to byte-exact JSON (sort keys, no whitespace, ASCII-safe).
2. SHA-256 hashed into a Merkle leaf.
3. Batched with other events at your cadence.
4. The Merkle root is committed on-chain via APSProvenance.
5. Every anchored event gets a tenant-authenticated proof from GET /v1/events/{id}/proof. The proof payload contains canonicalJson, proof, and batchId; submit those fields to POST https://verify.attesto.eu/v1/public/verify or paste them into the /verify page for independent verification.

You never handle keys, wallets, or gas — Attesto pays the gas and handles the on-chain flow.

Production behavior

• Defaults to https://verify.attesto.eu; override base_url only for private or staging deployments.
• Use this SDK from server-side code only. Attesto system API keys are bearer secrets and must never be embedded in browser bundles, mobile apps, logs, or client-visible environment variables.
• Validates the key shape locally before making network calls. Production system keys are shown once when the system is registered in Attesto.
• Validates base_url locally and accepts only http or https origins.
• Adds an Idempotency-Key header automatically for single-event and batch writes.
• Retries transient 429, 5xx, and transport failures with exponential backoff.
• Caps batch ingestion at 1000 events per request.
• Never handles wallets, private keys, or gas in application code.