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="vnz_live_..."

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

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

Mark a tool as approval-required and a call to it returns needs_review — the agent pauses until a person says yes. An unsafe call (SSRF, not on the allowlist) is still blocked outright, never merely sent for review.

from venzx import Venzx, Policy, ApprovalRequired

vx = Venzx()
guard = vx.guard_for(
    policy=Policy().allow_tools("search").require_approval("send_email"),
    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)   # route to Slack / a dashboard / an email link

# Or approve inline with your own approver (True = allow, False = deny):
guard = vx.guard_for(
    policy=Policy().require_approval("send_email"),
    on_approval=lambda res: ask_human_in_slack(res),
)

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.

→ venzx on PyPI · Product features · Live demo