Arkova Python SDK

Typed Python client for the Arkova Verification APIs.

Install

pip install 'arkova>=2.6.0'

Python 3.10 or newer is supported.

Supported methods

• anchor(data=None, *, fingerprint=None, file_mime=None)
• anchor_bulk(inputs, *, dry_run=None, duplicate_strategy=None, batch_id=None)
• fingerprint(data)
• search(q, type="all", cursor=None, limit=50)
• search_public(query, *, type=None, limit=None) for anonymous public search across organizations, members, secured records and public-record imports
• create_attestation(*, attestation_type, subject_identifier, attester_name, claims, ...) to create an attestation (POST /api/v1/attestations); recorded as an individual's own statement unless an organization administrator asks for an organizational attester_type
• verify(public_id)
• verify_fingerprint(fingerprint)
• list_anchors(...) for API-key-scoped private record listing with pagination and optional time or private-tag filters
• get_anchor(public_id)
• get_anchor_submission_status(public_id)
• anchor_import(rows, *, action, description=None, user_tags=(), organization_tags=()) for 1 to 100 already-fingerprinted spreadsheet rows (never retried automatically)
• get_merkle_proof(public_id), get_record(public_id), get_fingerprint(fingerprint), get_document(public_id), get_organization(public_id)
• list_folders(...), create_folder(...), update_folder(...), bind_folder_connector(...), delete_folder(...), move_records(...), move_records_by_public_id(...)
• import_source_preview(...) and import_source_confirm(...) for external record sources (need a short-lived user_access_token; never retried automatically)
• register_agent, list_agents, get_agent, update_agent, revoke_agent, create_agent_key, admit_computeid_agent (see "Agent lifecycle")
• list_orgs()

AnchorImportResultRow.recipient_status and the 500-character description check in anchor_import are included from 2.6.1 onward (see CHANGELOG.md). Check the installed version before using them.

Quick start

import os
from arkova import Arkova

with Arkova(api_key=os.environ["ARKOVA_API_KEY"]) as arkova:
    results = arkova.search("registered nurse", type="record", limit=5)
    for item in results.results:
        print(item.public_id, item.snippet)

Public search (no API key required)

search_public() calls the anonymous GET /api/v1/public/search route. It returns four groups (orgs, members, anchors, imports) that only ever carry public identifiers and URLs, never internal ids or emails. A configured API key is still sent and only raises your rate limit; a client constructed with api_key="" sends no Authorization header for this call.

from arkova import Arkova

with Arkova(api_key="") as arkova:
    page = arkova.search_public("acme university", type="orgs", limit=5)
    for org in page.results.orgs:
        print(org.display_name, org.verified, org.profile_url)
    if page.unavailable:
        print("partial results, unavailable groups:", page.unavailable)

query is trimmed and must be 2 to 200 characters, type is one of "all", "orgs", "members", "anchors", "imports", and limit is 1 to 50 per group. Invalid input raises ArkovaError(code="invalid_public_search_query") before any network call. A 503 with code="search_unavailable" means every requested group failed. AsyncArkova.search_public() has the same contract.

Anchor a document (HAKI-REQ-02)

anchor() fingerprints data client-side (SHA-256, in-process — the raw content is never sent, only the 64-char hex fingerprint) and submits it for network anchoring. Pass a pre-computed fingerprint instead if you already hashed the document elsewhere.

from arkova import Arkova

with Arkova(api_key="ak_live_...") as arkova:
    # Raw content — fingerprinted in-process before anything is sent.
    receipt = arkova.anchor("document content goes here")
    print(receipt.public_id, receipt.status)  # "PENDING" -> "SUBMITTED" -> "SECURED"
    print(receipt.record_uri)  # public verification page for this anchor

    # Or: you already have the fingerprint.
    receipt = arkova.anchor(fingerprint="a" * 64)

The same fingerprint always returns the same public_id — anchoring identical content twice is a no-op. file_mime, when supplied, is sent as source-document metadata; the document bytes remain local.

Use get_anchor_submission_status(public_id) for authenticated, caller-scoped queue or instant-processing state:

with Arkova(api_key="ak_live_...") as arkova:
    status = arkova.get_anchor_submission_status(receipt.public_id)
    print(status.action, status.anchor_status, status.instant_status, status.retryable)

