attesto 0.1.2

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 attesto import AttestoClient

attesto = AttestoClient(api_key="atto_live_...")  # issued when you register a system
ack = attesto.log_event(
    type="inference",
    status="verified",
    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)

Async

from attesto import AsyncAttestoClient

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

Proofstream v2

import os
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",
        payload={"decision": "approve", "score": 91},
    )
    batch = attesto.log_events(
        stream.stream_id,
        [
            {"source_ref": "upstream-event-124", "payload": {"score": 88}},
            {
                "source_ref": "upstream-event-125",
                "event_type": "decision",
                "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.

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

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