ZABTA DOCS

SDK Reference

Complete API reference for the Zabta Python SDK v0.3.0.

Installation

bash
pip install zabta

Current version: 0.3.0. Requires Python 3.8+.

Authentication

python
from zabta import ZabtaClient

client = ZabtaClient(
    api_key="zbt_your_api_key_here",
    agent_id="your_agent_id"
)
  • API keys use the zbt_ prefix — see the API Keys guide for details
  • Get your API key from the Zabta dashboard after registering an agent
  • The agent_id ties evaluations to a specific registered agent

Core methods

client.evaluate()

The primary method. Sends an agent action to the policy engine for evaluation.

python
result = client.evaluate(
    action="send_email",
    context={
        "recipient": "user@example.com",
        "content": "Follow-up on your inquiry",
        "jurisdiction": "eu"
    }
)
ParameterTypeRequiredDescription
actionstrYesThe action the agent wants to perform
contextdictNoAdditional context for policy evaluation (jurisdiction, data types, user info)

Returns an EvaluateResult object (see Response model below).

client.check() and client.is_allowed()

check() is an alias for evaluate() — same signature, same return type, reads better at some call sites. is_allowed() is a boolean convenience: it calls evaluate() internally and catches any exception, including a connection failure, returning False rather than letting it propagate. This is the SDK's concrete errs-toward-denial behavior — an unreachable API can never silently read as “allowed.”

python
# check() — same as evaluate(), reads better at a call site
result = client.check("send_email", "customer_data", {"has_pii": True})

# is_allowed() — boolean convenience. Catches ANY exception (including
# ConnectionError) and returns False. It errs toward denial: a network
# blip or an unreachable API never accidentally reads as "allowed."
if client.is_allowed("send_email", "customer_data"):
    send_email(customer)

client.governed()

An instance-method decorator that wraps a function with the same evaluate/check/is_allowed decision model:

python
# client.governed() — the policy-evaluation decorator (same decision
# model as evaluate()/check()/is_allowed()).
@client.governed(action="send_email", resource="customer_data")
def send_client_email(to: str, subject: str, body: str):
    # Your agent logic here
    ...

If the policy engine returns deny, the wrapped function raises GovernanceError instead of executing. If escalate, it raises EscalatedError.

Two decorators share the name “governed.” A second, standalone governed() is also exported from zabta directly (from zabta import governed). It is not the same function: it wraps client.request_action() — the action-logging / human-approval workflow — takes action_type (not action) and a requires access-level string, has no context parameter, and only ever raises GovernanceError (denial or pending-approval both use it; there is no escalation-specific exception on this path). Prefer client.governed() above unless you specifically need the request/approval logging flow.
python
# A second, different decorator also ships at the top level: governed()
# from zabta.decorators. It wraps client.request_action() — the
# action-logging / human-approval workflow, not policy evaluation — and
# takes different arguments (action_type, not action; no context).
from zabta import governed

@governed(client, action_type="send_email", requires="execute")
def send_client_email(to: str, subject: str, body: str):
    ...

Auto-instrumentation

Automatically patches supported LLM provider libraries when you call zabta.init() — there is no per-provider client method (no client.instrument_openai()). After instrumentation, every API call to the provider is evaluated by Zabta's policy engine before execution. No code changes needed in your existing agent logic.

python
import zabta

# auto_patch=True (the default) instruments whichever of OpenAI,
# Anthropic, and LangChain are installed. There is no client-side
# instrument_openai()-style method — this is the only entry point.
zabta.init("zbt_your_api_key", auto_patch=True)

Supported providers

Each is detected and patched only if the corresponding package is installed — pip install zabta[openai], zabta[anthropic], zabta[langchain], or zabta[all] for every provider.

  • OpenAI (chat completions, completions)
  • Anthropic (messages)
  • LangChain (chains, agents)

Response model

Every evaluate() / check() call returns an EvaluateResult with these fields:

FieldTypeDescription
result.decisionstr"allow", "deny", or "escalate"
result.allowedboolTrue if decision is allow
result.deniedboolTrue if decision is deny
result.escalatedboolTrue if decision is escalate
result.reasonstrHuman-readable explanation of the decision
result.policystrName of the policy that triggered the decision (includes custom policies)
result.citationstr | NoneRegulatory citation (e.g., "GDPR Art. 5"), when applicable
result.layerstr | None"universal", "jurisdiction", "sectoral", or "custom"
result.risk_tierstr | None"minimal", "limited", "high", or "unacceptable"
result.evaluation_time_msintTime taken for evaluation in milliseconds
result.explanationstr | NoneAI-generated plain-language explanation — only populated when you pass explain=True to evaluate(); requires Starter plan or above
result.fix_suggestionstr | NoneAI-generated suggestion for resolving a deny/escalate — same explain=True gating

The result.policy and result.layer fields reflect custom policies you create in the dashboard, in addition to built-in jurisdiction and sectoral policies.

Example usage

python
result = client.evaluate(action="classify_applicant", context={"jurisdiction": "eu"})

if result.denied:
    print(f"Blocked: {result.reason}")
    print(f"Policy: {result.policy}")
    print(f"Citation: {result.citation}")
elif result.escalated:
    print(f"Needs review: {result.reason}")
else:
    # Proceed with agent action
    proceed()

Error handling

python
from zabta.exceptions import (
    ZabtaError,               # Base class for every Zabta SDK exception
    AuthenticationError,      # Invalid or missing API key (401/403)
    ConnectionError,          # Cannot reach api.zabta.ai (network failure)
    APIError,                 # Non-2xx response after retries (status_code, detail)
    GovernanceError,          # Action denied by policy (from evaluate()/check()/governed())
    EscalatedError,           # Action requires human review (from governed())
    AgentNotRegisteredError,  # Agent not registered — call client.start() first
    RetryExhaustedError,      # All retry attempts exhausted (attempts, last_error)
)

ConnectionError and RetryExhaustedError are what you see when Zabta is unreachable — the SDK does not silently choose allow or deny on your behalf. evaluate() and check() let the exception propagate; catch it and decide within your own risk tolerance. is_allowed() makes that choice for you by catching every exception and returning False — see Core methods above.

Configuration options

python
client = ZabtaClient(
    api_key="zbt_...",
    agent_id="...",
    base_url="https://api.zabta.ai",  # Default, override for self-hosted
    timeout=30,                         # Request timeout in seconds
)

Related

SDK QuickstartPolicy EngineCompliance ScoringBroker Quickstart