AsyncArkova.get_anchor_submission_status(public_id) provides the same contract asynchronously. Current servers include action, credit_state, instant_status, and idempotent on new receipts; the receipt model keeps these fields optional so retained responses from older deployments remain parseable. The dedicated status model is the strict source for the current durable state.

Bulk-anchor documents (HAKI-REQ-02)

anchor_bulk() anchors up to BULK_ANCHOR_MAX_ROWS (1000) documents in one call. Each BulkAnchorInput row provides exactly one of fingerprint (a pre-computed 64-char hex SHA-256) or data (raw content, fingerprinted client-side the same way anchor() does it) — you can mix both forms across rows in one call.

from arkova import Arkova, BulkAnchorInput

with Arkova(api_key="ak_live_...") as arkova:
    with open("contract.pdf", "rb") as f:
        contract_bytes = f.read()

    result = arkova.anchor_bulk(
        [
            # Already hashed elsewhere — send the fingerprint directly.
            BulkAnchorInput(fingerprint="a" * 64, external_id="invoice-001"),
            # Raw content — the SDK hashes it for you before it's ever sent.
            BulkAnchorInput(
                data=contract_bytes,
                credential_type="CONTRACT_PRESIGNING",
                document_type="contract",
                matter_or_case_ref="CASE-42",
            ),
        ],
        duplicate_strategy="skip",
        batch_id="nightly-2026-07-28",
    )

    print(result.queued, result.duplicates, result.errors)
    for anchor in result.anchors or []:
        print(anchor.public_id, anchor.status)

Options: dry_run=True validates every row (including dedup checks) without queuing or deducting credits — result.anchors is None on a dry run. duplicate_strategy controls what happens when a fingerprint already exists in-batch or in your org; the server default is "fail" (raises ArkovaError(code="duplicate_fingerprints") on any duplicate — pass "skip", "supersede", or "link" to proceed instead). batch_id is your own correlation ID, echoed back and surfaced in audit events.

Limits: empty input returns a zero-row response immediately, no network call. More than 1000 rows raises ArkovaError(code="batch_too_large") — the SDK does not auto-chunk (splitting a logical batch across requests would let a duplicate fingerprint slip past the cheaper intra-batch check and would deduct credits per chunk instead of atomically for the whole batch; split manually and correlate with a shared batch_id if you need more than 1000 rows). A row with neither fingerprint nor data (or with both) raises ArkovaError(code="invalid_request"), checked before any network call.

Verify a fingerprint

from arkova import Arkova

fingerprint = "a" * 64

with Arkova(api_key="ak_live_...") as arkova:
    result = arkova.verify_fingerprint(fingerprint)
    print(result.verified, result.public_id)

Verify a public ID

from arkova import Arkova

with Arkova(api_key="ak_live_...") as arkova:
    result = arkova.verify("ARK-2026-ABC")
    print(result.verified, result.description, result.confidence_scores)

Upgrade from 2.2.0. In 2.2.0 verify() raised ArkovaError: Arkova API returned an unexpected response shape on any record carrying compliance controls — compliance_controls was typed as a mapping, but the API returns a list of control-ID strings. Fixed in 2.3.0 (2.2.1 was never published — see CHANGELOG.md); there is no workaround on 2.2.0 short of bypassing the model layer.

verify() returns the rich v1 verification shape, including API-RICH-01 fields such as compliance_controls (a list of control-ID strings, accompanied by compliance_controls_note whenever present), chain_confirmations, parent_public_id, version_number, file_mime, and file_size, plus API-RICH-02 fields confidence_scores and sub_type, and fingerprint_source / proof_availability / proof_availability_note, when the API response includes them. The same optional rich fields are typed on v2 verify_fingerprint() and get_anchor() responses, so newer API payloads are not silently hidden by the SDK model layer.

Async client

import asyncio
from arkova import AsyncArkova


async def main() -> None:
    async with AsyncArkova(api_key="ak_live_...") as arkova:
        orgs = await arkova.list_orgs()
        print([org.display_name for org in orgs.organizations])


asyncio.run(main())

Errors and retries

ArkovaError preserves the API v2 RFC 7807 problem document and the Retry-After header when present. code carries the machine-readable error code — from the plain-JSON error field on v1 write-path errors (anchor() / anchor_bulk() codes include "insufficient_credits", "duplicate_fingerprints", "batch_too_large", "invalid_request"), or the RFC 7807 type slug on v2 problem documents.

