atbash-sdk

Python SDK for Atbash — the safety layer that evaluates AI agent actions against operator-defined policies before execution.

Installation

pip install atbash-sdk

Requires Python 3.9+. Server-side only — private keys are used for local signing and must never be embedded in clients you don't control.

The PyPI distribution name is atbash-sdk; the import package is atbash:

from atbash import Atbash, generate_keypair

Quickstart

import os
from atbash import Atbash

# 1. Construct an Atbash client with the private key you saved during
#    agent creation. The constructor validates the key and derives the
#    matching public key for you.
client = Atbash(os.environ["ATBASH_AGENT_KEY"])

# 2. Submit an action for judgment, before executing it.
#    The SDK signs the transaction locally and sends it to the judge API.
#    Private key stays on your machine — never sent over HTTP.
result = client.judge_action(
    "Transfer $50,000 to external wallet 0xabc",
    "Outbound AML check — new recipient, over threshold",
)

# 3. Act on the decision. `allow` is the gate; `verdict` says why.
if result.allow:
    # Proceed with the action
    ...
elif result.verdict == "HOLD":
    # Held — operator must approve in the dashboard
    print("Held for review:", result.tool_call_id)
else:
    # Refused — agent is jailed in Enforcement tier
    raise RuntimeError(f"Blocked: {result.reason}")

Before this works, the agent must be onboarded at atbash.ai — assigned to an org, with a policy pack attached, and the org tier set to Audit+ or Enforcement.

How it works

client.judge_action() performs a two-step flow:

1. Sign locally — signs the transaction using the agent's private key. The key never leaves your machine.
2. Request verdict — sends the signed transaction, tool_call_id, and agent_pubkey to the Atbash judge API. The server broadcasts it to the Chromia blockchain and returns a verdict.

Don't have an agent yet?

There are two ways to create an agent:

1. Dashboard (recommended) — create an agent at atbash.ai/risk-engine/agents. The dashboard generates the keypair, assigns the agent to your org, and lets you attach a policy pack — all in one step.

2. Programmatic — generate a new keypair locally or bring an existing secp256k1 private key from another platform, then onboard it via the dashboard:

from atbash import generate_keypair

kp = generate_keypair()
print("Save this private key somewhere safe:", kp.priv_key)
# `kp` is a frozen typed record. `repr(kp)` redacts the private key so a
# stray print() or exception traceback can't leak it.

Note: After generating a key programmatically, you must still onboard the agent at atbash.ai/risk-engine/agents — assign it to an org and attach a policy pack before judge_action will work.

Secret storage

• Load the private key from an environment variable (ATBASH_AGENT_KEY) or a secret manager — never hardcode it.
• Never commit .env files containing the key.
• If a key leaks, stop using it and create a new agent in the dashboard.

Typed records (no surprise private keys in logs)

All multi-field results — KeyPair, AgentAuth, RedactResult, MemoryDiffResult, etc. — are typed records with attribute access (kp.priv_key, not kp["priv_key"]). Secret-bearing records redact themselves in repr() so a stray print() or exception traceback never leaks the private key:

>>> from atbash import generate_keypair
>>> kp = generate_keypair()
>>> kp
KeyPair(priv_key='<redacted>', pub_key='02b421a3a9…')
>>> kp.priv_key  # explicit access still works
'a9e794180b9472de…'

Verdicts

Every judge_action call returns one of three verdicts:

Gate on allow, not on verdict. allow is the canonical permission the SDK computes from every signal in the response; verdict is one of those signals, and reads as the reason. The two differ when an organisation does not enforce an agent's verdicts — Shadow Mode on an agent, or org-wide Monitor mode — where the action is judged and recorded on chain but deliberately not applied. Branching on verdict re-derives a denial the organisation chose not to make.

NB: If your org is on the Audit tier, the judge returns "No verdict" — actions are logged on-chain for the audit trail but not evaluated by an AI provider. Upgrade to Audit+ or Enforcement at atbash.ai/risk-engine/settings for active verdicts.

API

Judge

Atbash.judge_action(
    action: str,
    context: str = "",
    *,
    tool_name: str = "",
    tool_args_json: str = "",
) -> JudgeResult

