controlzero

AI agent governance for Python. Policies, audit, and observability for tool calls. Works locally with no signup.

v1.0.0 is a complete rewrite. If you depend on control-zero<1.0.0 (the hosted-mode SDK), pin your requirement: control-zero<1.0.0 to stay on the legacy v0.3.x. The new v1.0.0+ is a local-first SDK with a different API surface; see the Python SDK reference for details.

Hello World

from controlzero import Client

cz = Client(policy={
    "rules": [
        {"deny":  "delete_*", "reason": "Hello World: deletes are blocked"},
        {"allow": "*",        "reason": "Hello World: everything else is fine"},
    ]
})

print(cz.guard("delete_file", {"path": "/tmp/foo"}).decision)  # "deny"
print(cz.guard("read_file",   {"path": "/tmp/foo"}).decision)  # "allow"

11 lines. No API key. No signup. Run it.

Install

pip install controlzero

Why

Your AI agents call tools. Some of those tools should never be called by an agent without a human in the loop. controlzero is the policy layer between the model's output and the tool execution. Decisions are fail-closed by default.

You can use it offline with a local YAML or JSON file or Python dict. When you want to share policies across a team or get a hosted audit dashboard, sign up at controlzero.ai and set CONTROLZERO_API_KEY.

Quickstart with the CLI

# 1. Generate a starter policy file with examples and comments
controlzero init

# 2. Edit controlzero.yaml in your editor

# 3. Validate it
controlzero validate

# 4. Test a tool call against the policy
controlzero test delete_file

The generated controlzero.yaml is the tutorial. It ships with annotated rules covering the common patterns: allow lists, deny lists, wildcards, and the catch-all.

Templates available (controlzero init -t <name>):

• generic — Hello World template (default)
• rag — RAG agent template (block exfiltration)
• mcp — MCP server template
• cost-cap — model allow-listing and cost guards
• claude-code — Claude Code hook starter
• langchain — LangChain tool guardrails
• crewai — CrewAI starter policy
• cursor — Cursor / editor hook starter
• autogen — AutoGen starter policy
• codex-cli — Codex CLI hook starter
• gemini-cli — Gemini CLI hook starter
• kiro — Kiro (AWS) hook starter (CLI: GA; IDE: PRIVATE PREVIEW)
• antigravity — Google Antigravity (IDE + agy CLI) hook starter (BETA)

Loading a policy

Three ways:

from controlzero import Client

# From a Python dict
cz = Client(policy={
    "rules": [
        {"deny": "delete_*"},
        {"allow": "read_*"},
    ]
})

# From a YAML file
cz = Client(policy_file="./controlzero.yaml")

# From an environment variable
# (set CONTROLZERO_POLICY_FILE=./controlzero.yaml)
cz = Client()

If a policy file exists in the current directory it is picked up automatically -- controlzero.yaml, controlzero.yml, or controlzero.json are auto-detected in that order (first existing wins). No environment variable needed. The file may be YAML or JSON; both use the identical schema.

Policy schema

version: '1'
rules:
  # Block any tool whose name starts with "delete_"
  - deny: 'delete_*'
    reason: 'Deletes need human approval'

  # Allow specific known-good tools
  - allow: 'search'
  - allow: 'read_*'

  # tool:method syntax
  - allow: 'github:list_*'
  - deny: 'github:delete_repo'

  # Catch-all
  - deny: '*'
    reason: 'Default deny'

Rules are evaluated top to bottom. The first match wins. If no rule matches, the call is denied (fail-closed).

Localized block messages (reason_localized)

The reason shown when a rule fires is plain UTF-8 text, so you can write it in any language. To serve more than one language from the same policy, add a reason_localized map and select a locale with the CONTROLZERO_LOCALE environment variable (e.g. ko, ko-KR):

version: '1'
rules:
  - deny: 'delete_*'
    reason: 'Deletes need human approval' # English default
    reason_localized:
      ko: '삭제는 사람의 승인이 필요합니다'
  - allow: '*'

• When CONTROLZERO_LOCALE is unset (or set to a locale the rule has no entry for), the plain reason is used -- existing behavior is unchanged.
• The SDK's own built-in messages (no-rule-match, DLP block, empty/observe bundle) ship an English default plus a Korean translation, selected the same way. The packs are pure in-binary maps (no runtime i18n dependency), so this is air-gap safe.
• Localization is display-only: it never changes which rule matches or the decided effect.

