Python SDK

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

Installation

pip install control-zero

Requirements:

Python 3.9 or later
No additional system dependencies

Configuration

Basic Configuration

from control_zero import ControlZeroClient

client = ControlZeroClient(api_key="cz_live_your_api_key_here")
client.initialize()

Environment Variables

You can pass the API key explicitly or set it via the CZ_AGENT_NAME environment variable for the agent name. The API key must always be passed directly to the constructor.

export CZ_AGENT_NAME="my-analyst-agent"

import os
from control_zero import ControlZeroClient

client = ControlZeroClient(api_key=os.environ["CZ_API_KEY"])
client.initialize()

Configuration Options

Parameter	Type	Default	Description
`api_key`	`str`	--	Your project API key. Must start with `cz_live_` or `cz_test_`. Required.
`agent_name`	`str`	`CZ_AGENT_NAME` env or `"default-agent"`	Name used to identify this agent in audit logs and policy conditions.
`base_url`	`str`	`"https://api.controlzero.ai"`	Control Zero API base URL.
`enable_governance`	`bool`	`True`	If `False`, governance is skipped entirely (pass-through mode).
`policy_mode`	`str`	`"cloud"`	`"cloud"` fetches from server; `"local"` loads from file; `"hybrid"` uses both.
`policy_path`	`str`	`None`	Path to a `.czpolicy` bundle file. Required when `policy_mode` is `"local"` or `"hybrid"`.
`rust_engine`	`bool`	`True`	Use the compiled Rust policy engine when available. Falls back to Python if absent.
`refresh_interval_seconds`	`int`	`300`	How often the background refresh pulls a new bundle from the server.
`max_cache_age_seconds`	`int`	`86400`	Maximum age of a cached policy bundle before it is considered stale.

Advanced Configuration

from control_zero import ControlZeroClient

client = ControlZeroClient(
    api_key="cz_live_your_api_key_here",
    agent_name="finance-agent",
    policy_mode="hybrid",
    policy_path="/var/cache/my-agent.czpolicy",
    refresh_interval_seconds=60,
    rust_engine=True,
)
client.initialize()

Basic Usage

Calling a Tool

The primary method is call_tool(), which evaluates the active policies and then executes the tool call:

from control_zero import ControlZeroClient, PolicyDeniedError

with ControlZeroClient(api_key="cz_live_your_api_key_here") as client:
    result = client.call_tool(
        "database",
        "query",
        {"sql": "SELECT * FROM orders"},
        context={"user_id": "agent-001", "user_group": "analysts"},
    )
    print(result)

The context dict is optional. It is used for conditional policy rules that match on user_id, user_group, or custom tags.

Handling Denied Actions

When a policy denies the action, call_tool() raises PolicyDeniedError before the tool is invoked:

from control_zero import ControlZeroClient, PolicyDeniedError

with ControlZeroClient(api_key="cz_live_your_api_key_here") as client:
    try:
        client.call_tool(
            "filesystem",
            "write_file",
            {"path": "/data/output.csv", "content": "..."},
        )
    except PolicyDeniedError as e:
        print(f"Blocked: {e}")
        print(f"Reason: {e.decision.reason}")
        print(f"Policy: {e.decision.policy_id}")

Refreshing Policies

Force an immediate download of the latest policy bundle from the server:

client.sync_policies()

Loading a Local Bundle

Load a policy bundle from a .czpolicy file on disk:

client.load_policy_bundle("/var/cache/my-agent.czpolicy")

Context Manager

ControlZeroClient implements the context manager protocol. Using with calls initialize() on entry and close() on exit:

from control_zero import ControlZeroClient, PolicyDeniedError

with ControlZeroClient(api_key="cz_live_your_api_key_here") as client:
    result = client.call_tool("github", "list_issues", {"repo": "acme/app"})

Async Client

An async variant is also available for use in async frameworks:

import asyncio
from control_zero import AsyncControlZeroClient, PolicyDeniedError

async def main():
    async with AsyncControlZeroClient(api_key="cz_live_your_api_key_here") as client:
        try:
            result = await client.call_tool(
                "database",
                "query",
                {"sql": "SELECT * FROM orders"},
            )
            print(result)
        except PolicyDeniedError as e:
            print(f"Denied: {e.decision.reason}")

asyncio.run(main())

AsyncControlZeroClient accepts api_key, agent_name, and base_url. Its call_tool() and initialize() methods are coroutines.

API Reference

`ControlZeroClient`

The main synchronous client class.

`init(api_key, **kwargs)`

Creates a new Control Zero client. See Configuration Options for all parameters.

Raises ValueError immediately if api_key does not start with cz_live_ or cz_test_.

`initialize() -> None`

