VENZX Python SDK

Python 3.8+. The sync client only needs requests.

pip install venzx            # sync client
pip install "venzx[async]"   # also installs httpx for the async client

Set your API key once in the environment (create keys in your dashboard):

export VENZX_API_KEY="pai-..."

One complete script. After pip install venzx and setting your key above, copy this whole block into a file and run it — you'll watch VENZX allow a safe action, block an unsafe one, and pause a risky one for a human, all in a few lines.

# save as try_venzx.py, then:  python try_venzx.py
from venzx import Venzx, Policy

vx = Venzx()  # reads VENZX_API_KEY from the environment

# Your rulebook: search runs freely, sending email needs a human's OK,
# and nothing may reach an address outside your own API.
policy = (
    Policy()
    .allow_tools("search", "send_email")
    .require_approval("send_email")
    .allow_domains("api.yourapp.com")
)

def check(tool, args):
    r = vx.inspect_tool_call(tool, args, policy=policy)
    print(f"{tool:12} -> {r.decision:12} ({r.decided_by})  {r.reason or ''}")
    return r

# 1) A safe, allow-listed action — runs.
check("search", {"q": "refund policy"})
# -> search       -> allow        (allow)

# 2) A tool that's NOT on the allowlist — blocked outright.
check("run_shell", {"cmd": "rm -rf /data"})
# -> run_shell    -> block        (tool_allowlist)  run_shell not allowed

# 3) A risky action — held for a human instead of running.
r = check("send_email", {"to": "customer@acme.com"})
# -> send_email   -> needs_review (approval_required)
if r.needs_approval:
    print("Held. The approver was emailed a one-click Approve / Reject link.")

That's the whole product in one screen: allow, block, pause for a human — and every one of those decisions is written to a tamper-evident log you can export. Everything below is detail on each piece.

The core check is the tool call — every action your agent tries to take. VENZX inspects it and returns a verdict before it runs.

from venzx import Venzx, Policy

vx = Venzx()  # reads VENZX_API_KEY

r = vx.inspect_tool_call(
    "send_email",
    {"to": "customer@acme.com"},
    policy=Policy().allow_tools("search").require_approval("send_email"),
)

r.decision        # "allow" | "block" | "needs_review"
r.blocked         # True / False
r.needs_approval  # True when a human must approve first
r.reason          # plain-language explanation of the verdict
r.decided_by      # which check decided — e.g. "ssrf", "secret_scan", "allow"
r.duration_ms     # guard pipeline time, server-measured

A Policy sets the rules for a call (or every call). It's fluent, typed, and validated client-side. Only fields you set are sent.

from venzx import Policy

policy = (
    Policy()
    .allow_tools("search", "read_file")          # may run freely
    .require_approval("send_email", "delete")    # pause for a human
    .allow_domains("api.yourapp.com")            # outbound allowlist
    .limit(max_tool_calls=20, max_tokens=20_000, max_cost=0.50)  # per-run caps
)

guard.tool_call("send_email", {"to": addr}, policy=policy)

The raw client returns a verdict; the Guard acts on it for you — so you get automatic enforcement with no per-call if blocked: branching.

from venzx import Venzx, Policy

vx = Venzx()
guard = vx.guard_for(
    policy=Policy().allow_tools("search").require_approval("send_email"),
    on_tool_block="raise",       # blocked action → raise venzx.Blocked
    on_approval="raise",         # risky action  → raise ApprovalRequired
    fail_open=True,              # if VENZX is down, don't break your app
)

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

client = guard.wrap_openai(OpenAI())   # drop-in: prompt + reply guarded

Guard any tool in one line — works with LangChain, CrewAI, LlamaIndex, MCP handlers, or plain functions. Just decorate the function your agent calls:

@guard.protect_tool          # every call is checked before it runs
def send_email(to: str, body: str):
    ...   # blocked -> raises Blocked; risky -> waits for approval

Compliance note (v0.7.0+): even with fail_open=True, if VENZX can't write the tamper-evident audit record for a call, the Guard fails closed and blocks it — an action that runs without a log would break the guarantee. Ordinary outages (network, timeout) still honour fail_open.

Mark a tool as approval-required and a call to it returns needs_review — the action is held, and the approver is emailed a one-click Approve / Reject link (no login). The agent resumes only when they say yes. An unsafe call (SSRF, not on the allowlist) is still blocked outright, never merely sent for review.

from venzx import Venzx, Policy

vx = Venzx()
policy = Policy().allow_tools("search").require_approval("send_email")

r = vx.inspect_tool_call("send_email", {"to": addr}, policy=policy)
if r.needs_approval:
    # Held. The approver already got the email; block until they decide.
    decision = vx.wait_for_approval(r.approval_id, timeout=600)
    if decision["status"] == "approved":
        send_email(addr)          # a human said yes
    # "rejected" / "expired" → do not run

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

Two different timers. wait_for_approval(timeout=…) is how long your agent blocks; the approval's own expiry (default 2h) is when VENZX auto-rejects it. They're independent — if your wait times out the request is still live, so poll again (or rely on the webhook) rather than assuming it was declined.

Auto-reject on timeout. A held request auto-rejects after a deadline (default 2h), set per policy with approval_expiry_seconds. Prefer a webhook? Configure one and VENZX POSTs it on every decision (polling still reflects the result if delivery fails). Every step — created, emailed, approved / rejected / timed-out — is written to the tamper-evident log, exportable as CSV or a printable PDF for an auditor, and viewable on a read-only status page.

A run pins a run_id + policy across many calls, so per-run budgets (tool calls, tokens, cost) are enforced across the whole agent run — even across servers.

run = vx.run(policy=Policy.strict().limit(max_tool_calls=5))
run.inspect_input(user_msg)
run.inspect_tool_call("search", {"q": "..."})   # budget shared across the run

• run(...) — a run session that shares per-run spend caps across many calls.
• inspect_many([...]) — batch inspection with stop_on_block.
• stream(...) — streaming inspection (Server-Sent Events) for large bodies.
• AsyncVenzx — the same surface, awaited (pip install "venzx[async]").
• Auto-retry on 429/5xx with backoff; hooks (on_request/on_response/on_block).

Every failure raises a typed VenzxError subclass — AuthenticationError, RateLimitError, InsufficientCreditsError, Blocked, ApprovalRequired, and more — so you can handle each precisely.

This page covers the essentials. The exhaustive reference — every method, parameter, and the typed result objects — ships with the package and on PyPI.

HTTP endpoints

→ venzx on PyPI · Product features · Live demo