Python SDK

Supported modes: Hosted Hybrid Local Available in: Free Solo Teams

When NOT to use this

If you are not changing your application code, the gateway proxy gives you governance with zero code changes. If you are governing developer AI tools (Claude Code, Cursor, Codex CLI), use coding hooks instead.

The Control Zero Python SDK provides policy enforcement for AI agents running in Python environments.

Installation

pip install controlzero

Requirements:

• Python 3.9 or later
• No additional system dependencies

Quickstart

Choose your deployment mode:

Hosted

Hosted Policy pulled from dashboard. Audit in cloud. Recommended for most teams.

from controlzero import Client

# SDK fetches your signed policy bundle on first call.
# Manages audit automatically. No local config needed.
client = Client(api_key="cz_live_your_api_key_here")
# Or use env var: export CONTROLZERO_API_KEY="cz_live_..."

Install: pip install controlzero (cloud mode dependencies ship in the base install as of 1.4.3).

Hybrid

Hybrid API key plus a local policy file. When you pass policy or policy_file explicitly to Client() alongside an api_key, your local policy governs enforcement and the hosted/dashboard bundle is IGNORED for that instance -- audit still ships to your dashboard. The SDK prints a one-time warning so the choice is visible (pass strict_hosted=True to raise HybridModeError instead). CONTROLZERO_LOCAL_OVERRIDE does not apply to this explicit-argument case; it only affects auto-discovery (an api_key is set and you do NOT pass policy/policy_file, but a local file is found via CONTROLZERO_POLICY_FILE or controlzero.yaml/.yml/.json in the working directory), where the hosted bundle wins by default and CONTROLZERO_LOCAL_OVERRIDE=1 flips to the discovered local file.

from controlzero import Client

# Your local file governs enforcement; the hosted bundle is ignored.
client = Client(
    api_key="cz_live_your_api_key_here",
    policy_file="controlzero.yaml",
)

# The local file already governs enforcement; the API key only routes audit to the dashboard.

Your local policy governs enforcement; the API key only routes audit to the dashboard. The SDK prints a one-time warning noting the dashboard policy is ignored for this instance (or raises HybridModeError if you set strict_hosted=True).

Local

Local No API key. No network calls. Fully offline. Air-gap compatible.

from controlzero import Client

# Inline policy -- no file needed
client = Client(policy={
    "rules": [
        {"allow": "database:query",   "reason": "Reads are permitted"},
        {"deny":  "database:execute", "reason": "Writes are blocked"},
    ]
})

# Or from a file
client = Client(policy_file="controlzero.yaml")

Audit written to ./controlzero.log.

Configuration

Hosted Hosted Mode (Recommended)

Pass only your project API key. The SDK pulls the signed policy bundle from the dashboard at first call, verifies its signature, decrypts locally, and enforces every call against the dashboard policy. Audit entries ship to the remote trail automatically.

from controlzero import Client

client = Client(api_key="cz_live_your_api_key_here")

Local With a Local Policy File

from controlzero import Client

client = Client(policy_file="controlzero.yaml")

Hybrid Hybrid Mode (API Key + Local Policy)

from controlzero import Client

# Explicit policy_file + api_key: the local file governs enforcement,
# the hosted bundle is IGNORED, audit still ships remotely.
client = Client(api_key="cz_live_your_api_key_here", policy_file="controlzero.yaml")

Inline Policy

from controlzero import Client

client = Client(policy={
    "rules": [
        {"allow": "database:query", "reason": "Reads are permitted"},
        {"deny": "database:*", "reason": "All other database operations are blocked"},
    ]
})

Legacy and canonical action names both work

database:query, database:execute, and database:delete are legacy action names that match the same calls as the canonical database:read, database:write, and database:admin classes. New policies should prefer the canonical names; existing rules with the legacy names continue to work without changes. See Read-only database recipe for the SQL semantic class mapping.

Wildcard resource matching

A rule that lists resources: ["*"] matches every call regardless of whether the caller passed a context.resource. Use this when a rule should apply universally and you do not have (or do not want to pass) a per-call resource:

