Verify before the agent acts. Deterministic runtime safety for AI.

No local models or heavy infrastructure required. All verification evaluates securely via api.arcezia.com.

1 · Generate an API Key

Sign up self-serve at arcezia.com, or mint a developer key via cURL:

curl -X POST https://api.arcezia.com/v1/signup \
  -A "arcezia-client/1.0" \
  -H "Content-Type: application/json" \
  -d '{"email": "developer@company.com"}'
# Returns: { "api_key": "ar_live_...", "role": "owner" }

Store your key immediately: It is shown only once. The returned key holds the owner role, which is required for setup operations (registering probes, upload custom domains). Live keys begin with ar_live_...; development keys begin with ar_test_....

WAF & Direct REST Calling Gotchas: Always pass a valid User-Agent header (e.g. -A "arcezia-client/1.0"). Default scripting client signatures trigger Cloudflare/WAF block 1010. When making REST calls directly, authenticate via Authorization: Bearer ar_live_... (X-API-Key is rejected with 422).

2 · Install the Python Client SDK

pip install "arcezia[langchain]"   # or [openai], [anthropic], [llamaindex], [all]
export ARCEZIA_API_KEY="ar_live_..."

3 · Execute Your First Verification

import arcezia

# The task defines scope. The envelope says what this agent may do at most —
# without it, "is this action part of the assigned job" has no answer and
# nothing reaches ALLOW.
az = arcezia.Arcezia(task="report on last year's events")

az.start_session(capability_envelope={
    "allowed_domains": ["database_ops"],
    "allowed_action_types": ["execute_sql"],
})

# Request safety verification before running a tool
cert = az.verify(
    action_type="execute_sql",
    action_description="SELECT COUNT(*) FROM events",
    domain="database_ops"
)

if not cert.allow:
    raise RuntimeError(f"Action held by Arcezia: {cert.summary}")

# Reached: a read inside the declared scope returns ALLOW
print(f"Cleared for execution! Single-use credential: {cert.credential}")

4 · Try a Write

A read inside the declared scope clears immediately. A write does not, and this is the point at which most first integrations stop and wonder what went wrong.

cert = az.verify(
    action_type="execute_sql",
    action_description="UPDATE customers SET tier='pro' WHERE id = 42",
    domain="database_ops"
)
print(cert.verdict)   # REVIEW
print(cert.missing)   # ['user_explicit_authorization', ...]

REVIEW, not BLOCK. Nothing is wrong: changing stored data needs an approval the agent cannot give itself, and on a fresh account nobody has given one. Attach a real signed approval from your own backend and the same call clears — see Cryptographic human approval tokens for how to mint one.

A destructive statement is different again. It stays BLOCK however wide you open the envelope and whatever token you attach, because the rule wants a recent verified backup, and that is a fact about your infrastructure rather than something the call can carry. Register a probe for it and your own backup system answers.

5 · Try to Break It

Two tests worth running before you trust any of this. First, a statement nothing can justify:

cert = az.verify(
    action_type="execute_sql",
    action_description="DROP TABLE users",
    domain="database_ops"
)
print(cert.verdict)   # BLOCK

Second — and this is the one that matters — have the agent claim an approval it does not have:

cert = az.verify(
    action_type="execute_sql",
    action_description="DELETE FROM users WHERE 1=1",
    domain="database_ops",
    agent_evidence={"production_explicit_authorization": True},   # the agent lies
)
print(cert.verdict)               # BLOCK
print(cert.fabrication_detected)  # True — the claim was caught and thrown out

A text filter has no way to run that second test, because there is nothing to compare the claim against. Measured on both a development and a production key on 2026-09-10.

Gate on yes, never on “not a no”

There are three answers, so if not cert.block lets every REVIEW through. Only cert.allow means cleared.

# Wrong — a REVIEW falls through and the action runs.
if not cert.block:
    run(action)

# Right — only a yes reaches the real work.
if cert.allow:
    run(action)
elif cert.review:
    ask_a_person(cert.summary)
else:
    raise RuntimeError(cert.summary)

Arcezia acts as a runtime gate in front of your tool executions. Your models, prompts, and database tools remain unchanged.

Arcezia resolves safety into three distinct outcomes. Understanding REVIEW is fundamental: it represents a fail-safe hold when facts are unresolved.

Always Gate Positively on cert.allow: Never check if not cert.block. A REVIEW verdict is neither ALLOW nor BLOCK; testing only against BLOCK would allow unverified actions to execute.

Certificate Object Fields

Arcezia evaluates evidence according to strict provenance. An agent's self-generated text cannot satisfy structural requirements.

