Metadata-Version: 2.4
Name: attesto
Version: 0.3.0
Summary: Attesto AI Python SDK — log verifiable AI events to Polygon via one function call.
Author-email: Attesto <info@attesto.eu>
License-Expression: Apache-2.0
Project-URL: Homepage, https://attesto.eu
Project-URL: Documentation, https://docs.attesto.eu/manuals/sdks.html
Project-URL: Security, https://attesto.eu/security
Keywords: ai,compliance,eu-ai-act,polygon,merkle,audit
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.27
Requires-Dist: pydantic>=2
Requires-Dist: cryptography>=42
Provides-Extra: dev
Requires-Dist: build>=1.2; extra == "dev"
Requires-Dist: pytest>=8; extra == "dev"
Requires-Dist: pytest-asyncio>=0.24; extra == "dev"
Requires-Dist: pytest-httpx>=0.35; extra == "dev"
Requires-Dist: ruff>=0.7; extra == "dev"
Dynamic: license-file

# Attesto Python SDK

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

```bash
pip install attesto
```

## Quick start

```python
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)
```

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

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

```python
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:

```python
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.

## 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:

```python
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.

```python
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.

```python
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

```python
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

```python
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

```python
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:

```python
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

```python
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

```python
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.