# Policy rule with universal resource matching
{"effect": "allow", "actions": ["database:read"], "resources": ["*"]}

# This call matches even though context.resource is not set:
cz.guard("database", method="query", args={"sql": "SELECT 1"})

Non-wildcard resource patterns (for example, resources: ["table/orders"]) still require the caller to supply a matching context.resource for the rule to fire. This keeps narrow rules narrow.

Environment Variables

You can pass the API key explicitly or set it via the CONTROLZERO_API_KEY environment variable. The agent name can be set via CZ_AGENT_NAME.

export CONTROLZERO_API_KEY="cz_live_your_api_key_here"
export CZ_AGENT_NAME="my-analyst-agent"

from controlzero import Client

client = Client()  # reads CONTROLZERO_API_KEY from environment

Configuration Options

Parameter	Type	Default	Description
api_key	str	CONTROLZERO_API_KEY env	Project API key. Enables hosted mode: SDK auto-pulls the signed policy bundle from the dashboard and ships audit to the remote trail.
policy	dict	None	Inline policy dict with rules.
policy_file	str	None	Path to a YAML or JSON policy file.
strict_hosted	bool	False	Raise HybridModeError when an api_key is passed together with an explicit policy/policy_file, instead of using the local policy with a one-time warning. Useful in CI to catch unintended local overrides.
refresh_interval_seconds	int	60	Hosted-mode: how often to check the dashboard for a fresh policy bundle. Pass None to disable auto-refresh. Pass 0 to refresh on every guard() call (tests only).
log_path	str	"./controlzero.log"	Local audit log file path.
log_rotation	str	"daily"	Audit log rotation interval.
log_retention	str	"30 days"	How long to keep rotated logs.
log_compression	str	None	Compress rotated logs (e.g. "gz").
log_format	str	"json"	Audit log format.

Policy refresh (hosted mode)

When you update a policy on the dashboard, long-running SDK processes automatically pick up the change within the refresh interval (default: 60 seconds). The SDK issues a conditional request so an unchanged bundle costs only a small roundtrip.

Three knobs:

• refresh_interval_seconds=60 (default): check every minute.
• refresh_interval_seconds=None: disable the background check. Combine with client.refresh() for fully manual control.
• client.refresh(): force an immediate reload. Returns True if the bundle actually changed, False if the dashboard had no updates.

from controlzero import Client

# Long-lived agent, picks up dashboard changes within 60s automatically.
client = Client(api_key="cz_live_your_api_key_here")

# Or: force an immediate reload after a known dashboard change.
changed = client.refresh()
if changed:
    print(f"Policy updated at {client.last_refreshed_at.isoformat()}")

Network errors during a background refresh are logged at WARNING level and do not raise; the last-known-good bundle keeps working until the next successful pull.

Basic Usage

Evaluating a Tool Call

The primary method is guard(), which evaluates the active policies:

from controlzero import Client, PolicyDeniedError

client = Client(policy_file="controlzero.yaml")

decision = client.guard(
    "database",
    method="query",
    args={"sql": "SELECT * FROM orders"},
)
print(decision.effect)     # "allow" or "deny"
print(decision.reason)     # human-readable reason
print(decision.policy_id)  # matching policy ID

Handling Denied Actions

When a policy denies the action, check the decision effect or catch PolicyDeniedError:

from controlzero import Client, PolicyDeniedError

client = Client(policy_file="controlzero.yaml")

decision = client.guard(
    "filesystem",
    method="write_file",
    args={"path": "/data/output.csv", "content": "..."},
)
if decision.effect == "deny":
    print(f"Blocked: {decision.reason}")
    print(f"Policy: {decision.policy_id}")

Context Manager

Client implements the context manager protocol. Using with calls close() on exit:

from controlzero import Client

with Client(policy_file="controlzero.yaml") as client:
    decision = client.guard("github", method="list_issues", args={"repo": "acme/app"})

API Reference

Client

The main synchronous client class.

__init__(api_key=None, policy=None, policy_file=None, strict_hosted=False, log_path="./controlzero.log", log_rotation="daily", log_retention="30 days", log_compression=None, log_format="json")