from arkova import Arkova, ArkovaError

try:
    with Arkova(api_key="ak_live_...") as arkova:
        arkova.get_anchor("ARK-DOC-MISSING")
except ArkovaError as exc:
    print(exc.status_code, exc.code, exc.problem.type if exc.problem else None)

The client retries 429 and 5xx responses by default and respects Retry-After within max_retry_delay (seconds, default 5). Longer server waits return the original error without an early retry. Pass retries=0 to disable retries.

Offline proof verification (no network, no API key)

verify_bundle verifies an exported Arkova proof package entirely offline — an independent re-derivation of the documented bundle format (Merkle recompute with the CVE-2012-2459 structural guard, fixed-offset on-chain payload decode, 80-byte header rules, timestamp honesty, Ed25519 signed bundles). It makes zero network calls and never contacts Arkova; on-chain confirmation runs only against canned or caller-supplied independent-node responses.

import json
from arkova import verify_bundle

packet = json.load(open("proof.json"))
outcome = verify_bundle(packet)          # recompute-only
print(outcome.verdict, outcome.reason_code)  # "VERIFIED" / None, or a frozen code

Every NOT-VERIFIED outcome carries one frozen machine reason code (arkova.REASON_CODES), kept in lockstep with the TypeScript reference verifier via a cross-runtime parity gate in the Arkova repo. A passing signature never substitutes for the cryptographic recompute; a failing explicitly-requested signature check fails the verdict closed.

Agent lifecycle

Arkova and AsyncArkova expose matching register_agent, list_agents, get_agent, update_agent, revoke_agent, create_agent_key, and admit_computeid_agent methods. Configure an organization API key with agents:manage; ComputeID admission accepts API-key authentication only and never widens scopes. Save keys returned by key creation and admission immediately; lifecycle writes are never automatically retried. The restricted key returned by admission is not automatically an agent-management key.

Webhook management (2.7.0 release candidate)

The methods in this section are prepared for arkova 2.7.0 and were not included in 2.6.2. Verify PyPI availability before installing; source metadata is not publication proof. Both Arkova and AsyncArkova provide the same 11 operations:

Use an organization API key with webhooks:manage. Every route, including reads, also requires the key creator's current owner or administrator membership in that exact organization. A key's scope alone does not grant that membership. These methods use the existing client base URL and API-key authentication for /api/v1/webhooks; they do not acquire credentials or change permissions.

For registration and updates, scope="self" targets the key's organization; scope="self_and_descendants" explicitly requests its descendant organizations. When you send a scope, the SDK requires the successful response to acknowledge that exact value. Omit scope to preserve the server's registration default or leave an existing endpoint's scope unchanged. An omitted description is also left unchanged on updates; description=None explicitly clears it. Older read responses may omit scope, in which case endpoint.scope is None.

Returned models expose only documented fields. Endpoint reads never expose a secret; save the WebhookEndpointWithSecret.secret from creation securely when it is returned. Delivery history excludes raw payloads, receiver bodies, and error text. Dead letters expose endpoint_display_url, with URL credentials, query parameters, and fragments removed. History and dead-letter limits are 1–100 (default 50); their total is the number of rows returned, not an exhaustive count or a cursor. Attempt number zero is valid.

A replay receipt with replayed=True confirms the replay was attempted, not that the receiver accepted it. Only ok=True with a 2xx status_code reports delivery success. Likewise, resolving a dead letter or receiving a materialization recovery receipt does not prove delivery. Recovery's request_id is an explicit idempotency identifier; retain it when checking the result of an uncertain request.

Webhook operations make one request and are never automatically retried, including when the client has retries configured. Error codes retain known server reasons such as ssrf_blocked, payload_refused, and authorization_unavailable; raw server and transport details are omitted. A timeout, transport failure, or invalid successful mutation response warns that the operation may have succeeded. Inspect its receipt before deciding to retry. These client methods do not change delivery schedules or guarantee delivery.

Retry waits are bounded: max_retry_delay (default 5 seconds, available on sync and async clients). If a valid server Retry-After requests a longer wait, the client surfaces the original error and retry metadata without sleeping or retrying early. Short valid waits retain the server minimum.