Python SDK — Deep Reference

Companion to the short plugins/python/README.md. This page covers the details you need when things go sideways: fail modes, timeouts, exception semantics, adapter internals, and testing patterns.

Package on PyPI: vanzik. Import name: vanzik. Before 2.0 the package was agentguardproxy, imported as agentguard (see MIGRATION.md).

Source: plugins/python/vanzik/core.py (Guard client, exceptions, constants), plugins/python/vanzik/decorators.py (@guarded), and plugins/python/vanzik/adapters/*.py (framework integrations). The package __init__.py re-exports the public surface — always import from the package root.

---

Table of contents#

• Install and import
• Environment variables
• Guard constructor in detail
• check() — argument guide
• Fail-mode semantics
• Timeouts
• approve, deny, wait_for_approval
• Exception hierarchy
• @guarded decorator
• Adapters
• Testing against a fake proxy

---

Install and import#

pip install vanzik            # core only
pip install vanzik[langchain]
pip install vanzik[crewai]
pip install vanzik[browser-use]
pip install vanzik[mcp]
pip install 'vanzik[openai-agents]'
pip install vanzik[all]       # everything
pip install vanzik[dev]       # pytest, coverage, etc.

from vanzik import (
    Guard, CheckResult, guarded,
    VanzikError, VanzikDenied,
    VanzikApprovalRequired, VanzikApprovalTimeout,
    VanzikAuthError,
)

vanzik.__version__ is the SDK's version string, e.g. "1.2.0" — the same as the vanzik release it ships in.

The core SDK has no runtime dependencies — it uses only the standard library (urllib, json, dataclasses, functools, time). Adapters pull in their respective frameworks through PyPI extras.

---

Environment variables#

Explicit arguments always win over env vars; env vars always win over package defaults. Variables for the server and CLI: CLI.md § Environment variables.

---

Guard constructor in detail#

Guard(
    base_url: str = "",
    agent_id: str = "",
    timeout: int = 5,
    api_key: str = "",
    fail_mode: str = "deny",   # or "allow"
    tenant_id: Optional[str] = None,
    parent_session_id: Optional[str] = None,   # v1.4
)

• base_url — trailing / is stripped. Empty string falls back to VANZIK_URL, then to "http://localhost:8080".
• agent_id — sent with every check() as the agent_id JSON field. Policy agents.<id>.override blocks key off this value.
• timeout — per-HTTP-call timeout in seconds. Applies to /v1/check, /v1/approve, /v1/deny, and each poll inside wait_for_approval.
• api_key — empty string falls back to VANZIK_API_KEY. Required whenever the server is started with --api-key.
• fail_mode — see Fail-mode semantics. Invalid values raise ValueError at construction so misconfiguration fails at startup rather than mid-request.
• tenant_id — None falls back to VANZIK_TENANT_ID; an explicit "" ignores the env var. Empty or "local" uses the legacy /v1/… URLs; any other value uses /v1/t/<tenant>/…, with the tenant URL-encoded.
• parent_session_id — sent as parent_session_id on every check() unless the call passes its own. See Linking sub-agent sessions.

Linking sub-agent sessions#

When an orchestrator starts sub-agents, give each sub-agent its own session_id and name the orchestrator's session as its parent. Policy ignores the parent; the audit log records it, and GET /v1/sessions and the dashboard's Sessions view show the sessions linked.

orchestrator = Guard(agent_id="planner")
orchestrator.check("shell", command="ls", session_id="run-42")

researcher = Guard(agent_id="researcher", parent_session_id="run-42")
researcher.check("network", domain="api.github.com", session_id="run-42/researcher")

@guarded accepts it. The framework adapters use the Guard's parent_session_id and never read one from tool input: the model writes the tool arguments, and it must not choose which session its calls are filed under.

Guard instances are thread-safe for concurrent check() calls (all state read-only after init). They are cheap; creating one per request is fine, and reusing one is slightly cheaper.

---

check() — argument guide#

guard.check(
    scope: str,
    *,
    action: str = "",
    command: str = "",
    path: str = "",
    domain: str = "",
    url: str = "",
    session_id: str = "",
    est_cost: float = 0.0,
    meta: Optional[dict] = None,
    approval_id: str = "",
    parent_session_id: str = "",   # v1.4
) -> CheckResult

parent_session_id names the session that started session_id (see Linking sub-agent sessions). It defaults to the Guard's parent_session_id and is omitted from the body when empty.

approval_id replays a previously issued approval after wait_for_approval resolved ALLOW — the server consumes the one-shot capability, enforces --approval-validity, reserves cost, and audits the execution as allow:approved (see APPROVAL_WORKFLOW.md). Omitted from the body when empty.

All fields past scope are keyword-only. Fill the ones your scope cares about:

Unset fields are dropped from the request body to keep audit entries clean. est_cost=0.0 is also dropped (use a non-zero value when you want the cost engine to consider the call).

The request body is always JSON with Content-Type: application/json and capped at 1 MiB by the server (configurable via the policy proxy.request.max_body_bytes key).

The returned CheckResult#

@dataclass
class CheckResult:
    decision: str           # "ALLOW", "DENY", "REQUIRE_APPROVAL"
    reason: str
    matched_rule: str = ""
    approval_id: str = ""
    approval_url: str = ""

Convenience booleans: .allowed, .denied, .needs_approval.

---

Fail-mode semantics#

These failures of check() collapse into a synthetic CheckResult controlled by fail_mode:

• urllib.error.URLError — connection-phase failure (connect refused, DNS, SSL handshake), and HTTP error statuses (urllib raises HTTPError, a URLError subclass, for 4xx/5xx).
• OSError — post-connect failure (ConnectionResetError, BrokenPipeError, socket.timeout surfaced from resp.read()).
• json.JSONDecodeError — the body parsed as non-JSON; from the caller's perspective this is indistinguishable from an unreachable proxy.
• A response that isn't a valid decision: a Content-Type other than application/json, or a JSON body without a decision field (reason Vanzik returned unexpected content-type … / Vanzik returned malformed response body).

# fail_mode="deny" (default)
CheckResult(decision="DENY",  reason="Vanzik unreachable (deny): <original error>")

# fail_mode="allow"
CheckResult(decision="ALLOW", reason="Vanzik unreachable (allow): <original error>")

When to use "allow": rarely. Acceptable only if (a) your agent is already in a trusted environment, (b) Vanzik is advisory rather than authoritative, and (c) you log and alert on the "Vanzik unreachable" reason so the outage is visible.

When to stick with "deny": any production path where an unsupervised agent must not act. That is the safe default.

---

Timeouts#

wait_for_approval quietly swallows individual poll failures (URLError, HTTP errors other than 401/403, other OSErrors, and a body that is not JSON) and keeps retrying until the deadline — the assumption is that the server is momentarily unreachable but will come back within the approval window. A final deadline miss returns CheckResult(decision="DENY", reason="Approval timed out"), which the @guarded(wait_for_approval=True) wrapper surfaces as VanzikApprovalTimeout.

Pick timeout higher than your human-SLA. If approvers need 15 minutes on average, timeout=300 will fire false negatives.

Restarts pause approvals; they don't kill them. The server persists the approval queue by default (--persist), so pending IDs survive a restart and your poll loop picks up where it left off. The exceptions: a server running --persist=false loses every pending ID on restart, and an entry created in the final ≥1 s store-sync window before a hard crash may be gone. In either case the poll loop retries against a queue that no longer has the entry — catch the timeout and re-issue the check call (which creates a new approval ID).

---

approve, deny, wait_for_approval#

guard.approve("ap_1a2b3c…")               # True / False
guard.deny("ap_1a2b3c…")                  # True / False
guard.wait_for_approval("ap_1a2b3c…",
                        timeout=300,
                        poll_interval=2)  # CheckResult

All three send Authorization: Bearer <api_key> when api_key is set. If the server was started without --api-key, the key is ignored. If the server was started with --api-key and you do not set one on the SDK side, the server answers 401: approve/deny return False, and wait_for_approval raises VanzikAuthError (with .status) on the first 401/403 instead of polling until the deadline.

approve/deny treat any URLError as failure and return False. They do not distinguish network error from 4xx. If you need that distinction, call the HTTP API directly.

---

Exception hierarchy#

The first three subclasses are raised by the @guarded decorator; VanzikAuthError is raised by wait_for_approval (and so also through @guarded(wait_for_approval=True)) when a status poll gets 401/403. All extend PermissionError so legacy except PermissionError: keeps working.

PermissionError
└── VanzikError         .result: Optional[CheckResult]
    ├── VanzikDenied
    ├── VanzikApprovalRequired   .approval_id, .approval_url
    ├── VanzikApprovalTimeout    .approval_id
    └── VanzikAuthError          .status

The .result attribute lets you skip re-parsing the reason string:

try:
    do_thing()
except VanzikDenied as e:
    log.warning("blocked by rule %s: %s",
                e.result.matched_rule, e.result.reason)
except VanzikApprovalRequired as e:
    notify_slack(f"Action {e.approval_id} waiting at {e.approval_url}")

---

@guarded decorator#

@guarded(
    scope: str,
    guard: Optional[Guard] = None,
    *,
    wait_for_approval: bool = False,
    approval_timeout: int = 300,
    approval_poll_interval: int = 2,
    **check_kwargs,
)

Behavior:

1. Creates a default Guard() if guard=None (picks up env vars).
2. Extracts a command string: the first positional arg, else the command= or cmd= kwarg, else "".
3. Calls guard.check(scope, command=..., **check_kwargs).
4. Branches on the decision:
   • ALLOW → runs the wrapped function.
   • REQUIRE_APPROVAL + wait_for_approval=False → raises VanzikApprovalRequired immediately.
   • REQUIRE_APPROVAL + wait_for_approval=True → blocks on wait_for_approval. A resolved ALLOW is replayed through guard.check(..., approval_id=...) (consuming the one-shot approval and reserving cost) and the function runs only if that replay allows; a resolved DENY or a denied replay raises VanzikDenied; a refused replay that re-entered the approval flow raises VanzikApprovalRequired with the new id; timeout raises VanzikApprovalTimeout.
   • DENY → raises VanzikDenied.

Any **check_kwargs you pass are forwarded verbatim to Guard.check(). Useful for pinning session_id=, meta=, etc.

---

Adapters#

All adapters import lazily — the extras you do not install are never imported and never crash the core SDK.

Per-framework usage, scope-inference rules (keyword tables, runtime upgrades, default_scope fallback), version pins, and deny/approval return semantics are documented once in ADAPTERS.md.

---

Testing against a fake proxy#

Option A — run the real binary#

The repo's tests/test_end_to_end_real_server.py does exactly this: spawn ./vanzik serve --policy <test-policy.yaml> on a random port, talk to it with Guard, and assert decisions. This is the most realistic path and what CI uses.

Option B — a stdlib http.server stub#

For unit tests, stand up a local HTTP server that answers /v1/check with a canned decision. The SDK sends its requests with urllib, so mocks that patch the requests library (such as responses) never see the call: the check fails closed and the test only looks like it passed. A real socket on 127.0.0.1 works and needs nothing beyond the standard library:

import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

import pytest
from vanzik import Guard


class StubVanzik(BaseHTTPRequestHandler):
    """Answers every POST /v1/check with one canned decision."""

    def do_POST(self):
        if self.path != "/v1/check":
            self.send_error(404)
            return
        length = int(self.headers.get("Content-Length", 0))
        self.server.requests.append(json.loads(self.rfile.read(length)))
        body = json.dumps({
            "decision": "DENY",
            "reason": "destructive command",
            "matched_rule": "deny:shell:rm-rf",
        }).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):  # keep pytest output quiet
        pass


@pytest.fixture
def stub_vanzik():
    server = HTTPServer(("127.0.0.1", 0), StubVanzik)  # port 0: any free port
    server.requests = []
    threading.Thread(target=server.serve_forever, daemon=True).start()
    yield server
    server.shutdown()
    server.server_close()


def test_denies_rm_rf(stub_vanzik):
    guard = Guard(f"http://127.0.0.1:{stub_vanzik.server_port}")
    r = guard.check("shell", command="rm -rf /")
    assert r.denied
    assert "destructive" in r.reason
    assert stub_vanzik.requests[0]["command"] == "rm -rf /"

Option C — fail_mode="allow" for disconnected tests#

If you intentionally do not run a proxy in your unit tests and you want check() to succeed, construct Guard(..., fail_mode="allow"). Only do this in tests; it is the wrong default anywhere else.

---

Related docs#

• docs/DEPLOYMENT.md — getting the server set up so the SDK can reach it.
• docs/POLICY_REFERENCE.md — what you pass to check().
• docs/TROUBLESHOOTING.md — "Python SDK always returns DENY", "approvals disappeared after restart", etc.
• docs/ADAPTERS.md — per-framework deep dive (written separately so this file does not balloon).