Creates a new Control Zero client.

Raises ValueError if both policy and policy_file are provided.

guard(tool, args=None, method="*", raise_on_deny=False, context=None) -> PolicyDecision

Evaluates a tool call against the loaded policy.

Parameters:

Name	Type	Required	Default	Description
tool	str	Yes	-	Tool name. Combined with method to form the action.
args	dict	No	None	Arguments for DLP scanning and conditional evaluation.
method	str	No	"*"	Method name. The action evaluated is "{tool}:{method}".
raise_on_deny	bool	No	False	If True, raises PolicyDeniedError on a deny decision.
context	dict	No	None	Optional context with resource and tags keys for matching.

Returns: PolicyDecision with effect, reason, policy_id, and dlp_findings.

close() -> None

Flushes buffered audit logs, wipes secrets from memory, and closes the HTTP connection.

PolicyDecision

Returned by guard().

Attribute	Type	Description
effect	str	"allow" or "deny".
policy_id	str or None	The policy that matched, if any.
reason	str	Human-readable explanation.
dlp_findings	list	List of DLP matches found in the arguments.

PolicyDeniedError

Can be raised when a policy denies an action. The tool is never invoked.

Attribute	Type	Description
e.decision	PolicyDecision	The full decision object.
e.decision.effect	str	Always "deny".
e.decision.reason	str	Human-readable explanation of the decision.
e.decision.policy_id	str or None	ID of the policy that matched, if any.

BundleSignatureError

Raised when a policy bundle fails cryptographic signature verification. The SDK will not load a tampered bundle.

Error Handling

from controlzero import Client, PolicyDeniedError

client = Client(policy_file="controlzero.yaml")

decision = client.guard("database", method="query", args={"sql": "SELECT 1"})
if decision.effect == "deny":
    print(f"Policy denied: {decision.reason}")

client.close()

Usage with MCP

Control Zero is designed to govern MCP tool calls. Pass the MCP server name and tool name directly to guard():

from controlzero import Client

with Client(api_key="cz_live_your_api_key_here") as client:
    decision = client.guard(
        "filesystem",          # MCP server name
        method="read_file",    # MCP tool name
        args={"path": "/data/report.pdf"},
    )
    if decision.effect == "deny":
        print(f"MCP tool blocked: {decision.reason}")

For a detailed walkthrough of MCP governance patterns, see Governing MCP tool calls.

LLM Provider Wrappers

The Python SDK includes standalone wrapper functions that add governance to popular LLM provider clients. Each wrapper intercepts calls, evaluates policies, and logs decisions without changing the provider's native API surface.

OpenAI

from controlzero import Client
from controlzero.integrations.openai import wrap_openai
import openai

client = Client(api_key="cz_live_your_api_key_here")
wrapped = wrap_openai(openai.OpenAI(), client)

response = wrapped.chat.completions.create(
    model="gpt-5.4",
    messages=[{"role": "user", "content": "Hello"}],
)

Google AI (Gemini)

from controlzero import Client
from controlzero.integrations.google import wrap_google
from google import genai

client = Client(api_key="cz_live_your_api_key_here")
google_client = genai.Client(api_key="your-google-ai-key")
wrapped_model = wrap_google(google_client.models, client)

response = wrapped_model.generate_content(
    model="gemini-2.0-flash",
    contents="Summarize the quarterly report",
)
print(response.text)

Anthropic

from controlzero import Client
from controlzero.integrations.anthropic import wrap_anthropic
import anthropic

client = Client(api_key="cz_live_your_api_key_here")
wrapped = wrap_anthropic(anthropic.Anthropic(), client)

message = wrapped.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
)

Available Wrappers

Function	Provider	Import Path
wrap_openai()	OpenAI	controlzero.integrations.openai
wrap_anthropic()	Anthropic (Claude)	controlzero.integrations.anthropic
wrap_google()	Google AI (Gemini)	controlzero.integrations.google

All wrappers apply pre-flight checks (model blocking, cost estimation, PII detection) and log the request to the audit trail. If a policy denies the request, PolicyDeniedError is raised before the call reaches the provider.