VENZX Python SDK

Official Python client for VENZX — an action-control guard for AI agents. Before your agent runs a tool, VENZX checks it and decides:

• Control — only tools on the allowlist run; calls reaching an internal / cloud-metadata address or a non-allowlisted domain are blocked (SSRF).
• Stop leaks — block a tool call whose arguments carry a known credential format (AWS, GitHub, OpenAI, Anthropic, Google, Stripe, Slack keys, JWTs, private keys). An opt-in high-entropy layer adds coverage for unknown formats.
• Approve — high-risk tools pause for a human (needs_review) instead of running automatically.
• Cap — per-run limits on tool calls, tokens, and dollars so a loop can't burn your budget.
• Prove — every decision lands in a tamper-evident audit log.

This SDK wraps the public HTTP API and exposes the full policy surface — tool allowlists, human-approval gates, domain allowlists, per-run spend caps — plus run sessions, retries, hooks, a Guard that auto-handles verdicts, and an async client.

Just want the 3 lines to get going? → Install → set your key → Quick start. Everything after that is reference — reach for it when you need it.

Contents

• Install & authenticate
• Quick start — check an action in 6 lines
• Policies — your rulebook (allow / approve / cap)
• Guard — the recommended way; auto-handles every verdict
• Human-approval gate — pause risky actions for a person
• Run sessions & spend caps
• If VENZX is unreachable — fail-open vs fail-closed
• Async · Errors · Health & monitoring

---

Install

pip install venzx           # sync client (depends only on requests)
pip install "venzx[async]"  # also installs httpx for the async client

Requires Python 3.8+.

Authenticate

export VENZX_API_KEY="pai-..."

Quick start

The simplest check: ask VENZX about an action, get a verdict back.

from venzx import Venzx, Policy

vx = Venzx()  # reads VENZX_API_KEY
policy = Policy().allow_tools("search", "read_file")

verdict = vx.inspect_tool_call("run_shell", {"cmd": "rm -rf /data"}, policy=policy)

if verdict.blocked:
    print("VENZX blocked it:", verdict.reason)   # run_shell not in allowlist

Most apps don't write if verdict.blocked: by hand. They use the Guard (recommended), which acts on the verdict for you — one decorator or one wrapped client and every call is enforced automatically. The raw check above is shown first so you can see exactly what VENZX returns.

The three inspect stages

tool_call is the one that does the work — allowlist, SSRF/domain block, approval gate, and spend caps. input / output are accepted for API compatibility but are pass-through (no content checks).

vx.inspect_tool_call("send_email", {"to": "customers@evil.com"})   # the guard
vx.inspect_input(user_prompt)                                       # pass-through
vx.inspect_output(model_response_text)                              # pass-through

All three return an InspectResult:

Attribute	Meaning
decision	"allow", "block" or "needs_review"
blocked / allowed	convenience booleans
needs_approval	true when a tool must be approved by a human
degraded	true when this is a fail-open verdict (see below)
findings	list of Finding objects (why it was decided)
reason	short human reason for a block / needs_review
run_id / request_id	correlate a run / a log entry
processing_time_seconds	server-side latency
decided_by	which check produced the verdict (see below)
duration_ms	guard pipeline time in ms (server-measured)
raw	the untouched JSON, for forward compatibility

decided_by — why a call got its verdict

Every verdict names the exact check that produced it, so a log line tells you why at a glance (blocked_by: ssrf):

r = vx.inspect_tool_call("fetch_url", {"url": "http://169.254.169.254/"})
print(r.decision, r.decided_by, f"{r.duration_ms}ms")   # block ssrf 2.1ms

Values: ssrf, tool_allowlist, domain_allowlist, secret_scan, rate_limit, token_limit, cost_limit, approval_required, auto_allow_rule, guard_error (a check failed closed), allow, stage_passthrough.

Policies — the customization surface

A Policy is the deterministic controls — the four action controls plus the tool-argument secret-leak guard. Set it per call, or via default_policy for every call. It maps 1:1 onto what the API accepts, and only the fields you set are sent. Build one fluently:

from venzx import Policy

policy = (
    Policy()
    .allow_tools("search", "read_file", "send_email")
    .require_approval("send_email", "delete")   # pause these for a human
    .allow_domains("api.yourapp.com")
    .block_secret_exfiltration(True)            # scan tool args for live secrets
    .limit(max_tool_calls=10, max_tokens=20_000, max_cost=0.50)
)

vx.inspect_tool_call("send_email", {"to": "x@acme.com"}, policy=policy)

The whole policy surface:

Builder / field	What it does
allow_tools(*names)	tool allowlist — any tool not listed is blocked
require_approval(*names)	tools that must be approved by a human (returns needs_review)
allow_domains(*domains)	outbound destination allowlist (blocks SSRF / non-listed hosts)
block_secret_exfiltration(enabled=True, *, high_entropy=False)	scan tool-call arguments for a live credential (API key, token, private key) and block it. Known formats are on by default; pass enabled=False to allow a tool that legitimately receives a secret, or high_entropy=True to also catch unknown random-looking credentials
limit(max_tool_calls=, max_tokens=, max_cost=, fail_open=)	per-run spend caps. fail_open=True lets a call through if the spend backend is unreachable (default fails closed / blocks). Cost control only — security rules always fail closed

What happens if our counter is unavailable. The per-run spend caps are counted in Redis. If that backend is unreachable, VENZX blocks the call rather than run it uncounted — including calls that are otherwise perfectly safe. That is the default because a per-process fallback would silently undercount across instances and let a run blow past its cap.

It also means a spend-counter outage stops your agent, not just your caps. If you would rather keep running — reasonable, since a spend cap is cost control and not a security boundary — set fail_open=True on limit(...). The security rules (tool allowlist, SSRF and domain blocking, the credential scan) always fail closed and are not affected by this setting.

Set a client-wide default that every call inherits (per-call policies merge on top, and win on conflicts):

vx = Venzx(default_policy=Policy().allow_tools("search"))
vx.inspect_tool_call("search", {"q": "x"})                       # uses the default
vx.inspect_tool_call("search", {"q": "x"},
                     policy=Policy().limit(max_tool_calls=3))      # default + tighter cap

guard() — stop the agent on a block

guard* is like inspect* but raises Blocked instead of returning a blocked verdict — handy for short-circuiting an agent step:

from venzx import Blocked

try:
    vx.guard_tool_call("send_email", {"to": user_supplied_address})
    send_the_email()
except Blocked as e:
    log.warning("VENZX refused the tool call: %s", e.result.reason)

Guard — decide and auto-handle (recommended)

The raw client gives you a verdict; the Guard acts on it for you, so a block becomes automatic handling instead of per-call if blocked: branching. You set the action once and every check is enforced.

from venzx import Venzx, Policy

vx = Venzx()
guard = vx.guard_for(
    policy=Policy().allow_tools("search").require_approval("send_email"),
    on_block="safe_message",      # a blocked answer  → a safe message
    on_approval="raise",          # a needs_review    → pause for a human
    safe_message="Sorry, I can't do that.",
    fail_open=True,               # if VENZX is down, don't break the app
    on_event=send_to_slack,       # route incidents to a webhook/Slack/your DB
)

safe_reply = guard.output(model_reply)      # text safe to send (or a safe msg)
guard.tool_call("send_email", {"to": addr}) # raises Blocked if not allowed

Human-in-the-loop approval — high-risk tools pause for a person instead of running automatically. List them with require_approval(...); a tool_call to one returns needs_review, and the Guard applies your on_approval action:

from venzx import Venzx, Policy, ApprovalRequired

guard = vx.guard_for(
    policy=Policy()
        .allow_tools("search", "read_file")        # run freely
        .require_approval("send_email", "delete"),  # pause for a human
    on_approval="raise",   # default: raise ApprovalRequired so the agent halts
)

try:
    guard.tool_call("send_email", {"to": addr})     # high-risk → paused
    send_email(addr)                                 # only runs if approved
except ApprovalRequired as e:
    queue_for_review(e.result)                       # Slack / dashboard / email link

Or approve inline with your own approver (a Slack prompt, a CLI y/n, a dashboard) — return True to let it run, False to deny:

guard = vx.guard_for(
    policy=Policy().require_approval("send_email"),
    on_approval=lambda res: ask_human_in_slack(res),  # True = approve, False = deny
)
guard.tool_call("send_email", {"to": addr})   # blocks on deny, returns on approve

A hard block (SSRF, not-in-allowlist) is never downgraded to an approval prompt — unsafe calls are still refused outright.

Managed approval (built-in) — you don't have to build the review UI. When a tool call returns needs_review, VENZX holds it and emails the approver a one-click Approve / Reject link (no login). The inspect result carries the held request; wait for the decision, then act:

from venzx import ApprovalSpent

r = vx.inspect_tool_call("send_email", {"to": addr}, policy=policy)
if r.needs_approval:
    vx.wait_for_approval(r.approval_id, timeout=600)   # blocks until decided
    try:
        vx.consume_approval(r.approval_id)   # claim it — succeeds exactly once
    except ApprovalSpent:
        ...                                  # already used, rejected, or stale
    else:
        send_email(addr)                     # a human said yes, and you hold the claim

# Non-blocking? poll instead of waiting:
status = vx.get_approval(r.approval_id)["status"]
# pending | approved_unclaimed | used | rejected | expired