Tamper detection and quarantine

The policy YAML supports a settings: section that controls how the SDK responds when it detects that the local policy file has been modified outside of normal channels (manual edits, unexpected hash changes, etc.):

version: '1'
settings:
  tamper_behavior: warn # Options: warn | deny | deny-all | quarantine
rules:
  - deny: 'delete_*'
  - allow: '*'

Quarantine recovery. When a machine enters quarantine (deny-all or quarantine), every tool call is denied until you re-establish trust with one of these commands:

controlzero enroll
controlzero policy-pull
controlzero sign-policy

Org-level policy signing. When a machine is enrolled via controlzero enroll, it receives the organization's signing public key. Policy bundles pulled from the backend are cryptographically signed and verified by the SDK automatically. No extra configuration is required.

Tamper alert reporting. In quarantine mode, the SDK reports a tamper alert to the Control Zero backend so your team can see it on the dashboard.

Local audit log

Every decision (allow and deny) is written to a local audit log in every mode — local, hybrid, and hosted. The local log is never skipped, so controlzero tail, cz debug-bundle, and the tamper hash-chain always have a record to read.

controlzero tail

Default paths:

• Local / unenrolled mode (no API key): ./controlzero.log, with daily rotation and 30-day retention.
• Hosted mode (CONTROLZERO_API_KEY set): ~/.controlzero/audit.log when you do not pass an explicit log_path. Local audit is written in addition to the remote dashboard sink, not instead of it — the remote sink is layered on top. In hosted mode, PII and financial DLP matched_text is redacted from the local plaintext row (the secret category is already hashed); the remote sink keeps full fidelity.

Configure rotation via the client (honoured in any mode):

cz = Client(
    policy_file="./controlzero.yaml",
    log_path="./logs/controlzero.log",
    log_rotation="10 MB",        # rotate at 10 MB, or "daily", or "1 hour"
    log_retention="30 days",
    log_compression="gz",        # gzip rotated files
    log_format="json",           # or "pretty"
)

Hybrid mode

Default (T103, 2026-05-12): when CONTROLZERO_API_KEY is set, the hosted (dashboard) policy wins. Pass CONTROLZERO_LOCAL_OVERRIDE=1 to force the local file as a debug fallback.

If you BOTH set an API key AND pass a policy= / policy_file= arg to Client(...), the explicit local arg wins (caller is intentional) and you get a loud WARN log on init:

WARNING: controlzero: explicit local policy overrides the hosted bundle. ...

This makes accidental prod bypass impossible to miss. For prod environments, opt into strict mode to raise instead:

cz = Client(api_key="cz_live_...", policy=local_policy, strict_hosted=True)
# HybridModeError: explicit local policy overrides the hosted bundle ...

Coding agent hooks

controlzero hook-check runs inside Claude Code, Gemini CLI, and Codex CLI on every tool use and evaluates the call against your policy before it fires. It extracts a canonical tool:method from the tool arguments so rules can target database:SELECT vs database:DROP, or allow Bash:git while denying Bash:rm. Multi-statement SQL and compound shell commands are resolved to the most dangerous token, so a SELECT ... ; DROP TABLE users; payload matches database:DROP, not database:SELECT. See Hook action extraction for the full extraction rules, security model, and per-tool examples.

Framework examples

Full integration guides at docs.controlzero.ai/docs/integrations:

• LangChain
• LangGraph
• CrewAI
• OpenAI Agents SDK
• Anthropic tool use
• Pydantic AI
• AutoGen
• MCP servers
• Raw HTTP / no framework

Hosted mode

When you want a dashboard, audit search, team policies, and approval workflows, sign up at controlzero.ai and set the API key:

import os
os.environ["CONTROLZERO_API_KEY"] = "cz_live_..."

from controlzero import Client
cz = Client()  # picks up the API key from env, audit ships remote

Human-in-the-Loop approvals

Approvals let a policy block a tool call until a human approver decides allow or deny.

The short path: escalate_on_deny + guard_with_approval

Tag the rule, and await guard_with_approval instead of guard. The SDK raises the approval request, polls it to a terminal state, and returns an allow only if an approver approved:

rules:
  - deny: 'database:delete:*'
    escalate_on_deny: true
    reason: 'destructive database delete needs a human'