Submit an action for judgment before execution. Signs the transaction locally and sends it to the judge API for a verdict.

class Atbash:
    def __init__(
        self,
        privkey: str,                       # 64-char hex secp256k1 private key (0x-prefix ok)
        *,
        endpoint: str = "https://atbash.ai",
        timeout: float = 120.0,
        chain: ChainConfig | None = None,          # Full chain override (BRID + nodeUrls)
        network: str | None = None,                # "public" | "private" preset selector
        node_urls: Sequence[str] | None = None,    # Paired with blockchain_rid
        blockchain_rid: str | None = None,         # Paired with node_urls
        org_name: str | None = None,               # Default org — routes calls to that org's chain
        org_encryption_pub_key: str | None = None, # Default org encryption key
        verify_pubkey: str | None = None,          # 66-hex judge response-sign pubkey
        authenticate: bool = False,                # Preflight the bearer against the judge
        fail_closed: bool = True,                  # audit_tool_call denies on error
        debug: bool = False,                       # Emit classifier probe logs
        logger: Any = None,                        # object with .info(...) / .warn(...)
    ) -> None: ...

@dataclass(frozen=True)
class JudgeResult:
    verdict: str         # "ALLOW", "HOLD", "BLOCK", or "No verdict"
    action_type: str     # "allow", "hold_for_user_confirm", or "block"
    reason: str          # Human-readable explanation
    confidence: float    # 0–1
    provider: str        # Which provider evaluated the action
    latency_ms: float    # Inference time
    tool_call_id: str    # Unique ID for this judgment
    on_chain: bool       # Whether the record was written on-chain
    status: str = ""     # "logged" on the AUDIT tier; empty otherwise

One-call guard: audit_tool_call

Atbash.audit_tool_call(input: ToolCallInput) -> Decision

Redact secrets → sign → judge → allow/deny Decision, failing closed by default (any error denies unless fail_closed=False on the client):

from atbash import Atbash, ToolCallInput

client = Atbash.from_config()
decision = client.audit_tool_call(ToolCallInput(
    tool_name="shell.exec",
    args={"cmd": "rm -rf /tmp/cache"},
    context="cleaning build cache",
))
if not decision.allow:
    raise RuntimeError(f"{decision.verdict}: {decision.reason}")

Secret-shaped values in args/context are redacted before signing, so they never reach the signed bytes, the request, the on-chain log, or the LLM prompt.

Log tool call (low-level)

Atbash.log_tool_call(
    action: str,
    context: str = "",
    *,
    tool_name: str = "",
    tool_args_json: str = "",
) -> LogToolCallResult

Sign the transaction locally. Returns LogToolCallResult(success, tool_call_id, signed_hex, error). Use this if you need to separate the signing step from the verdict request.

Poll judgment status

Atbash.get_judgment_status(
    judgment_id: str,
    agent_pubkey: str | None = None,
    chain: ChainConfig | None = None,   # Chain the judgment was signed against
) -> JudgmentStatus

Check whether a held action has been approved or rejected by an operator. Pass chain when polling a judgment signed on a chain other than the client's default — the server routes the query to that chain via the brid parameter.

@dataclass(frozen=True)
class JudgmentStatus:
    status: Literal["pending", "answered", "error"]
    verdict: str
    reason: str
    judgment_id: str
    on_chain: bool | None
    cached: bool | None
    response_time_ms: int | None

Agent identity

load_agent(privkey: str) -> AgentAuth
generate_keypair() -> KeyPair
derive_public_key(priv_key_hex: str) -> str
is_valid_private_key(hex: str) -> bool
load_agent_from_file(key_path: str | None = None) -> AgentAuth

load_agent(privkey) is the canonical loader — pass in the private key from the dashboard, get back AgentAuth(privkey, pubkey) ready for use. It accepts 0x-prefixed, padded, or mixed-case input and raises ValueError on malformed keys. Use generate_keypair() only for local development; for production, create agents in the dashboard so operators can attach policies.

load_agent_from_file(key_path) resolves the key from disk (defaults to ~/.config/atbash/guard-client-key), trimming whitespace and validating the format.