Reading a status is not permission (0.8.0+). get_approval / wait_for_approval only tell you where a request stands; a decided-but- unclaimed one reads approved_unclaimed. Permission comes from consume_approval, which succeeds for exactly one caller and is bound to the API key that made the request. Every later attempt raises :class:ApprovalSpent — deliberately not a Blocked, because "ask for a new approval" and "you aren't allowed" need different handling.

Upgrading from 0.7.x? The old shape was if wait_for_approval(...)["status"] == "approved": run(). That comparison is now always false, so the action never runs — safe, but broken. It has to become the consume_approval call above. Using the Guard instead? Nothing to change: it claims the authorization for you on every approval path.

• Two timers, don't confuse them — wait_for_approval(timeout=…) is how long your agent waits; the approval's own expiry (default 2h) is when it auto-rejects. They're independent: if your wait times out, the request is still live — poll again or rely on the webhook, don't assume it was declined.
• Timeout — a held request auto-rejects after its expiry (default 2h), set per policy with approval_expiry_seconds.
• Webhook — configure one and VENZX POSTs approval.decided on every decision (retried; polling still reflects it if delivery fails).
• Proof — every step is written to the tamper-evident log, exportable as CSV or PDF, and viewable on a read-only status page.
• "Don't ask again" — an approver can auto-allow that tool below an amount.

The async client mirrors both: await avx.get_approval(id) / await avx.wait_for_approval(id).

One-line integration — wrap a whole function:

@guard.protect                     # checks input + output automatically
def answer(prompt: str) -> str:
    return my_llm(prompt)

Guard any tool in one line — works with LangChain, CrewAI, LlamaIndex, MCP handlers, or plain functions. Decorate the function your agent calls as a tool and every call is checked before it runs (blocked → raises Blocked; risky → waits for approval; allowed → runs unchanged):

@guard.protect_tool                 # uses the function name as the tool name
def send_email(to: str, body: str):
    ...

@guard.protect_tool(name="delete")  # or name it explicitly
def remove(path: str):
    ...

Drop-in for an OpenAI-style client — the prompt and the reply are checked (and the reply rewritten) with no other code changes:

client = guard.wrap_openai(OpenAI())
client.chat.completions.create(messages=[...])   # now guarded

Reliability: the Guard never throws unexpectedly (only Blocked, and only when you choose "raise"). If VENZX itself errors, fail_open=True lets traffic through so an outage can't take your app down; fail_open=False fails closed for security-critical paths. The one exception: if the service can't write the tamper-evident audit row (AUDIT_UNAVAILABLE), the Guard fails closed even with fail_open=True — an unlogged action would break the guarantee. Every handled decision (and any guard error) is reported to on_event so it can be routed automatically.

Credits: each guard.input / guard.output / guard.tool_call is one /v1/inspect call = one credit, same as the raw client. @protect checks input and output by default (2 credits/call); use @guard.protect(check=("output",)) for one. The Guard adds no new pricing — it's client-side convenience over the same metered endpoint.

Run sessions

Per-run budgets (tool calls, tokens, cost) are enforced across calls that share a run_id. A Run pins the id and a shared policy so you don't repeat them:

run = vx.run(policy=Policy().allow_tools("search").limit(max_tool_calls=5))

run.inspect_input(user_prompt)
run.inspect_tool_call("search", {"q": "..."})
run.guard_output(model_reply)
print("run id:", run.run_id)   # server-allocated on the first call

Batch

results = vx.inspect_many([
    {"stage": "input",  "text": prompt},
    {"stage": "output", "text": reply},
], stop_on_block=True)

Streaming

from venzx import Stage

for event in vx.stream(Stage.OUTPUT, text=long_text):
    if event.type == "progress":
        print(f"{event.pct}% — {event.step}")
    elif event.type == "result":
        print("decision:", event.result.decision)

Hooks & client config

vx = Venzx(
    timeout=20.0,
    max_retries=3,                       # retries 429/502/503/504 + connection errors
    backoff_cap=8.0,
    default_headers={"X-Env": "prod"},
    on_unreachable="fail_closed",        # what to do if VENZX is down (see below)
    on_block=lambda r: alerts.page(r.reason),
    on_response=lambda r: metrics.observe(r.processing_time_seconds),
)

What happens if VENZX is unreachable?

Your call. After exhausting retries on a connection error / timeout, the SDK does one of two things, controlled by on_unreachable (or the VENZX_ON_UNREACHABLE env var):

Mode	Behaviour	Use when
"fail_closed" (default)	Raises the connection error, so the guarded action does not run.	Enforcement matters more than uptime — a blocked agent is safer than an unchecked one.
"fail_open"	Returns a synthetic allow verdict with result.degraded == True, so the agent keeps running.	Availability matters more — you'd rather the agent proceed unguarded than stop.

