annexops_sdk (Python)

The customer-side clients for AnnexOps — AILogger (EU AI Act Article 12 runtime log), ConsentLogger (GDPR Article 7 consent receipts), and SchemaScanner (private-database schema discovery for data mapping). The stdlib-only Python mirror of the TypeScript @annexops/sdk, with the same hashes-only / schema-only wire contracts.

Engineering documentation, not legal advice. AnnexOps does not certify conformity. Keeping these records supports (but does not by itself discharge) your EU AI Act Article 12 record-keeping and GDPR Article 7 consent-recording obligations — whether they satisfy Article 12/7 for your system is a determination for you and your counsel.

Zero third-party dependencies: hashlib, uuid, urllib, json, threading and friends only.

Install

pip install annexops-sdk

Requires Python ≥ 3.9. Zero third-party runtime dependencies.

Quickstart — AILogger (runtime inference log, Article 12)

import os
from annexops_sdk import AILogger

logger = AILogger(api_key=os.environ["ANNEXOPS_API_KEY"], system_key="my-model-v1")
logger.log_inference(input=prompt, output=response)
logger.close()  # on shutdown — flushes the remainder

Quickstart — ConsentLogger (GDPR consent receipts, Article 7)

import os
from annexops_sdk import ConsentLogger

consent_logger = ConsentLogger(
    api_key=os.environ["ANNEXOPS_API_KEY"],  # the same key as AILogger
    purpose_key="marketing-emails",
)
consent_logger.log_consent(
    subject_id=user_email,
    action="grant",
    consent_text=cookie_banner_notice_text,
    consent_version="v3",
)
consent_logger.log_consent(subject_id=user_email, action="withdraw")  # no consent_text needed
consent_logger.close()  # on shutdown — flushes the remainder

subject_id is hashed at source with SHA-256, then run through a second server-side HMAC step with a secret pepper AnnexOps holds — a low-entropy identifier (an email, a user id) can't be reversed from the stored value the way a bare hash could. consent_text (high-entropy notice copy) is a bare SHA-256, the same treatment AILogger gives your prompts/responses; your code never needs to know the difference. Pass subject_hash/ consent_text_hash instead of the raw values if you'd rather hash yourself (same mutual-exclusivity rule as AILogger's input/input_hash).

Quickstart — SchemaScanner (data-map schema push)

SchemaScanner reads the column metadata of one of your databases — schema, table, and column names plus their declared types — and pushes it to AnnexOps, where it is classified into your data map. Only names and types are read; no row is ever queried, and no data value ever leaves your process. The query is a fixed information_schema read the SDK owns — you never pass SQL.

You bring your own driver (psycopg or pymysql); the SDK bundles none. Give it a runner that executes exactly the SQL it hands you and returns the rows (dicts keyed table_schema/table_name/column_name/data_type):

import os
import psycopg
from psycopg.rows import dict_row
from annexops_sdk import SchemaScanner

conn = psycopg.connect(os.environ["DATABASE_URL"])

def runner(sql):
    with conn.cursor(row_factory=dict_row) as cur:
        cur.execute(sql)  # runs exactly the SQL passed — the SDK never interpolates
        return cur.fetchall()

scanner = SchemaScanner(
    api_key=os.environ["ANNEXOPS_API_KEY"],  # the same key as AILogger
    store_key="prod-postgres",               # a stable name you choose per database
    source="postgres",                       # "postgres" | "mysql"
    store_name="Production Postgres",         # optional label
)

result = scanner.scan_and_push(runner)
conn.close()
print(f"Pushed {result['elements_ingested']} columns; {result['classifications']} classified.")

For MySQL, pass source="mysql" and a pymysql runner (a DictCursor). scan(runner) and push() are also available separately. A very large schema is truncated at 50 000 columns and reported as completeness: "partial".

This is a single-shot push — no batching, no interval timer, no close(); just the shared retry loop (network / 5xx / 429 retried with full-jitter backoff, 401/400 never retried). Grant the scanning credential a read-only role that can read information_schema and nothing more.

What arrives in AnnexOps

AILogger POSTs to POST /v1/runtime-logs; ConsentLogger POSTs to POST /v1/consent-receipts; SchemaScanner POSTs to POST /v1/data-stores/schema-push. Each is the only endpoint its client calls. Runtime-log events land under Runtime Logs, appended to the per-(org, system_key) hash chain; consent receipts land under Consent, appended to the per-(org, purpose_key) hash chain; pushed schema lands under Data Stores, classified into your data map (feeding RoPA and DSAR). Viewing events/receipts/stores, verifying chain integrity, and exporting records all happen in the AnnexOps portal UI, signed in — they are portal features, not SDK or API-key endpoints.

Key handling

The full ak_live_<prefix8>_<secret32> key is shown exactly once at mint (Settings → API keys); both loggers require the full 49-char key — the same key works for both, nothing extra to mint. NEVER commit the key to source — it matches common secret-scanner patterns. Rotate by minting a new key and revoking the old. Neither logger ever logs the key or places it in an exception message.

Hashes only, never content

log_inference(input=..., output=...) computes input_hash/output_hash at source via sha256_hex(...) (exported, for callers who want to hash themselves — "you hash what you pass": strings are UTF-8-encoded, bytes pass through). log_consent(subject_id=..., consent_text=...) computes subject_hash/consent_text_hash the same way. Raw content never leaves your process — the wire types have no field that could carry it, and the server-side schema is strict.

Retries + idempotency

Same batch, same event_ids on every retry: event_id is generated once at log time and the retry loop re-POSTs the byte-identical body. Full jitter, 3 retries, 30 s cap, 10 s request timeout; retries network errors / 5xx / 429, never other 4xx. Configure via max_retries, backoff_base_ms, backoff_cap_ms, request_timeout_ms (identical constructor options on both loggers).

Flush + close

Each logger buffers up to flush_at: 100 events (hard-clamped ≤ 500, the wire batch max) and flushes on interval (flush_interval_ms: 5000, a daemon-thread timer that never keeps the interpreter alive) or on close(). Bodies are split to ≤ 500 events and ≤ 1 MiB each. Always call .close() on shutdown to avoid losing events; per-event server-side rejections surface in the returned {"accepted": ..., "rejected": [...]} and via the optional on_rejected callback. close() is bounded: its TOTAL wait is capped by close_timeout_s (default: the retry policy's worst case plus slack — 135 s with default options). Within that one budget it first joins any in-flight flushes, then runs its own final flush on a bounded path (a daemon worker joined with the remaining deadline) — so a wedged network stack can never hang your process shutdown, even when the wedge happens inside the final flush itself; a timeout surfaces via on_error with a static message. close() does not lock the instance (mirror of the TS SDK): log_inference() /log_consent() after close() still buffers and still auto-flushes at flush_at; a subsequent explicit flush() sends the rest.

License

MIT.