Zabta SDK

Python SDK for Zabta — govern, monitor, and manage your AI agents.

Installation

pip install zabta

This is a stable release — install the plain package name, no version pin needed. (The companion zabta-broker daemon is still in beta and must be pinned to an exact version; see "Credential Leasing" below.)

Quick Start

from zabta import ZabtaClient

client = ZabtaClient(
    api_key="zbt_your_key_here",
    agent_name="My Agent",
)
client.start()

# Check before acting
result = client.evaluate("send_email", "customer_data", {"has_pii": True})
if result.allowed:
    send_email(customer)
elif result.escalated:
    print(f"Needs approval: {result.reason}")
else:
    print(f"Denied: {result.reason}")

client.stop()

Policy Evaluation

# Full evaluation with details
result = client.evaluate("delete", "customer_record", {"is_irreversible": True})
print(result.decision)     # "allow", "deny", or "escalate"
print(result.policy)       # "Kill Switch"
print(result.citation)     # "OWASP ASI08, EU AI Act Art. 14"
print(result.layer)        # "universal"

# Quick boolean check
if client.is_allowed("read", "public_data"):
    read_data()

# Decorator — auto-checks before executing
@client.governed(action="process_refund", resource="payment")
def process_refund(order_id, amount):
    stripe.refunds.create(charge=order_id, amount=amount)

Monitoring Mode

Just log what your agent does (no enforcement):

client.start()
client.log_action("answered_ticket", input_summary="Resolved billing question")
client.log_action("sent_email", input_summary="Welcome email to new customer")
client.stop()

LLM Middleware

Auto-log every OpenAI or Anthropic call:

from zabta.middleware import wrap_openai
import openai

client = ZabtaClient(api_key="zbt_xxx")
oai = wrap_openai(openai.OpenAI(), client)

# This call is automatically logged with token usage, cost, and latency
response = oai.chat.completions.create(model="gpt-4o", messages=[...])

Credential Leasing (local Broker)

Instead of putting long-lived secrets in environment variables, lease a scoped credential from the local Zabta Broker for the duration of a block. The Broker holds the secret, checks policy, and hands it over (or refuses). It's a separate package — this is optional, not required for policy evaluation:

pip install zabta-broker==0.1.0b2

Pin the exact version — the Broker is pre-1.0 and --pre would opt your whole dependency tree into pre-releases. Full setup (vault, daemon, cloud registration) is in the Broker Quickstart.

Before — the key lives in the environment and is set once, globally:

import os, stripe
stripe.api_key = os.environ["STRIPE_API_KEY"]

def charge_customer(cust, amount):
    return stripe.Charge.create(customer=cust, amount=amount, currency="usd")

After — lease the key per operation, and pass it to a per-call client:

import stripe
import zabta

def charge_customer(cust, amount):
    with zabta.credential("stripe", scopes=["charges:create"]) as key:
        client = stripe.StripeClient(api_key=key)          # per-call client
        return client.charges.create(
            customer=cust, amount=amount, currency="usd",
        )

Use a per-call client, not stripe.api_key = key. Setting the module-level stripe.api_key inside the block mutates global state — under concurrency another task can read or clobber it at the wrong moment. Always bind the leased key to a local client instance, as above.

The block yields a usable secret or raises — never None, never "". A denial is always an exception: GovernanceError (policy denied), AgentNotRegisteredError (agent DID not registered with the Broker), EscalatedError (requires human approval), AuthenticationError (missing/stale Broker session token), ConnectionError (Broker unreachable).

Enforcement happens at checkout. Once the Broker hands over the credential, you hold the real secret. The lease TTL bounds the Broker's audit record, not the secret — Zabta does not revoke or time-limit the key mid-use. Scope the with block to the work that needs the credential, and rely on the credential's own lifecycle (rotation, provider-side expiry) for its validity.

Async agents use the same contract:

async with zabta.acredential("stripe", scopes=["charges:create"]) as key:
    client = stripe.StripeClient(api_key=key)
    await client.charges.create_async(...)

Identity: the Broker identifies agents by DID. Provide it via the ZABTA_AGENT_DID environment variable or agent_did="did:...". Discovery defaults to http://127.0.0.1:9477 (ZABTA_BROKER_URL to override); the per-session token is read from ~/.zabta-broker/session.token. require_approval raises EscalatedError immediately unless you pass approval_timeout=<seconds> to poll for a human decision.

API Reference

ZabtaClient

Parameter	Type	Default	Description
api_key	str	required	API key (zbt_ or aos_ prefix)
base_url	str	https://api.zabta.ai	API URL
agent_name	str	"Unnamed Agent"	Display name
agent_type	str	"custom"	Agent type

Methods:

• evaluate(action, resource, context) -> EvaluateResult — check policy
• check(action, resource, context) -> EvaluateResult — alias for evaluate
• is_allowed(action, resource) -> bool — quick boolean check
• governed(action, resource, context) — decorator for auto-checking
• start() / stop() — lifecycle management with heartbeats
• log_action(action_type, ...) — fire-and-forget logging
• request_action(action_type, ...) — request with approval flow

EvaluateResult

• .decision — "allow", "deny", or "escalate"
• .allowed / .denied / .escalated — boolean helpers
• .reason — human-readable explanation
• .policy — deciding policy name
• .citation — regulatory citation
• .layer — universal / jurisdiction / sectoral

Backward Compatibility

from agentos import AgentClient still works. aos_ API keys still work.

Changelog

0.4.0

• Credential leasing — zabta.credential() / zabta.acredential() context managers for leasing short-lived, policy-checked secrets from a local Zabta Broker instead of holding them in environment variables. Yields a real secret or raises; never a placeholder.
• Broker transport — the HTTP client the leasing calls use to reach the Broker daemon (zabta-broker, a separate package, published independently).
• Exception taxonomy for leasing — AgentNotRegisteredError, GovernanceError, EscalatedError, AuthenticationError, ConnectionError each mean a distinct failure (identity, policy, approval, session, connectivity) rather than one generic error.

Credential leasing requires a running zabta-broker daemon on the same machine — it does nothing without one. Everything else in this package (evaluate, check, is_allowed, governed, auto-instrumentation) talks only to the cloud API and needs no local daemon.

0.3.0 and earlier

Policy evaluation, auto-instrumentation for OpenAI/Anthropic/LangChain, and action logging. See PyPI release history for details.

License

MIT