Operations

Methods that sign transactions and write to the Chromia blockchain.

Queries

Configuration

You can pass configuration inline to the Atbash(...) constructor or use the built-in config module that reads from ~/.config/atbash/config.json.

Inline

from atbash import Atbash

client = Atbash(
    privkey,
    endpoint="https://your-instance.example.com",
    timeout=30.0,
)
result = client.judge_action("Transfer $500", "finance")

Persistent config

Save configuration once — the SDK resolves values with priority: flag > env var > config file.

from atbash import Atbash, save_user_config

# Save once
save_user_config({
    "agentKey": "9cd07a...",
    "orgName": "my_org",
})

# Then construct from config anywhere
client = Atbash.from_config()
result = client.judge_action("Transfer $500", "finance")

Atbash.from_config() also validates the configured judge endpoint against the trusted allowlist (or your self-hosted policy) and wires the endpoint's response-signing pubkey as the default verify_pubkey.

The direct constructor applies the same policy. Atbash(privkey, endpoint=...) accepts only an https:// allowlisted judge host, or a self-hosted judge when verify_pubkey is also given (the 66-hex response-signing key, which then verifies every judge response). Plaintext http:// is accepted only for loopback (localhost, 127.0.0.1, ::1) local development. Anything else raises ValueError at construction, so a misconfigured or injected endpoint can no longer receive the agent's signed actions or return unsigned verdicts.

Config file location: ~/.config/atbash/config.json

Advanced: The SDK connects to the default Atbash Chromia chain. To use a different chain, pass node_urls= and blockchain_rid= to the Atbash(...) constructor.

Secret redaction

Before each audit_tool_call signs anything, the SDK scans args and context for secret-shaped values (API keys, tokens, JWTs, PEM blocks, etc.) and replaces matches with [REDACTED:<kind>]. Redaction happens before signing, so secrets never reach the signed bytes, the request body, the on-chain log, or the prompt sent to the AI provider.

When the redactor fires, you'll see a warning via the configured logger:

[atbash] redacted secrets before judge call { tool: "exec", count: 2, kinds: ["anthropic", "generic_token"] }

Common kinds:

• anthropic, openai, github, google, aws_access_key, stripe, slack, jwt, private_key_pem — high-confidence vendor patterns; if you see these, a real secret was almost certainly in your input
• context_secret — a value next to a label like api_key=, token:, password=, etc.
• generic_token — long random-looking strings (32+ alphanumeric chars). Catches unknown-vendor secrets, but can also match UUIDs, content hashes, and other opaque identifiers. The judge sees [REDACTED:generic_token] instead of the original; for verdict purposes the shape of the action matters more than the exact ID, so this is generally safe — but worth knowing if you see it unexpectedly.
• base64 — long base64-encoded values; can match legitimate image/file data

Redaction is silent at the consumer level — the SDK's caller still has the original arguments. Only what's sent to the judge (and persisted on chain via the verdict log) is scrubbed.

You can also call the redactor directly:

from atbash import redact_secrets, contains_secret

result = redact_secrets('curl -H "Authorization: Bearer sk-…" https://api.example.com')
result.redacted   # 'curl -H "Authorization: Bearer [REDACTED:openai]" https://api.example.com'
[(m.kind, m.length) for m in result.found]   # [('openai', 51)]

contains_secret("plain string")               # False

Telemetry

The SDK reports anonymous usage metrics — how often each SDK call is made, how long it takes, whether it succeeded, and a source label such as sdk, cli or plugin:langchain. They are sent over an authenticated connection to your Atbash endpoint and are used to find which integrations need attention.

Opt out:

• Environment (recommended for air-gapped deployments): ATBASH_TELEMETRY_DISABLED=1
• File: create ~/.config/atbash/telemetry.json with {"enabled": false}

When telemetry is on. Creating an Atbash(...) client enables it only for an endpoint that already receives the agent's credentials: Atbash's own service, or any endpoint you connect to with authenticate=True.

A self-hosted judge on the default authenticate=False sends no telemetry. That judge is trusted to return verdicts, and nothing more — telemetry would be the only thing handing it your agent's credentials.