vx = Venzx(on_unreachable="fail_open")
r = vx.inspect_tool_call("send_email", {"to": "x@acme.com"})
if r.degraded:
    log.warning("VENZX was unreachable — action ran without a guard check")

A real verdict from the service always has degraded == False, so you can always tell an enforced decision apart from a fail-open one.

fail_open covers being unreachable, and nothing else. A connection error or timeout is rescued; a reply from the API is not. In particular, running out of credits still stops your agent — it raises InsufficientCreditsError whatever on_unreachable is set to.

That is deliberate: an empty balance is a known state you were warned about by email, not a surprise outage, and quietly running unguarded for a week is worse than failing loudly. Watch for the low-balance alert (Settings → Notifications, at a threshold you choose), or catch the exception yourself if you would rather degrade than stop:

from venzx import InsufficientCreditsError

try:
    verdict = vx.inspect_tool_call(name, args, policy=policy)
except InsufficientCreditsError:
    page_someone("VENZX credits exhausted")
    verdict = None   # your call what to do next

Note this is the client-side availability switch. Separately, the VENZX service itself fails closed on any internal error: if a security check crashes, its DNS lookup times out, or the rate/spend backend (Redis) is unreachable, the server returns a block (never a silent allow), and decided_by is guard_error.

One case always fails closed — even with fail_open=True (since 0.7.0). If the service could not write the tamper-evident audit row for your call it returns AUDIT_UNAVAILABLE, and the Guard blocks regardless of your fail_open setting. An action that runs without a log would break the core guarantee, so it is never allowed through. Ordinary availability failures (network, timeout, other 5xx) still honour fail_open.

Health & monitoring

h = vx.get_health()                       # unauthenticated; safe for a monitor
print(h["status"])                        # "ok" or "degraded"
print(h["checks"])                        # {mongo, redis, email: {status: up|down|...}}
print(h["stats_today"])                   # {total, allow, block, needs_review}

get_health() returns the report even when the service is degraded (HTTP 503) — the code is in h["http_status"] — and only raises when the service is truly unreachable. stats_today is the daily tally you can show a customer: "500 checks, 3 approvals, 1 block." The async client mirrors it: await avx.get_health().

Async

import asyncio
from venzx import AsyncVenzx, Policy

async def main():
    async with AsyncVenzx() as vx:                       # needs venzx[async]
        r = await vx.inspect_tool_call("search", {"q": "x"},
                                    policy=Policy().allow_tools("search"))
        results = await vx.inspect_many(batch, concurrency=8)
        async for event in vx.stream("output", text=long_text):
            ...

asyncio.run(main())

Error handling

Every error is a subclass of VenzxError:

from venzx import (
    Venzx, VenzxError, Blocked,
    AuthenticationError, PermissionDeniedError, RateLimitError,
    InvalidRequestError, InsufficientCreditsError, AuditUnavailableError,
)

try:
    vx.guard_output(text)
except Blocked as e:
    handle_block(e.result)
except InvalidRequestError as e:
    print("bad request:", e.validation_errors)
except RateLimitError as e:
    print("retry after", e.retry_after)
except InsufficientCreditsError:
    print("top up your credits")
except VenzxError as e:
    print("error:", e)

Transient failures (HTTP 429/502/503/504 and connection errors) are retried automatically with exponential backoff, honouring Retry-After. Authentication and permission errors are not retried.

API key scopes and rotation

An API key can be limited, from the API Keys page in your dashboard, to a subset of what the API offers:

Scope	Allows
inspect	POST /v1/inspect and /v1/inspect/stream
approvals:read	reading approval status (get_approval, wait_for_approval)
approvals:consume	spending an approval (consume_approval)

Calling an endpoint the key isn't permitted to use raises PermissionDeniedError with code == "INSUFFICIENT_SCOPE". A key created without an explicit scope list — and every key issued before scopes existed — has full access, so no existing integration changes.

Rotating a key from the dashboard issues a replacement immediately and gives the old key a grace period (24 h by default) rather than killing it, so you can roll a fleet onto the new secret without downtime. Once that window closes the old key raises AuthenticationError with code == "KEY_EXPIRED".

from venzx import AuthenticationError, PermissionDeniedError

try:
    vx.consume_approval(request_id)
except PermissionDeniedError as e:
    if e.code == "INSUFFICIENT_SCOPE":
        # This key can read approvals but not spend them.
        alert_ops(e.body.get("required_scope"))
except AuthenticationError as e:
    if e.code == "KEY_EXPIRED":
        # This key was rotated; deploy the replacement.
        alert_ops("VENZX key expired — switch to the rotated key")

License

MIT — see LICENSE.