zanii-blackbox

The Python SDK for Zanii Blackbox, the flight recorder for AI agents.

Blackbox records every model call and tool call your agent makes, for any industry and any organisation, from a gateway that runs outside the agent's process. The records are hash-chained, so an edit or a deletion shows, and anchored as receipts that anyone can verify offline. This SDK is optional. It adds what only the agent knows (steps, its own tools, its checks, the outcome) as a second, independent record, and it reads and verifies what the gateway recorded.

pip install zanii-blackbox
# or
uv add zanii-blackbox

Python 3.11 or later. Fully typed.

Before you start

You need a running Blackbox gateway, self-hosted with Docker, Kubernetes or from source. A hosted UAE region is planned; write to info@zanii.agency for early access. The gateway holds your provider keys. Your agent gets a session token instead, and its model calls go through the gateway:

export BLACKBOX_URL=https://blackbox.example.com
export BLACKBOX_ADMIN_KEY=...     # or a tenant key (bbxt_...)

Record a run

import os

import anthropic
from zanii_blackbox import session

url = os.environ["BLACKBOX_URL"]

with session(
    url=url,
    admin_key=os.environ["BLACKBOX_ADMIN_KEY"],
    label="refund-agent",
    budget_usd="2.00",
    flight_plan={
        "objective": "Process today's refund queue",
        "expected_tools": ["lookup_order", "issue_refund"],
    },
) as s:
    # Model calls go through the gateway with the session's token, never your provider key.
    client = anthropic.Anthropic(base_url=f"{url}/anthropic", api_key=s.token)

    s.step("load queue")
    s.tool_call("lookup_order", {"order": 8812})
    s.tool_result("lookup_order", {"ok": True})
    s.verify("amount-matches-invoice", True)  # your own check, recorded as evidence

    # A risky tool waits for a person, who sees the arguments in the cockpit.
    answer = s.request_approval(
        "issue_refund",
        reason="Damaged on delivery",
        args={"order": 8812, "amount_aed": "1250.00"},
    )
    if answer == "approved":
        ...  # issue the refund

    s.close("success")

answer is only "approved" when a person approved it. "rejected", "timeout" and "error" all mean don't run the tool.

With asyncio, use s = await asession(...) and async with. The calls that wait on the gateway have async versions: aflush, astate, arequest_approval and aclose.

What the SDK guarantees

• It never raises into your agent. If the gateway is down, the session keeps recording to disk and reports the problem in s.stats.last_error.
• It loses nothing it accepted. Every event is written to a local spool before it is sent, numbered without gaps, and resent until the gateway confirms it. After a crash, blackbox drain sends what's left.
• It stays out of the way. Recording is non-blocking with bounded memory. A recorded call costs about 0.4 ms (measured over 20,000 calls on a laptop); shipping happens in the background.
• It has one runtime dependency: the Zanii core library.

Hallucination controls

The gateway checks every record, with no model needed: an invented tool or ID, arguments a tool's schema refuses, "done" right after an error, and amounts, doses, dates and IDs in answers that nothing the agent was given holds. With your organisation's own reference packs (prices, fees, rules, limits) it also catches answers that contradict them, and your own grounding model and judge can plug in. A risky answer can be held for a person before your customer sees it.

From the SDK, pin what the agent read, then check an answer's signed credential anywhere:

from zanii_blackbox import verify_answer_credential, verify_certificate

s.retrieved([{"id": "returns-policy", "version": "2026-10", "content": policy_text}])  # hashed here, never sent

# later, from GET /v1/sessions/:id/answers/:seq/credential and the session's record:
verify_certificate(credential)["ok"]                      # signed by the gateway
verify_answer_credential(credential, lines, bodies, packs)  # {"ok", "problems"}: the record backs every claim

The same checks run offline as pure functions: tool_hallucinations, grounding_findings, facts_in, load_reference_packs, answer_risk, detector_accuracy.

New in 0.13.0

• from_otlp(body): OpenTelemetry spans (GenAI semconv or OpenInference) as session events, the mapping behind gateway v0.15's POST /v1/traces. Point any framework's OTLP exporter at the gateway with the session's token, and its spans join the record.

New in 0.12.0

• Tax credit notes: tax_credit_note / render_tax_credit_note, the document and printable copy behind gateway v0.14's POST /v1/invoices/:number/credit-notes.

New in 0.11.0

