controlzero 1.8.2

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

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 migration guide 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 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 IDE) 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 ./controlzero.yaml exists in the current directory, it is picked up automatically. No environment variable needed.

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).

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: '*'

Mode	Behavior
warn	Log a warning but continue evaluating rules normally.
deny	Deny the current tool call that triggered the tamper check.
deny-all	Deny all tool calls and place the machine in quarantine until recovered.
quarantine	Same as deny-all, plus report a tamper alert to the backend dashboard.

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

When running without an API key, every decision is written to ./controlzero.log with daily rotation and 30-day retention. Tail it:

controlzero tail

Configure rotation via the client:

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"
)

When CONTROLZERO_API_KEY is set, audit ships to the remote dashboard and these log_* options are ignored with a warning.

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/sdk/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. An agent calls client.request_approval(decision, ...) whenever guard() returns a deny that is tagged escalate_on_deny: true, then waits on the returned PendingApproval for the human to respond.

Basic flow:

from controlzero import Client

cz = Client(api_key="cz_live_...")  # approvals require hosted mode

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

    # PendingApproval.wait() requires you to inject a `poll_fn`,
    # a callable that returns the latest backend snapshot of the
    # approval request. In 1.6.0 the SDK does NOT ship a built-in
    # HTTP poller; you wire one yourself (or use the helper that
    # ships with get_secret. See "Secret reads with approvals" below).
    import httpx
    api_url = "https://api.controlzero.ai"  # or your self-managed host

    def poll_fn(request_id: str) -> dict:
        resp = httpx.get(
            f"{api_url}/api/approval-requests/{request_id}",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=10,
        )
        resp.raise_for_status()
        return resp.json()

    # Block until the human approves, denies, or the SLA expires.
    resolved = pending.wait(poll_fn)
    if resolved.status == "approved":
        # proceed with the gated action
        ...
    else:
        # denied or timed_out, abort the tool call
        ...

wait() blocks the calling thread. For async code, use wait_async(), same contract, but the poll callable can be async def or a sync function (sync calls are dispatched to a thread executor so the event loop never blocks):

async def async_poll(request_id: str) -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{api_url}/api/approval-requests/{request_id}",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=10,
        )
        resp.raise_for_status()
        return resp.json()

resolved = await pending.wait_async(async_poll)

1.7.0 ships client.get_secret_poll_fn() as a built-in poller helper for the secret-read approval path. The general request_approval() flow still takes an explicit poll_fn so callers can plug their own HTTP stack.

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 [e-mail]

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.errors import SecretApprovalRequired

try:
    value = cz.get_secret("PROD_DB_PASSWORD")
except SecretApprovalRequired as exc:
    resolved = exc.pending.wait()
    if resolved.status == "approved":
        value = cz.get_secret("PROD_DB_PASSWORD")  # retry now that grant exists
    else:
        raise  # abort

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:

Code	Class	Meaning
E1701	HITLTimeoutError	Approver did not decide before timeout_s elapsed.
E1702	HITLBackendUnreachableError	POST to the approval endpoint failed after retries.
E1703	HITLPolicyVersionConflictError	SDK bundle is missing the rule that triggered the request.
E1704	HITLNotConfiguredError	Org has no approval settings row configured.
E1705	HITLNoApproverAvailable	Approver pool is empty or no member is active.
E1706	HITLIdentityNotInOrg	Operator email is not a member of the API key's org.
E1707	HITLIdentityRequired	No operator email set on this install.
E1708	HITLIdentityClaimRejected	Backend rejected the identity claim.
E1709	SecretValueLeakInPayload	Outbound payload contains a secret-shaped string. Aborted.
E1710	SecretApprovalRequired	Secret read requires approval; wait on exc.pending.
E1711	SecretNotFound	Named secret does not exist in the configured vault.

Full reference and runbooks: docs.controlzero.ai/hitl.

License

Apache 2.0