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 sarekThe public entry point is AiAccountability:
from sarek import AiAccountabilityAiAccountability.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,
) -> AiAccountabilityParameters
| Name | Type | Default | Description |
|---|---|---|---|
name | str | 'default' | Agent name. One identity and log file per name. |
role | str | 'unspecified' | The agent’s role, recorded in every entry. |
ledger_url | str | Core Service endpoint. | |
on_log | Callable | None | Called immediately after signing. |
on_stamped | Callable | None | Called 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) -> objectParameters
| Name | Type | Description |
|---|---|---|
client | object | The AI client to wrap. |
options['provider'] | str | Required 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
| Provider | Detection |
|---|---|
| OpenAI, Anthropic, Gemini, Azure OpenAI, Ollama (native), AWS Bedrock (Converse), Mistral, Cohere (V2) | Automatic |
| Groq, Together AI, OpenRouter, Perplexity, Fireworks | Automatic with the official SDK; hint when using an OpenAI client against their endpoint |
| xAI (Grok) | Hint: options={'provider': 'xai'} (wire-identical to OpenAI) |
Example
Auto-detected
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) -> NoneExample
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) -> VerifyResultBoth 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: boolExample
import sarek
result = sarek.core.stamp('model weights v3.2, sha256 9f86d081...')
check = sarek.core.verify(result.ldgp)The log flow
- 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.
- The entry is signed with the agent’s P-256 key.
on_log(entry, proof)fires immediately. The AI response is already on its way back to your code.- 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.
- The proof file is written, the log line is updated with
tx_hashandblock_number, andon_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 onlyAiLoggerProof
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 diskLocal 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