async def delete_users_table(cz):
    decision = await cz.guard_with_approval("database:delete", {"table": "users"})
    if decision.decision == "allow":
        ...  # an approver said yes
    else:
        ...  # denied, expired, revoked, timed out, or the backend was unreachable

Scope, because it matters: only this method escalates. A plain guard call on the same rule returns the deny, and the CLI hooks (controlzero hook-check) keep hard-blocking it -- a hook cannot await a remote approver, and prompting the local agent user would ask the restricted party to approve their own denial. Every non-approval outcome is a deny; an escalation never fails open. The audit row records which happened: escalation_status is requested when a human was asked and not_requested when the deny was enforced without asking one.

The long path: request_approval + wait()

For full control over the wait, client.request_approval(decision, ...) returns a PendingApproval you block on yourself.

Basic flow:

from controlzero import Client, PolicyDeniedError

cz = Client(api_key="cz_live_...")  # approvals run on any Postgres-backed deployment: hosted (SaaS), self-managed, or air-gapped

decision = cz.guard("delete_file", {"path": "/etc/passwd"})
if decision.denied and decision.requires_approval:
    pending = cz.request_approval(
        decision,
        message="agent wants to delete /etc/passwd; please confirm",
        timeout_s=300,
    )

    # Blocks until the human approves, denies, or the SLA expires.
    # No argument: request_approval() wired the backend poller
    # (GET /v1/sdk/approval-requests/{id}) from this client's API key
    # and API URL.
    try:
        pending.wait()
    except PolicyDeniedError:
        # denied, or the deadline passed (HITLTimeoutError subclasses
        # PolicyDeniedError). Abort the tool call.
        raise
    # approved -> proceed with the gated action; pending.status == "approved"

wait() blocks the calling thread. For async code, use wait_async(), same contract, event-loop friendly:

await pending.wait_async()

To drive polling yourself -- a different HTTP stack, or an in-process test backend -- pass a poll_fn: a callable taking a request_id and returning the backend's latest snapshot as a dict with a state key.

def poll_fn(request_id: str) -> dict:
    ...  # return {"state": "pending" | "approved" | "denied" | "expired"}

pending.wait(poll_fn)          # or: await pending.wait_async(async_poll)

An explicit poll_fn always wins over the wired one. A PendingApproval you construct yourself has neither and raises TypeError until you pass one.

client.get_secret_poll_fn() remains available as an explicit poller for the secret-read approval path.

Mock backend for tests

The SDK ships an in-process MockApprovalBackend so tests can exercise approval paths without standing up the real backend. Wire it into the polling loop by passing poll_fn:

from controlzero import PendingApproval
from controlzero.hitl.mock import MockApprovalBackend

backend = MockApprovalBackend("approve_after_2s", delay_s=0.05)
created = backend.create_request({"canonical_action": "delete_file"})
pending = PendingApproval(
    request_id=created["request_id"],
    idempotency_key="test-key",
    status="pending",
    created_at=created["created_at"],
    expires_at=created["expires_at"],
)
resolved = pending.wait(poll_fn=lambda rid: backend.get_request(rid))
assert resolved.status == "approved"

The five supported modes are approve_after_2s, approve_timed_after_2s, approve_forever_after_2s, deny_after_2s, and timeout.

Identity requirement

Every approval request must carry the operator email so the backend can route to a real person and stamp identity provenance on the grant. Set it once via the CLI:

controlzero install <agent> --email you@example.com

If the email is missing, request_approval() raises HITLIdentityRequired (E1707) before any HTTP traffic.

Secret reads with approvals

When a policy gates a secret behind approval, client.get_secret(name) raises SecretApprovalRequired (E1710) carrying a pending attribute the caller waits on:

from controlzero import PolicyDeniedError
from controlzero.errors import SecretApprovalRequired

try:
    value = cz.get_secret("PROD_DB_PASSWORD")
except SecretApprovalRequired as exc:
    # `pending` came from request_approval(), so it carries the poller
    # and wait() needs no argument. It raises PolicyDeniedError on a deny
    # or a timeout rather than returning a non-approved status.
    try:
        exc.pending.wait()
    except PolicyDeniedError:
        raise  # abort
    value = cz.get_secret("PROD_DB_PASSWORD")  # retry now that the grant exists

Exception classes

The 11 approval-related exception codes raised by this surface. Class names retain the HITL prefix because they are part of the stable public SDK API:

Full reference and runbooks: docs.controlzero.ai/docs/concepts/hitl-approval.

License

Apache 2.0