• Facts against the record compare currencies and units: AED 500 in a tool result no longer grounds USD 500 in the answer, nor 500 g grounds 500 mg (currency_codes, unit_of). Dates written with an English or Arabic month (3 March 2026, 3 مارس 2026) are facts too.
• blackbox policy-hook: a Claude Code / Codex PreToolUse hook that asks the gateway's policy before a local tool runs, and fails closed.
• Works with gateway v0.13 (POST /v1/sessions/:id/policy-check).

New in 0.10.0

• set_tenant_policy / tenant_policy / remove_tenant_policy: a tenant's own rules, hot on the gateway. set_provider_key / provider_keys / remove_provider_key: a tenant's own model keys, sealed. cancel_run(session_id, run_id): stop one run (calls carrying x-blackbox-run).
• load_policy refuses an args_match that JavaScript and Python would read differently (portable_regex_error).
• blackbox sessions --outcome success|failure|abandoned.
• Works with gateway v0.12: alert routing to PagerDuty, Teams, Alertmanager and Slack; the key passphrase from Vault or AWS KMS.

New in 0.9.0

• Client.create_key(tenant_id, scope="approve", expires_in_days=...): a key that only decides approvals (and labels findings), and keys that expire.
• Works with gateway v0.11: Idempotency-Key kept in the database (shared by every instance), the admin audit log anchored on the ledger and naming the signed-in person, and BLACKBOX_APPROVALS_NEED_PERSON.

New in 0.8.0

• blackbox drain --close (and drainSpools({close: true}) / drain_spools(close=True)): after shipping what a stopped agent left behind, each session is closed as abandoned.
• Works with gateway v0.10: alerts signed as Standard Webhooks, tenant keys that expire, more identity-provider algorithms (PS256, ES384, EdDSA).

New in 0.7.0

• Any number of serverless functions or workers can write one session at the same moment: a joined session sends its events unnumbered, each with an id, and a v0.9+ gateway numbers them in arrival order (a resend adds nothing).
• The repeat-call detector counts within one turn: a chat assistant or support bot asked for the same lookup in separate messages isn't stopped as a loop. A real loop inside one task still is.

New in 0.6.0

• Serverless: a session joined with its id and token in a fresh function (Vercel, Lambda, Cloud Run) numbers its events from the server's sdk_next, so none are lost. Join in each invocation and flush before returning.
• Fewer false alarms: an ISO date-time (2026-10-03T09:30:00+04:00) in a tool's arguments is a date, not an invented number, and an MCP call carrying out the model's own tool call isn't flagged a second time for the same value.

New in 0.5.0

• Held streams: open a session with hold_streams=True (or set BLACKBOX_HOLD_STREAMS=all on the gateway) and a risky streamed answer waits for a person, as a non-streamed one does. The client gets keep-alives, then the whole stream, or one error in the provider's own format.
• session.check_answer(seq=None): what in an answer would hold it ({seq, risks, hold}), for an app that streams by itself.

New in 0.4.0

• Hallucination controls: the checks above, reference packs, answer credentials (answer_claims, answer_credential, verify_answer_credential), session.retrieved(), accuracy from people's labels (detector_accuracy, Wilson bounds), the grounding and judge contracts, answer_risk for held answers.
• Keyed redaction tokens (token_for, resolve_tokens).
• Evidence for payments, health and public services (payment, health_access, decision and their verifiers); provable agent memory (memory_chain); record badges, agent CVs, SLAs, ledger co-signing and the post-quantum binding.
• Compliance reports in Arabic, and the hallucination-controls report (hallucination_facts).

Frameworks

The framework integrations are imported lazily, so installing this package doesn't pull any of them in.

Command line

The package installs a blackbox command:

blackbox sessions                     # the newest sessions, and whether each record verifies
blackbox tail ses_...                 # follow a session live
blackbox export ses_... --out run.json
blackbox verify run.json              # verify an audit bundle offline, anywhere
blackbox reconcile ...                # check an agent's own logs against the gateway's record
blackbox mcp-wrap --server files -- python server.py   # record a local stdio MCP server
blackbox drain                        # send spooled events left by a stopped agent

Verify a record offline

from zanii_blackbox import verify_bundle

An audit bundle carries the events, the chain and the anchors. Checking it needs no account and no network, so an auditor or a regulator can check it independently.

Also in TypeScript

@zanii/blackbox has the same API in camelCase. Both SDKs must produce the same output on the same shared test vectors.

Support

• Website: zanii.agency
• Email: info@zanii.agency
• Security reports: info@zanii.agency, privately, not in a public issue

Licensed under Apache-2.0.