Fetches the session config, secrets, and policy bundle from the Control Zero server. Must be called before call_tool().

In local mode, loads the bundle from policy_path without contacting the server.

`call_tool(tool, method, arguments=None, context=None) -> Any`

Evaluates policy for "{tool}:{method}" and executes the tool call if allowed.

Parameters:

Name	Type	Required	Description
`tool`	`str`	Yes	Tool name. Combined with `method` to form the policy action string.
`method`	`str`	Yes	Method name.
`arguments`	`dict`	No	Arguments forwarded to the tool.
`context`	`dict`	No	Identity context (`user_id`, `user_group`, `tags`) for conditional policy evaluation.

Returns: The tool result.

Raises: PolicyDeniedError if the active policy denies the action. RuntimeError if initialize() has not been called. ValueError if the tool is not found in the session config.

`sync_policies() -> bool`

Pulls the latest policy bundle from the server immediately. Returns True on success.

`load_policy_bundle(path) -> bool`

Loads a policy bundle from a local .czpolicy file. Returns True on success.

`check_llm_request(model, provider, estimated_tokens, functions=None, has_pii=False) -> dict | None`

Checks an LLM request against governance policies using the Rust engine. Returns a decision dict or None if the Rust engine is not available.

`process_pii(text, action="detect") -> dict | None`

Processes text for PII detection using the Rust engine. Returns a result dict or None if the Rust engine is not available.

`flush() -> None`

Flushes all buffered audit log entries to the server immediately.

`close() -> None`

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

`has_rust_engine -> bool`

Property. True if the compiled Rust policy engine is loaded and available.

`AsyncControlZeroClient`

Async variant of the client. Accepts api_key, agent_name, and base_url. All methods are coroutines. Supports async with.

`PolicyDeniedError`

Raised by call_tool() when an active policy denies the 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.

`PolicyDecision`

Returned internally and attached to PolicyDeniedError. Also returned by the Rust engine methods.

Attribute	Type	Description
`effect`	`str`	`"allow"`, `"deny"`, `"warn"`, or `"audit"`.
`policy_id`	`str`	The policy that matched, or `None`.
`reason`	`str`	Human-readable explanation.
`evaluated_rules`	`int`	Number of rules evaluated before a decision was made.
`evaluation_time_ns`	`int`	Evaluation time in nanoseconds.

`BundleVerificationError`

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

`BundleTamperError`

Raised when a policy bundle's integrity check detects modification after signing.

Error Handling

from control_zero import ControlZeroClient, PolicyDeniedError, BundleVerificationError, BundleTamperError

client = ControlZeroClient(api_key="cz_live_your_api_key_here")

try:
    client.initialize()
except BundleVerificationError:
    # Policy bundle failed Ed25519 signature check -- reject it
    raise
except BundleTamperError:
    # Bundle was modified after signing -- reject it
    raise
except Exception as e:
    # Network error, auth failure, or other initialization problem
    raise

try:
    result = client.call_tool("database", "query", {"sql": "SELECT 1"})
except PolicyDeniedError as e:
    print(f"Policy denied: {e.decision.reason}")
except RuntimeError as e:
    # call_tool() called before initialize()
    print(f"Client not ready: {e}")
finally:
    client.close()

Usage with MCP

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

from control_zero import ControlZeroClient, PolicyDeniedError

with ControlZeroClient(api_key="cz_live_your_api_key_here") as client:
    # Policy controls which MCP tools the agent can invoke
    try:
        result = client.call_tool(
            "filesystem",          # MCP server name
            "read_file",           # MCP tool name
            {"path": "/data/report.pdf"},
            context={"user_id": "agent-001"},
        )
    except PolicyDeniedError as e:
        print(f"MCP tool blocked: {e.decision.reason}")

For a detailed walkthrough of MCP governance patterns, see MCP Tool Control.

Installation​

Configuration​

Basic Configuration​

Environment Variables​

Configuration Options​

Advanced Configuration​

Basic Usage​

Calling a Tool​

Handling Denied Actions​

Refreshing Policies​

Loading a Local Bundle​

Context Manager​

Async Client​

API Reference​

ControlZeroClient​

__init__(api_key, **kwargs)​

initialize() -> None​

call_tool(tool, method, arguments=None, context=None) -> Any​

sync_policies() -> bool​

load_policy_bundle(path) -> bool​

check_llm_request(model, provider, estimated_tokens, functions=None, has_pii=False) -> dict | None​

process_pii(text, action="detect") -> dict | None​

flush() -> None​

close() -> None​

has_rust_engine -> bool​

AsyncControlZeroClient​

PolicyDeniedError​

PolicyDecision​

BundleVerificationError​

BundleTamperError​

Error Handling​

Usage with MCP​