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. Enforce the verdict
if result.verdict == "ALLOW":
    # Proceed with the action
    ...
elif result.verdict == "HOLD":
    # Held — operator must approve in the dashboard
    print("Held for review:", result.tool_call_id)
elif result.verdict == "BLOCK":
    # 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:

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

Integration patterns

Pre-execution gate

def safe_execute(client, action, context, execute):
    result = client.judge_action(action, context)
    if result.verdict == "BLOCK":
        raise RuntimeError(f"Blocked: {result.reason}")
    if result.verdict == "HOLD":
        raise RuntimeError(f"Held for review: {result.tool_call_id}")
    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

Proprietary — all rights reserved. See LICENSE.