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

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:

r = vx.inspect_tool_call("send_email", {"to": addr}, policy=policy)
if r.needs_approval:
    decision = vx.wait_for_approval(r.approval_id, timeout=600)   # blocks until decided
    if decision["status"] == "approved":
        send_email(addr)                    # a human said yes
    # "rejected" / "expired" → don't run

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

• 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.

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, 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.

License

MIT — see LICENSE.