Telemetry also needs an https:// endpoint, and is skipped for local addresses (localhost, 127.0.0.1, 0.0.0.0) so development servers never receive it.

Building an integration? If you call setup_telemetry(endpoint=..., get_auth_headers=...) directly — from a plugin or CLI that starts before an Atbash client exists — the endpoint rules above do not apply, because they live in the client. Pass get_auth_headers only for an endpoint you intend to authenticate to.

Integration patterns

Pre-execution gate

def safe_execute(client, action, context, execute):
    result = client.judge_action(action, context)
    if not result.allow:
        if result.verdict == "HOLD":
            raise RuntimeError(f"Held for review: {result.tool_call_id}")
        raise RuntimeError(f"Blocked: {result.reason}")
    return execute()

Polling a held action

import time

def wait_for_approval(client, tool_call_id, poll_interval=5.0):
    while True:
        status = client.get_judgment_status(tool_call_id)
        if status.status == "answered":
            return status.verdict
        if status.status == "error":
            raise RuntimeError(status.reason)
        time.sleep(poll_interval)

Checking agent jail status

policy = client.get_agent_policy(client.pubkey)
if policy.is_jailed:
    raise SystemExit("Agent is jailed — unjail via dashboard before retrying.")

Error handling

The SDK raises standard Exception subclasses. Known failure modes are enriched with a pointer to the relevant dashboard page, so the error message tells you where to fix the problem:

API error 404: {"error":"Agent not registered..."}
  → Onboard the agent at https://atbash.ai/risk-engine/agents

from atbash import AtbashAPIError

try:
    result = client.judge_action(action, context)
except AtbashAPIError as err:
    if "Agent not registered" in str(err):
        # Point the user at https://atbash.ai/risk-engine/agents to onboard.
        ...

Dashboard

Policy authoring, operator reviews, and agent management happen at atbash.ai. The SDK is the programmatic interface; the dashboard is the operator interface.

License

Licensed under the Apache License, Version 2.0. See NOTICE for attribution. Versions published before this change remain available under the license they shipped with.

Telemetry — what the SDK sends

The SDK reports how its own functions perform. It runs inside your agents and the data leaves your network, so the payload is metadata only and the list below is the whole of it.

Sent

Never sent. Action text, tool arguments, policy text, organisation names, memory file paths, memory contents, response bodies, and keys of any kind. error.path is the route alone — the query string is stripped where the error is built, because it carries org names, tool call ids and the agent pubkey. These are not fields that get emptied before sending — there is no field for them to go in.

agent.hash is de-identification, not anonymisation: agent public keys are readable from the chain, so anyone holding that list can match a hash back to an agent. Supply your own salted hasher to setup_telemetry(hash_agent_key=...) if you need it to be uncorrelatable.

Turning it off

export ATBASH_TELEMETRY_DISABLED=1

or create ~/.config/atbash/telemetry.json containing { "enabled": false }.

On a Private deployment, telemetry may already be off for everyone. The operator disables collection by leaving HONEYCOMB_API_KEY unset on their Atbash service; the telemetry route then accepts the request and discards the payload, so nothing is forwarded and agents never see an error or retry. That is an operator setting on the deployment, not one you control from the agent — the switches above are yours.

To keep the operational counters but drop the richer per-operation events:

export ATBASH_TELEMETRY_EVENTS_DISABLED=1

Sending it to your own collector

Point the SDK at any OTLP endpoint you control. Atbash is then bypassed entirely and no Atbash credential is created or transmitted — your headers are the only ones sent.

export ATBASH_TELEMETRY_COLLECTOR_URL=https://otel.yourcompany.example
export ATBASH_TELEMETRY_COLLECTOR_HEADERS='{"x-api-key":"your-key"}'

Metrics go to /v1/metrics and events to /v1/logs beneath that URL. https is required; a collector on localhost additionally needs ATBASH_TELEMETRY_ALLOW_LOOPBACK=1, so a mistyped URL cannot silently turn exporting off. A URL that is refused raises at setup rather than quietly falling back to Atbash.