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.

---

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

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

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=)	per-run spend caps

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.