Python SDK
対応モード: Hosted Hybrid Local 利用可能なプラン: Free Solo Teams
英語の原文 -- 翻訳は技術レビュー待ちです
The Control Zero Python SDK provides deterministic policy enforcement for AI agents running in Python environments. A deny stops the guarded call before it executes. Hosted policies arrive as signed bundles that are verified before use, and the SDK fails closed by default when it cannot establish coverage.
変更できないアプリケーションにはゲートウェイプロキシを、開発者向けの AI ツールにはコーディングフックを使用します。Python アプリケーションのコードを自分で管理できる場合に、この SDK を使用します。
インストール
pip install controlzero
要件:
- Python 3.9 以降
- 追加のシステム依存関係なし
クイックスタート
デプロイモードを選択します。
- Hosted
- Hybrid
- Local
Hosted ポリシーはダッシュボードから取得されます。
英語の原文 -- 翻訳は技術レビュー待ちです
Audit in cloud.
ほとんどのチームに推奨します。
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_..."
インストール: pip install controlzero(クラウドモードの依存関係は 1.4.3 以降、標準のインストールに含まれています)。
Hybrid API キーとローカルのポリシーファイルを組み合わせます。
英語の原文 -- 翻訳は技術レビュー待ちです
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.
この選択が分かるように、SDK は 1 回だけ警告を出力します(代わりに HybridModeError を発生させるには strict_hosted=True を渡します)。CONTROLZERO_LOCAL_OVERRIDE は、この明示的に引数を渡すケースには影響しません。影響するのは自動検出の場合のみです(api_key が設定されていて policy/policy_file を渡していないが、CONTROLZERO_POLICY_FILE または作業ディレクトリの controlzero.yaml/.yml/.json からローカルファイルが見つかった場合)。この場合はデフォルトでホスト型バンドルが優先され、CONTROLZERO_LOCAL_OVERRIDE=1 を設定すると、検出されたローカルファイルが優先されます。
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.
SDK は、このインスタンスではダッシュボードのポリシーが無視されることを示す警告を 1 回だけ出力します(strict_hosted=True を設定した場合は HybridModeError を発生させます)。
Local API キー不要。
英語の原文 -- 翻訳は技術レビュー待ちです
No network calls.
完全オフライン。エアギャップ環境に対応します。
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")
英語の原文 -- 翻訳は技術レビュー待ちです
Every decision and its event-level coverage is written to ./controlzero.log, so “did not run” remains distinguishable from “ran and found nothing.”
設定
Hosted Hosted モード(推奨)
英語の原文 -- 翻訳は技術レビュー待ちです
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 ローカルポリシーファイルを使う
from controlzero import Client
client = Client(policy_file="controlzero.yaml")
Hybrid Hybrid モード(API キー + ローカルポリシー)
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")
インラインポリシー
from controlzero import Client
client = Client(policy={
"rules": [
{"allow": "database:query", "reason": "Reads are permitted"},
{"deny": "database:*", "reason": "All other database operations are blocked"},
]
})
database:query、database:execute、database:delete は従来のアクション名で、正式な database:read、database:write、database:admin クラスと同じ呼び出しに一致します。新しいポリシーでは正式な名前を使うことを推奨しますが、従来の名前を使った既存のルールも変更なしで引き続き動作します。SQL のセマンティッククラスの対応については、読み取り専用データベースのレシピを参照してください。
ワイルドカードによるリソースの一致
resources: ["*"] を指定したルールは、呼び出し側が context.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"})
ワイルドカードでないリソースパターン(例: resources: ["table/orders"])では、ルールが働くために、呼び出し側が一致する context.resource を渡す必要があります。これにより、範囲の狭いルールは狭いままに保たれます。
環境変数
API キーは明示的に渡すことも、環境変数 CONTROLZERO_API_KEY で設定することもできます。エージェント名は 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
設定オプション
| パラメータ | 型 | デフォルト | 説明 |
|---|---|---|---|
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 | ルールを含むインラインのポリシー辞書。 |
policy_file | str | None | YAML または JSON のポリシーファイルへのパス。 |
strict_hosted | bool | False | api_key が明示的な policy/policy_file と一緒に渡された場合に、1 回だけ警告を出してローカルポリシーを使う代わりに HybridModeError を発生させます。意図しないローカルでの上書きを CI で検出するのに便利です。 |
refresh_interval_seconds | int | 60 | Hosted モード: ダッシュボードに新しいポリシーバンドルがないか確認する間隔。自動更新を無効にするには None を渡します。guard() の呼び出しごとに更新するには 0 を渡します(テスト専用)。 |
log_path | str | "./controlzero.log" | ローカルの監査ログファイルのパス。 |
log_rotation | str | "daily" | 監査ログのローテーション間隔。 |
log_retention | str | "30 days" | How long to keep rotated logs. |
log_compression | str | None | ローテーション済みログを圧縮します(例: "gz")。 |
log_format | str | "json" | 監査ログの形式。 |
ポリシーの更新(Hosted モード)
ダッシュボードでポリシーを更新すると、長時間動作している SDK プロセスは、更新間隔(デフォルト: 60 秒)以内に変更を自動的に取り込みます。SDK は条件付きリクエストを発行するため、バンドルが変わっていない場合の通信コストはわずかな往復だけです。
設定項目は 3 つあります。
refresh_interval_seconds=60(デフォルト): 1 分ごとに確認します。refresh_interval_seconds=None: バックグラウンドでの確認を無効にします。client.refresh()と組み合わせると、完全に手動で制御できます。client.refresh(): 即座に再読み込みします。バンドルが実際に変更された場合はTrue、ダッシュボードに更新がなかった場合はFalseを返します。
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.
基本的な使い方
ツール呼び出しの評価
主要なメソッドは guard() で、有効なポリシーを評価します。
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
拒否されたアクションの処理
ポリシーがアクションを拒否した場合は、判定の effect を確認するか、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}")
コンテキストマネージャ
Client はコンテキストマネージャのプロトコルを実装しています。with を使うと、終了時に close() が呼び出されます。
from controlzero import Client
with Client(policy_file="controlzero.yaml") as client:
decision = client.guard("github", method="list_issues", args={"repo": "acme/app"})
API リファレンス
Client
メインの同期クライアントクラスです。
__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")
新しい Control Zero クライアントを作成します。
policy と policy_file の両方が指定された場合は ValueError を発生させます。
guard(tool, args=None, method="*", raise_on_deny=False, context=None) -> PolicyDecision
読み込まれたポリシーに対してツール呼び出しを評価します。
パラメータ:
| 名前 | 型 | 必須 | デフォルト | 説明 |
|---|---|---|---|---|
tool | str | はい | - | ツール名。method と組み合わせてアクションになります。 |
args | dict | いいえ | None | DLP スキャンと条件評価に使う引数。 |
method | str | いいえ | "*" | メソッド名。評価されるアクションは "{tool}:{method}" です。 |
raise_on_deny | bool | いいえ | False | True の場合、拒否の判定で PolicyDeniedError を発生させます。 |
context | dict | いいえ | None | 一致判定に使う、resource と tags キーを持つ任意のコンテキスト。 |
戻り値: effect、reason、policy_id、dlp_findings を持つ PolicyDecision。
close() -> None
英語の原文 -- 翻訳は技術レビュー待ちです
Flushes buffered audit logs, wipes secrets from memory, and closes the HTTP connection.
PolicyDecision
guard() が返します。
| 属性 | 型 | 説明 |
|---|---|---|
effect | str | "allow" または "deny"。 |
policy_id | str or None | 一致したポリシー(ある場合)。 |
reason | str | 人が読める説明。 |
dlp_findings | list | 引数内で見つかった DLP の一致のリスト。 |
PolicyDeniedError
ポリシーがアクションを拒否したときに発生することがあります。
英語の原文 -- 翻訳は技術レビュー待ちです
The tool is never invoked.
| 属性 | 型 | 説明 |
|---|---|---|
e.decision | PolicyDecision | 完全な判定オブジェクト。 |
e.decision.effect | str | 常に "deny"。 |
e.decision.reason | str | 判定内容の人が読める説明。 |
e.decision.policy_id | str or None | 一致したポリシーの ID(ある場合)。 |
BundleSignatureError
英語の原文 -- 翻訳は技術レビュー待ちです
Raised when a policy bundle fails cryptographic signature verification. The SDK will not load a tampered bundle.
エラー処理
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()
MCP での使用
Control Zero は MCP のツール呼び出しを統制するよう設計されています。MCP サーバー名とツール名を、そのまま 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}")
MCP ガバナンスのパターンの詳しい手順については、MCP ツール呼び出しの統制を参照してください。
LLM プロバイダーラッパー
Python SDK には、一般的な LLM プロバイダーのクライアントにガバナンスを追加する、単体で使えるラッパー関数が含まれています。各ラッパーは呼び出しをインターセプトし、ポリシーを評価して、プロバイダー本来の API の形を変えずに判定を記録します。
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"}],
)
利用可能なラッパー
| 関数 | プロバイダー | インポート パス |
|---|---|---|
wrap_openai() | OpenAI | controlzero.integrations.openai |
wrap_anthropic() | Anthropic (Claude) | controlzero.integrations.anthropic |
wrap_google() | Google AI (Gemini) | controlzero.integrations.google |
すべてのラッパーは事前チェック(モデルのブロック、コスト見積もり、PII 検出)を適用し、リクエストを監査証跡に記録します。ポリシーがリクエストを拒否した場合は、呼び出しがプロバイダーに届く前に PolicyDeniedError が発生します。