apex-audit-sdk 0.1.0

Python capture SDK for the Apex Audit service: hash-at-source, tamper-evident audit trails for AI agent decisions.

apex-audit-sdk (Python)

Capture SDK for the Apex Audit service: tamper-evident, hash-at-source audit trails for AI agent decisions, sealed to a public ledger.

Three lines is all an agent loop needs:

txn = client.open_transaction(name="benefits eligibility review")
txn.log("reasoning", {"rationale": "Income below threshold; criterion B met."})
txn.log("output", {"recommendation": "approve", "confidence": 0.94})

Every entry is hashed on your side before it leaves the process (sha256 over the RFC 8785 canonical form of {transaction_id, seq, type, payload, prev_hash}), chained to the previous entry, and re-verified by the server on receipt. Nobody — including the service — can silently rewrite what your agent did.

Install

pip install apex-audit-sdk            # package name
python -c "import apex_audit"        # import name

From this repo (development):

cd packages/sdk-python
python3 -m venv .venv
.venv/bin/python -m pip install -e ".[dev]"
.venv/bin/python -m pytest

Requires Python >= 3.10. Runtime dependencies: httpx, rfc8785.

Quickstart

from apex_audit import AuditClient

client = AuditClient(
    api_url="http://localhost:4000",
    api_key="...",          # sent as Authorization: Bearer <key>
    timeout=10.0,
)

txn = client.open_transaction(name="benefits eligibility review")
txn.log("initiation", {"task": "benefits-eligibility-review", "case_ref": "EU-2026-00441"})
txn.log("action", {"tool": "case_management.lookup", "applicant_ref": "APP-99182"})
txn.log("reasoning", {"rationale": "Income below threshold; criterion B met."})
txn.log("output", {"recommendation": "approve", "confidence": 0.94})

txn.request_review()                                   # status -> in_review
txn.record_oversight(decision="approved", reviewer="j.okafor",
                     note="Consistent with policy 2026/14.")
txn.complete()                                         # status -> complete
sealed = txn.seal()                                    # Merkle root anchored on-ledger
print(sealed["merkle_root"], sealed["anchor_topic_id"])

client.close()   # or use:  with AuditClient(...) as client: ...

Context-manager style (opens the transaction on enter; nothing automatic on exit — review, completion and sealing stay explicit):

with client.transaction(name="loan pre-screen") as txn:
    txn.log("initiation", {...})
    ...

Entry types are restricted to initiation, action, reasoning, reference, output, oversight — anything else raises ValueError before any network call.

Error handling

The service reports errors as {"error": {"code", "message", "details"}}, mapped to exceptions:

from apex_audit import ApexAPIError, ChainConflictError, EntryHashRejectedError

try:
    txn.log("action", {"tool": "lookup"})
except ChainConflictError as e:       # 409 CHAIN_CONFLICT
    # another writer extended the chain, or a previous attempt landed.
    print(e.details["expected_seq"], e.details["expected_prev_hash"])
except EntryHashRejectedError as e:   # 422 ENTRY_HASH_INVALID
    # client/server canonicalisation disagreement — report it, don't retry.
    print(e.details["computed_by_server"], e.details["submitted"])
except ApexAPIError as e:             # everything else
    print(e.code, e.status, e.message, e.details)

The SDK never auto-retries POSTs. Appending an entry is not idempotent: if a request times out after the server stored it, a blind retry would be rejected as CHAIN_CONFLICT — and retry loops can mask genuine chain breaks. On any failure, local seq/prev_hash state is left unchanged so you can inspect, resynchronise, and retry deliberately.

Payload rules (the number caveat)

Payloads must be plain JSON: dict (string keys) / list / str / int / float / bool / None. NaN/Infinity, sets, datetimes, bytes, Decimals etc. raise ValueError with the offending path (e.g. payload.scores[2]).

Integers beyond 2^53 − 1 lose precision once they hit IEEE-754 doubles (RFC 8785) and Postgres jsonb. The SDK warns (PayloadPrecisionWarning) and recommends sending such values as strings:

txn.log("action", {"ledger_ref": str(9007199254740993)})   # exact
txn.log("action", {"ledger_ref": 9007199254740993})        # warns; value will round

Verifying later

data = client.get_transaction(txn_id)    # transaction + ordered entries
result = client.verify(txn_id)           # recompute + compare with ledger anchor
                                         # (404 until the service ships verify, M4)

Hash utilities are exported for independent verification:

from apex_audit import compute_entry_hash, compute_merkle_root, canonicalize, GENESIS_HASH

Full reference: docs/sdk/python.md.