attestify-sdk — Python

Add governed lane execution to any Python codebase in a few lines.

Install

pip install attestify-sdk

Or directly from the repo while the package is in early access:

pip install "git+https://github.com/attestifyagent/attestify-os.git#subdirectory=sdk/python"

Quick start

import os
from attestify import create_client

client = create_client(api_key=os.environ["ATTESTIFY_API_KEY"])

result = client.run_lane(
    input="Summarise the latest advances in quantum error correction.",
    lane_id="researcher-v2",
    session_id="my-session-001",
    options={"verify": True, "write_memory": True},
)

print(result.output)             # lane response text
print(result.receipt.run_id)     # unique run identifier
print(result.receipt.evidence)   # full EvidenceBundle dataclass
print(result.receipt.pricing)    # cost breakdown

Features

• create_client(api_key) — configure once, use anywhere
• client.run_lane(input, ...) — single call handles sessions, retries, idempotency
• Typed receipts — RunReceipt and EvidenceBundle dataclasses with full field coverage
• Auto-routing — omit lane_id and Attestify picks the best lane for the task
• Idempotency — pass idempotency_key to guarantee exactly-once execution
• Budget controls — pass constraints={"max_cost_usdc": 0.05} to cap spend per run
• Webhook support — pass options={"webhook_url": "..."} for async delivery
• Zero dependencies — stdlib only (urllib, json, uuid)

API Reference

create_client(api_key, base_url?, max_retries?, timeout_s?)

client.run_lane(input, ...)

Returns RunLaneResult(output, receipt, raw).

constraints fields:

constraints={
    "max_cost_usdc": 0.05,    # abort if estimated cost exceeds this
    "max_latency_ms": 10000,  # abort if estimated latency exceeds this
    "budget_id": "proj-abc",  # link to a named budget envelope
}

options fields:

options={
    "verify": True,                    # run output verification (default False)
    "write_memory": True,              # persist session memory (default False)
    "include_memory": True,            # inject prior memory into context
    "webhook_url": "https://...",      # async result delivery
}

client.authorize(...) — authority envelopes

Request a signed, server-enforced authority envelope before running one or more governed actions. run_lane(envelope=...) attaches it; /api/run verifies the signature, tenant/agent binding, expiry, live-policy epoch, and remaining action count / spend cap before every run, and atomically consumes one action.

envelope = client.authorize(
    agent_id="analyst-v1",
    intent="batch-report-generation",
    risk_class="low",
    max_spend_usd=0.50,
    action_count=20,     # this envelope authorises up to 20 actions
    expiry_window_s=300, # capped at 3600s server-side
)

result = client.run_lane(
    lane_id="analyst-v1",
    input="Summarise Q2 revenue trends.",
    envelope=envelope,
)

The client caches non-Enterprise envelopes locally and reuses them until the action count or expiry window is exhausted. If the agent's policy has per_action_mode enabled, the server forces action_count=1 and expiry_window_s=0 and the SDK never caches the result — every action gets a fresh envelope. A stale envelope (policy changed since issue) is rejected with HTTP 409; call authorize() again to obtain a fresh one.

client.get_receipt(loop_id)

Fetch a stored receipt by its loop_id.

receipt = client.get_receipt("loop_abc123")
print(receipt.verification)   # grade, score, output_hash
print(receipt.settlement)     # on-chain tx hash if x402 used

Returns RunReceipt.

RunReceipt fields

Available lanes

Omit lane_id entirely to let Attestify auto-route based on your input.

Environment variables

ATTESTIFY_API_KEY=atst_live_...

Async usage

The client is synchronous by design (zero dependencies). For async workflows:

import asyncio
from attestify import create_client

client = create_client(api_key=os.environ["ATTESTIFY_API_KEY"])

async def main():
    result = await asyncio.to_thread(
        client.run_lane,
        input="Analyse Q2 revenue trends",
        lane_id="analyst-v1",
    )
    print(result.output)

asyncio.run(main())