Metadata-Version: 2.4
Name: venzx
Version: 0.5.1
Summary: Python SDK for VENZX — control what your AI agent can do: allow, block, cap the spend, or pause for a human on every action, with a tamper-evident audit log.
Project-URL: Homepage, https://venzx.com
Project-URL: Documentation, https://venzx.com/docs
Project-URL: Live demo, https://venzx.com/try
Author: VENZX
License: MIT License
        
        Copyright (c) 2026 VENZX
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
License-File: LICENSE
Keywords: action-control,agent-security,ai,ai-agent,ai-agent-security,audit-log,guardrails,human-in-the-loop,langchain,llm,spend-limits,tool-call-approval,venzx
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.8
Requires-Dist: requests>=2.25.0
Provides-Extra: async
Requires-Dist: httpx>=0.23; extra == 'async'
Provides-Extra: dev
Requires-Dist: httpx>=0.23; extra == 'dev'
Requires-Dist: mypy>=1.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21; extra == 'dev'
Requires-Dist: pytest>=7; extra == 'dev'
Requires-Dist: types-requests; extra == 'dev'
Description-Content-Type: text/markdown

# VENZX Python SDK

Official Python client for **[VENZX](https://venzx.com)** — 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

```bash
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

```bash
export VENZX_API_KEY="sk-..."
```

## Quick start

```python
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).

```python
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                              |
| `raw`                     | the untouched JSON, for forward compatibility    |

## 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:

```python
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):

```python
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:

```python
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.

```python
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:

```python
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:

```python
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:

```python
@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):

```python
@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:

```python
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:

```python
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

```python
results = vx.inspect_many([
    {"stage": "input",  "text": prompt},
    {"stage": "output", "text": reply},
], stop_on_block=True)
```

## Streaming

```python
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

```python
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. |

```python
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.

## Async

```python
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`:

```python
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](./LICENSE).
