attesto 0.4.0

Attesto AI Python SDK — log verifiable AI events to Polygon via one function call.

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.

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.2.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.