Skip to Content
Sarek AI AccountabilitySDK Reference: Python

sarek

Python SDK for accountable AI. Wraps your AI client, logs every call as a signed entry, and anchors a proof per decision. Python 3.11 or later. Depends on cryptography, httpx and sarek-core.

pip install sarek

The public entry point is AiAccountability:

from sarek import AiAccountability

AiAccountability.init()

Creates a logger instance. Loads or creates the agent identity (a P-256 key pair on disk) and ensures the local directory structure exists.

Signature

@classmethod async def init( cls, name: str = 'default', role: str = 'unspecified', ledger_url: str = 'https://api.ai-ledger.com', on_log: Callable[[LogEntry, AiLoggerProof], None] | None = None, on_stamped: Callable[[AiLoggerProof], None] | None = None, ) -> AiAccountability

Parameters

NameTypeDefaultDescription
namestr'default'Agent name. One identity and log file per name.
rolestr'unspecified'The agent’s role, recorded in every entry.
ledger_urlstrCore Service endpoint.
on_logCallableNoneCalled immediately after signing.
on_stampedCallableNoneCalled when the proof is anchored (~10 s later).

Returns

AiAccountability (awaitable factory).

Example

from sarek import AiAccountability sarek = await AiAccountability.init( name='claims-triage', role='insurance-claims', on_log=lambda entry, proof: audit_db.insert(entry), on_stamped=lambda proof: audit_db.attach_proof(proof.log_id, proof.tx_hash), )

monitor()

Wraps an AI client and returns the wrapper. The original client is never mutated; wrapper classes delegate everything else via __getattr__. Both sync and async underlying clients are handled. Calls are made with keyword arguments, exactly like the unwrapped client.

Signature

def monitor(self, client, options: dict | None = None) -> object

Parameters

NameTypeDescription
clientobjectThe AI client to wrap.
options['provider']strRequired only for clients that cannot be distinguished automatically.

Returns

A wrapped client with the same API as the original. Each response carries the proof at response.ai_logger. Bedrock responses are plain dicts, so the proof is attached as response['ai_logger'].

Supported providers

ProviderDetection
OpenAI, Anthropic, Gemini, Azure OpenAI, Ollama (native), AWS Bedrock (Converse), Mistral, Cohere (V2)Automatic
Groq, Together AI, OpenRouter, Perplexity, FireworksAutomatic with the official SDK; hint when using an OpenAI client against their endpoint
xAI (Grok)Hint: options={'provider': 'xai'} (wire-identical to OpenAI)

Example

import openai import anthropic client = sarek.monitor(openai.AsyncOpenAI()) claude = sarek.monitor(anthropic.AsyncAnthropic())

Errors

Raises ValueError if the client cannot be recognized and no provider hint is given.

close()

Waits for all in-flight anchoring tasks. Call it before process exit so no proofs are lost. Safe to call more than once.

Signature

async def close(self) -> None

Example

await sarek.close()

sarek.core: stamp and verify directly

For data that is not an AI call, the package exposes the Core stamping primitives directly. No logger, no identity, no files.

Signature

def stamp(data: str, *, url: str | None = None, metadata: dict | None = None) -> StampResult def verify(proof: str, *, url: str | None = None) -> VerifyResult

Both are intentionally synchronous and blocking: stamp() waits for anchoring (~10 s), verify() is one network round-trip. Run them in a worker thread on hot paths.

Returns

@dataclass class StampResult: id: str hash: str ldgp: str tx_hash: str block_number: int timestamp: str metadata: dict | None = None @dataclass class VerifyResult: valid: bool tree_integrity: bool on_chain: bool

Example

import sarek result = sarek.core.stamp('model weights v3.2, sha256 9f86d081...') check = sarek.core.verify(result.ldgp)

The log flow

  1. Input and output are hashed separately with SHA3-256. Objects are serialized canonically (compact JSON, sorted keys), so the same call produces the same hashes in Python and TypeScript.
  2. The entry is signed with the agent’s P-256 key.
  3. on_log(entry, proof) fires immediately. The AI response is already on its way back to your code.
  4. The combined hash is anchored asynchronously in an executor (~10 s). Each anchoring uses its own client instance, so parallel calls never interfere. Failures are logged and never raised into your application.
  5. The proof file is written, the log line is updated with tx_hash and block_number, and on_stamped(proof) fires.

LogEntry

One entry per AI call, appended to ai-accountability/logs/<agent>.jsonl.

@dataclass class LogEntry: log_id: str # UUID, unique per call timestamp: str # ISO 8601 provider: str # one of the 14 supported providers model: str input_hash: str # SHA3-256, never the raw prompt output_hash: str # SHA3-256, never the raw completion prediction_id: str # provider response id, or a generated UUID input_tokens: int output_tokens: int total_tokens: int latency_ms: int finish_reason: str | None status_code: int # always 200; failed calls are not logged system_fingerprint: str | None environment: str | None # always None in Python sdk_version: str agent_id: str role: str signature: str # ECDSA P-256 over input_hash + output_hash tx_hash: str # filled in after anchoring block_number: int # filled in after anchoring citations: list[str] | None = None # Perplexity only

AiLoggerProof

Attached to every response as response.ai_logger and passed to both callbacks.

@dataclass class AiLoggerProof: log_id: str agent_id: str role: str input_hash: str output_hash: str ldgp: str # base64-encoded proof file tx_hash: str block_number: int proof_path: str # path to the .ldgp file on disk

Local files

ai-accountability/ .gitignore created automatically, protects private keys agents/ <name>/ identity.json private.key P-256 private key (chmod 600) logs/ <name>.jsonl one line per AI call proofs/ <name>/ log-<log_id>.ldgp one proof per call
Last updated on