メインコンテンツまでスキップ
このページは機械翻訳であり、十分なレビューを経ていません。英語の原文が正となります。セキュリティ、プライバシー、データの取り扱い、コンプライアンス、ライセンスに関する記述は、技術レビューが完了するまで英語のまま掲載しています。 英語の原文を読む

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 ポリシーはダッシュボードから取得されます。

英語の原文 -- 翻訳は技術レビュー待ちです

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 以降、標準のインストールに含まれています)。

設定​

英語の原文 -- 翻訳は技術レビュー待ちです

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_keystrCONTROLZERO_API_KEY envProject API key. Enables hosted mode: SDK auto-pulls the signed policy bundle from the dashboard and ships audit to the remote trail.
policydictNoneルールを含むインラインのポリシー辞書。
policy_filestrNoneYAML または JSON のポリシーファイルへのパス。
strict_hostedboolFalseapi_key が明示的な policy/policy_file と一緒に渡された場合に、1 回だけ警告を出してローカルポリシーを使う代わりに HybridModeError を発生させます。意図しないローカルでの上書きを CI で検出するのに便利です。
refresh_interval_secondsint60Hosted モード: ダッシュボードに新しいポリシーバンドルがないか確認する間隔。自動更新を無効にするには None を渡します。guard() の呼び出しごとに更新するには 0 を渡します(テスト専用)。
log_pathstr"./controlzero.log"ローカルの監査ログファイルのパス。
log_rotationstr"daily"監査ログのローテーション間隔。
log_retentionstr"30 days"How long to keep rotated logs.
log_compressionstrNoneローテーション済みログを圧縮します(例: "gz")。
log_formatstr"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​

読み込まれたポリシーに対してツール呼び出しを評価します。

パラメータ:

名前型必須デフォルト説明
toolstrはい-ツール名。method と組み合わせてアクションになります。
argsdictいいえNoneDLP スキャンと条件評価に使う引数。
methodstrいいえ"*"メソッド名。評価されるアクションは "{tool}:{method}" です。
raise_on_denyboolいいえFalseTrue の場合、拒否の判定で PolicyDeniedError を発生させます。
contextdictいいえ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() が返します。

属性型説明
effectstr"allow" または "deny"。
policy_idstr or None一致したポリシー(ある場合)。
reasonstr人が読める説明。
dlp_findingslist引数内で見つかった DLP の一致のリスト。

PolicyDeniedError​

ポリシーがアクションを拒否したときに発生することがあります。

英語の原文 -- 翻訳は技術レビュー待ちです

The tool is never invoked.

属性型説明
e.decisionPolicyDecision完全な判定オブジェクト。
e.decision.effectstr常に "deny"。
e.decision.reasonstr判定内容の人が読める説明。
e.decision.policy_idstr 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()OpenAIcontrolzero.integrations.openai
wrap_anthropic()Anthropic (Claude)controlzero.integrations.anthropic
wrap_google()Google AI (Gemini)controlzero.integrations.google

すべてのラッパーは事前チェック(モデルのブロック、コスト見積もり、PII 検出)を適用し、リクエストを監査証跡に記録します。ポリシーがリクエストを拒否した場合は、呼び出しがプロバイダーに届く前に PolicyDeniedError が発生します。