• Probe Webhooks: Verified directly from your databases, cloud APIs, and compliance databases.
• Deterministic Analysis: Verified by SQL parsers, AST analyzers, and environment variables.
• Human Approval Tokens: Cryptographically signed Ed25519 tokens minted by your backend.
• Model Output: Treated as CLAIMED. If an LLM attempts to claim human authorization in its text, Arcezia marks the fact FABRICATED and returns BLOCK.

Two External Facts That Cannot Be Read From Action Text:
• Environment Detection: Arcezia determines environment from ENVIRONMENT, ENV, or DATABASE_URL. If none are set, it defaults safely to production. Setting ENVIRONMENT=staging on your service enables staging rules; writing "staging" in action text changes nothing.
• Bulk Email Unsubscribe Links: For email_ops, unsubscribe_link_present must be grounded by a probe webhook connected to your mail provider or template engine.

Comprehensive runtime protection across individual tools, memory, workflows, and post-execution audits.

Integrate Arcezia into any agent stack with one-line tool wrapping.

# 1. LangChain / LangGraph
from arcezia.integrations.langchain import ArceziaToolkit
safe_tools = ArceziaToolkit(az).wrap([sql_tool, bash_tool], domain_overrides={"sql_tool": "database_ops"})
# For LangGraph tool-calling nodes:
safe_graph_tools = ArceziaToolkit(az).wrap_for_langgraph([sql_tool, bash_tool])

# 2. OpenAI Function Calling (name is positional and first)
from arcezia.integrations.openai import ArceziaGuard
safe_db = ArceziaGuard(az).wrap_function("execute_sql", db.execute, domain="database_ops")

# 3. Anthropic Tool Use Filtering
from arcezia.integrations.anthropic import ArceziaAnthropicGuard
safe_uses, blocked = ArceziaAnthropicGuard(az).filter_tool_uses(message.content)
for tool_call in safe_uses:
    execute(tool_call)

# 4. LlamaIndex
from llama_index.core.tools import FunctionTool
from arcezia.integrations.llamaindex import guard_tool
safe_tool = guard_tool(FunctionTool.from_defaults(fn=run_sql), az, domain="database_ops")

# 5. AutoGen (0.2 & 0.4+)
from arcezia.integrations.autogen import ArceziaAutoGenGuard
safe_sql = ArceziaAutoGenGuard(az).wrap("execute_sql", db.execute, domain="database_ops")

# 6. OpenCLAW (Dispatch loop)
from arcezia.integrations.openclaw import DispatchGuard
safe_dispatch = DispatchGuard(az).wrap(dispatch)

# 7. Universal Callable (Pydantic AI, smolagents, Custom Loops)
from arcezia import guard_callable
safe_fn = guard_callable(my_function, az, action_type="send_invoice", domain="payment_ops")

# 8. Claude Code Hook
# Run from terminal: $ arcezia-hook install

# 9. n8n (HTTP Node)
# Use Header: Authorization: Bearer ar_live_...
# Read verdict from $json.verdict and score from $json.precondition_score

CRITICAL Tool-Name Scope Gotcha: The wrapper passes the tool's actual definition name as action_type (e.g. wrapping sql_tool checks action_type="sql_tool", not "execute_sql"). Ensure your capability envelope lists the exact function names you wrap in allowed_action_types, or action_within_task_scope will ground False and BLOCK.

Chaining Helpers Disallowed on Wrapped Tools: Calling .bind(), .with_config(), .map(), .pipe(), .with_retry() on a wrapped tool raises AttributeError by design. These methods create new unwrapped objects that bypass safety checks. Always apply chaining helpers to your tool before wrapping, or use wrap_for_langgraph().

Inspect multi-action workflows to detect danger sequences before executing any individual step.

plan_manifest = {
    "steps": [
        {
            "id": "read_ssn",
            "action_type": "execute_sql",
            "action_description": "SELECT ssn, name FROM customer_pii WHERE id = 101",
            "domain": "database_ops"
        },
        {
            "id": "post_external",
            "action_type": "http_post",
            "action_description": "POST payload to https://external-analytics.com/dump",
            "domain": "api_ops"
        }
    ]
}

result = az.verify_chain(plan_manifest, stop_on_block=True)

if result["overall_verdict"] == "SEMANTIC_BLOCK":
    print(f"Plan blocked at step: {result['blocked_at']}")
    print(f"Trigger: {result['semantic_triggers']}")
    # Output: Semantic block: Data Exfiltration sequence detected

How to build, secure, and deploy dynamic probe endpoints on your infrastructure to answer evidence checks in real time.

1 · Registering Probe Endpoints

