annexops-sdk (Python)

The customer-side clients for AnnexOps — AILogger (EU AI Act Article 12 runtime log, including agent tool calls and approvals), 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 (npm), 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

Agent tool calls and approvals

If your AI system is an agent that calls tools, record each tool call and each human approval on the same logger. The tool arguments and result are hashed at source and discarded; only the tool name, the outcome, the approval reference and SHA-256 hashes are sent.

# a human approved the call (approval_ref is your own id: never a name or e-mail)
logger.log_approval(tool_name="send_email", approval_ref="appr-1042",
                    decision="approved", arguments=args_json)

# the call that relied on that approval
logger.log_tool_call(
    tool_name="send_email",
    arguments=args_json,   # hashed at source, never sent
    result=result_json,    # hashed at source, never sent
    outcome="succeeded",   # "succeeded" | "failed" | "blocked" (the call did not run)
    approval_ref="appr-1042",
)

tool_name is canonicalised before it is sent (trimmed, lower-case, - and spaces become _). tool_name and approval_ref are identifiers (letters, digits, _ . : / -, at most 128 characters), never free text. Pass arguments_hash / result_hash instead of the raw values if you hash yourself. Tool events go into the same system_key hash chain as your inference events; in the portal, the AI system page shows them in the Agent tool calls card.

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 © AnnexOps