venzx 0.5.0

Python SDK for VENZX — runtime security for AI agents (prevents leaks, keeps proof, alerts you).

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

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": "[e-mail]"})   # 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
raw	the untouched JSON, for forward compatibility

Policies — the customization surface

A Policy is the four action controls. 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")
    .limit(max_tool_calls=10, max_tokens=20_000, max_cost=0.50)
)

vx.inspect_tool_call("send_email", {"to": "[e-mail]"}, 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)
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.

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. 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": "[e-mail]"})
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.

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.