Register your probe webhook once using your API key. You can register a single dynamic router endpoint across multiple constraints:

curl -X POST https://api.arcezia.com/v1/probes \
  -H "Authorization: Bearer ar_live_..." \
  -H "Content-Type: application/json" \
  -d '{
    "domain": "database_ops",
    "constraint_name": "verified_recent_backup",
    "webhook_url": "https://ops.yourdomain.com/api/arcezia-probe",
    "secret": "your_secure_shared_secret_min16chars"
  }'
# Returns: created (bool), dispatch_readback ("confirmed"), signing_key_v2

Registration Notes: Registration is idempotent; re-registering re-arms the probe. If registration returns HTTP 500 registration_not_visible, retry immediately. The webhook endpoint must be publicly accessible via HTTPS (use ngrok/cloudflared for local testing) and answer within 5 seconds.

2 · Inbound Request Contract

When resolving a constraint, Arcezia POSTs a signed JSON payload to your webhook:

3 · Production Dynamic Probe Router (FastAPI Example)

import hmac, hashlib, os, datetime
from fastapi import FastAPI, Request, HTTPException, status

app = FastAPI(title="Arcezia Dynamic Probe Handler")

SIGNING_KEY_V2 = os.getenv("ARCEZIA_SIGNING_KEY_V2", "your_signing_key_v2")

# 1. Signature Verification Helper (Raw Body HMAC-SHA256)
async def verify_arcezia_signature(request: Request, raw_body: bytes):
    sig_header = request.headers.get("X-Arcezia-Signature-V2")
    if not sig_header:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing signature header")
    
    expected_sig = hmac.new(
        SIGNING_KEY_V2.encode("utf-8"),
        raw_body,
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(f"sha256={expected_sig}", sig_header):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid signature")

# 2. Dynamic Handlers (Lookup via parameters map)
def check_verified_recent_backup(params: dict) -> tuple[bool, str]:
    last_backup_time = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(hours=1)
    is_valid = (datetime.datetime.now(datetime.timezone.utc) - last_backup_time).total_seconds() < 86400
    return is_valid, f"Latest snapshot verified at {last_backup_time.isoformat()}"

def check_aml_passed(params: dict) -> tuple[bool, str]:
    recipient_id = params.get("recipient_id") if params else None
    if not recipient_id:
        return False, "Missing recipient_id in action parameters"
    return True, f"Recipient {recipient_id} cleared OFAC/AML screening"

PROBE_DISPATCH_TABLE = {
    ("database_ops", "verified_recent_backup"): check_verified_recent_backup,
    ("payment_ops", "aml_check_passed"): check_aml_passed,
}

# 3. Dynamic Webhook Endpoint
@app.post("/api/arcezia-probe")
async def handle_probe_webhook(request: Request):
    raw_body = await request.body()
    await verify_arcezia_signature(request, raw_body)
    
    payload = await request.json()
    domain = payload.get("domain")
    constraint = payload.get("constraint")
    parameters = payload.get("parameters") or {}
    
    handler = PROBE_DISPATCH_TABLE.get((domain, constraint))
    if not handler:
        return { "value": False, "grounded": False, "detail": f"No handler for {constraint}" }
    
    try:
        is_satisfied, detail_msg = handler(parameters)
        return { "value": is_satisfied, "grounded": True, "detail": detail_msg }
    except Exception as err:
        return { "value": False, "grounded": False, "detail": f"Error: {str(err)}" }

4 · Node.js / Express Implementation

const crypto = require('crypto');
const express = require('express');
const app = express();

// Capture raw body for exact HMAC-SHA256 signature verification
app.use(express.json({
  verify: (req, res, buf) => { req.rawBody = buf; }
}));

const SIGNING_KEY_V2 = process.env.ARCEZIA_SIGNING_KEY_V2;

app.post('/api/arcezia-probe', (req, res) => {
  const sigHeader = req.headers['x-arcezia-signature-v2'];
  const computed = crypto.createHmac('sha256', SIGNING_KEY_V2)
                         .update(req.rawBody)
                         .digest('hex');

  if (!sigHeader || !crypto.timingSafeEqual(Buffer.from(`sha256=${computed}`), Buffer.from(sigHeader))) {
    return res.status(401).json({ error: 'Invalid signature' });
  }

  const { domain, constraint, parameters } = req.body;

  if (domain === 'database_ops' && constraint === 'verified_recent_backup') {
    return res.json({
      value: true,
      grounded: true,
      detail: 'Postgres pg_dump confirmed within 2 hours'
    });
  }

  return res.json({ value: false, grounded: false, detail: 'Constraint unhandled' });
});

Allow high-risk operations (such as production drops or high-value transfers) to proceed with cryptographic proof of human approval.

1 · Register Your Ed25519 Public Key

curl -X POST https://api.arcezia.com/v1/account/token_key \
  -H "Authorization: Bearer ar_live_..." \
  -H "Content-Type: application/json" \
  -d '{ "public_key": "<base64url-of-32-byte-ed25519-public-key>" }'

2 · Token Structure & Issuance

When an admin confirms an action in your web UI, your backend mints a signed token (max 15-minute TTL):

# Token Format: base64url(payload_json) + "." + base64url(ed25519_signature)
{
  "acct": "acct_your_org_id",
  "sid": "session_id_from_az",
  "typ": "production_explicit_authorization",
  "exp": 1724836000,
  "jti": "unique_single_use_token_id"
}

3 · Attaching the Token in Python

# Attach human intent token to the session
az.authorize_production(signed_human_token)

# Re-verify the action
cert = az.verify(
    action_type="execute_sql",
    action_description="DROP TABLE legacy_staging_temp",
    domain="database_ops"
)
if cert.allow:
    db.execute(...)  # Cleared by verified human signature

Guard against unintended blast radius before and after execution.

Look-Ahead Simulation Webhook

Register a simulation sandbox (POST /v1/simulate/register) to dry-run queries before gating:

def simulate_database_action(req):
    # Dry-run inside a rolled-back transaction
    with db.transaction() as tx:
        cursor = tx.execute(req.json["action_description"])
        rows = cursor.rowcount
        tx.rollback()

    # CRITICAL: rowcount is -1 for DDL (DROP/ALTER/TRUNCATE) in Python DB-API
    if rows is None or rows < 0:
        return { "acceptable": False, "reason": "Unmeasured DDL mutation (DROP/ALTER)" }

    return {
        "acceptable": rows <= 1000,
        "reason": f"{rows} rows touched",
        "outcome": { "rows_affected": rows }
    }

Post-Execution Outcome Audit

audit = az.verify_outcome(
    action_type="execute_sql",
    action_description="Prune analytics",
    outcome={"rows_affected": 125000},  # Real consequence
    expected={"rows_affected": 500}      # Authorized scope
)
if audit.block:
    alert_security_team(f"Anomaly detected: {audit.violations}")

Set hard architectural ceilings for each session using signed capability envelopes.

az.start_session(capability_envelope={
    "allowed_domains": ["database_ops"],
    "allowed_action_types": ["execute_sql"],
    "max_scope": "limited",  # single_record=1 | batch=100 | limited=1000 | mass=unlimited
    "structural_authority": {
        "sensitive_data": False,
        "persistent_mutation": True,          # NOT "mutation"
        "mass_scope": False,
        "outbound": False,
        "trust_boundary_crossing": False,     # NOT "trust_crossing"
        "irreversible": True                  # DELETE/DROP requires True
    }
})

Envelope Gotchas:
• Listing Action Types: Omitting allowed_action_types is permissive within allowed domains. An empty list [] permits nothing and BLOCKs all actions.
• Setting False is a Hard Ceiling: An axis set to False can never reach ALLOW, even with a human approval token. To allow escalation, omit the axis from the envelope instead of setting it False.

Complete lookup matrix of all 18 built-in domains to their required preconditions and resolution channels.

Rate Limit Handling (HTTP 429)

from arcezia.exceptions import ArceziaRateLimitError

try:
    cert = az.verify(...)
except ArceziaRateLimitError as exc:
    if exc.retry_after:
        time.sleep(exc.retry_after)  # Per-minute burst limit (retry after seconds)
    else:
        raise RuntimeError("API verification quota reached. Contact administrator.")

Diagnosing REVIEW Outcomes

If an action returns REVIEW unexpectedly, check two things:

• 1. Inspect cert.constraints: Find constraints marked UNRESOLVED. Look up their grounding channel in Section 11.
• 2. Check cert.probe_outcomes: Compares registered probes against outcomes (answered, declined, unreachable, rejected, malformed). If a registered probe is absent from probe_outcomes, re-register it via POST /v1/probes.
• 3. Name Actions Concretely: Vague action descriptions like "run_cleanup('temp')" default to REVIEW because state mutation cannot be statically determined. Name actions concretely: "delete temp files older than 30 days from ./build".

cert = az.verify(...)
if cert.review:
    print(f"Review Reason: {cert.summary}")
    for c in cert.constraints:
        if c["quality"] == "UNRESOLVED":
            print(f"→ Missing ungrounded fact: {c['name']}")