Python SDK
Full API reference for the Veto Python SDK (v0.12.0).
Installation
pip install vetoWith LLM provider extras:
pip install veto[openai] # OpenAI support
pip install veto[anthropic] # Anthropic support
pip install veto[gemini] # Google Gemini support
pip install veto[all] # All providersprotect(tools, **kwargs)
Recommended entrypoint for new integrations.
from veto import protect
safe_tools = await protect(tools)Single-tool overload:
safe_tool = await protect(tool)Common options:
safe_tools = await protect(tools, pack="financial")
safe_tools_cloud = await protect(tools, api_key="veto_...")
safe_tools_log = await protect(tools, mode="log")
safe_tools_shadow = await protect(tools, mode="shadow")Use Veto.init() directly when you need full lifecycle or initialization control.
Veto.init(options?)
Initialize Veto. Uses the VETO_API_KEY environment variable by default.
from veto import Veto
veto = await Veto.init()Options
Pass a VetoOptions dataclass to customize initialization:
from veto import Veto, VetoOptions
veto = await Veto.init(VetoOptions(
api_key="veto_abc123...",
base_url="https://api.runveto.com",
mode="strict",
on_approval_required=my_approval_handler,
approval_poll_interval=2.0, # seconds
approval_timeout=300.0, # seconds
))| Option | Type | Default | Description |
|---|---|---|---|
api_key | str | VETO_API_KEY env | API key for Veto Cloud |
base_url | str | "https://api.runveto.com" | Cloud API base URL |
mode | "strict" | "log" | "shadow" | "strict" | Operating mode |
log_level | str | "info" | Log level |
session_id | str | — | Session ID for tracking |
agent_id | str | — | Agent ID for tracking |
validators | list | — | Additional validators |
timeout | int | 30000 | API timeout in ms |
retries | int | 2 | Retry count |
on_approval_required | callable | — | Hook fired when a tool call needs human approval |
approval_poll_interval | float | 2.0 | Seconds between approval polls |
approval_timeout | float | 300.0 | Max seconds to wait for approval |
Operating mode precedence: explicit mode option > config mode > VETO_MODE env > strict.
veto.wrap(tools)
Wraps an array of tools with validation. Auto-registers tool signatures with Veto Cloud for policy generation.
wrapped_tools = veto.wrap(my_tools)veto.wrap_tool(tool)
Wraps a single tool instance.
safe_tool = veto.wrap_tool(my_tool)await veto.guard(tool_name, args, *, session_id=None, agent_id=None)
Run Veto validation as a standalone check without wrapping or executing a tool.
from veto import Veto, GuardResult
veto = await Veto.init()
result: GuardResult = await veto.guard(
"wire_transfer",
{"amount": 25000, "recipient": "vendor-123"},
session_id="session-42",
agent_id="agent-7",
)
if result.decision == "deny":
print(result.reason)GuardResult (Python dataclass)
@dataclass
class GuardResult:
decision: Literal["allow", "deny", "require_approval"]
reason: str | None = None
rule_id: str | None = None
severity: Literal["critical", "high", "medium", "low", "info"] | None = None
approval_id: str | None = None
shadow: bool | None = None
shadow_decision: str | None = NoneGuard behavior (important)
- Uses the same internal validation engine path as wrapped tool calls.
- Records every guard check in history (
get_history_stats()andexport_decisions()include guard calls). - Never raises
ToolCallDeniedError; deny outcomes are returned inGuardResult. - In log mode and shadow mode,
guard()still returns the real policy verdict (deny/require_approval) instead of converting toallow. - In shadow mode,
guard()includesshadow=Trueandshadow_decisionfor non-allow outcomes. session_id/agent_idoverride instance-level tracking values for that call only.require_approvalis returned directly for cloud approval checks.approval_idis populated when available.rule_idandseverityare populated from validation metadata when present.
veto.get_history_stats()
Returns statistics about validation decisions.
stats = veto.get_history_stats()
# {"total_calls": 5, "allowed_calls": 4, "denied_calls": 1, ...}veto.clear_history()
Resets the history statistics.
veto.clear_history()Error handling
When a tool call is blocked in strict mode, Veto raises a ToolCallDeniedError:
from veto.core.interceptor import ToolCallDeniedError
try:
await wrapped_tool.ainvoke(args)
except ToolCallDeniedError as e:
print(e.tool_name) # "transfer_funds"
print(e.reason) # "Amount 5000 exceeds limit of 1000"
print(e.call_id) # "tc_abc123"See the Error Handling Guide for strategies on retry, graceful degradation, and framework-specific patterns (OpenAI, Anthropic, LangChain).
When an approval poll times out, Veto raises an ApprovalTimeoutError:
from veto.cloud.client import ApprovalTimeoutError
try:
await wrapped_tool.ainvoke(args)
except ApprovalTimeoutError as e:
print(e.approval_id) # "apr_abc123"
print(e.timeout) # 300.0Cloud validation and approvals
The Python SDK validates tool calls through the Veto Cloud API. The server can return three decisions: allow, deny, or require_approval.
When a tool call requires approval, the SDK:
- Fires the
on_approval_requiredcallback (so your app can show approval UI) - Polls
GET /v1/approvals/:iduntil a human approves or denies - Returns the final decision to the agent
async def handle_approval(context, approval_id):
print(f"Tool '{context.tool_name}' needs approval: {approval_id}")
veto = await Veto.init(VetoOptions(
on_approval_required=handle_approval,
))The callback receives the full ValidationContext and the approval_id string. It can be sync or async.
Approval preference cache
Cache per-tool preferences to auto-resolve approvals without server polling:
# Auto-approve all future calls to "read_file"
veto.set_approval_preference("read_file", "approve_all")
# Auto-deny all future calls to "delete_database"
veto.set_approval_preference("delete_database", "deny_all")
# Check current preference
veto.get_approval_preference("read_file") # "approve_all"
# Clear preference for one tool
veto.clear_approval_preferences("read_file")
# Clear all preferences
veto.clear_approval_preferences()Client-side deterministic validation
The Python SDK evaluates deterministic constraints locally when a cached policy is available, identical to the TypeScript SDK. No network round-trip needed for simple checks like number ranges, string enums, or regex patterns.
This is automatic in cloud mode. When the SDK has a cached policy for a tool and that policy uses deterministic mode without session constraints or rate limits, validation runs locally. Decisions are logged back to the server asynchronously.
Supported constraint types
| Constraint | Applies to | Description |
|---|---|---|
required | all | Argument must be present |
not_null | all | Argument cannot be None |
minimum | numbers | Lower bound (inclusive) |
maximum | numbers | Upper bound (inclusive) |
greater_than | numbers | Lower bound (exclusive) |
less_than | numbers | Upper bound (exclusive) |
min_length | strings | Minimum string length |
max_length | strings | Maximum string length |
enum | strings | Allowed exact values |
regex | strings | Pattern match (max 256 chars) |
min_items | arrays | Minimum list length |
max_items | arrays | Maximum list length |
Policy cache
Policies are cached with stale-while-revalidate:
| Window | Default | Behavior |
|---|---|---|
| Fresh | 60s | Returns cached policy immediately |
| Max age | 5min | Returns stale policy while refreshing in background |
Background refreshes use asyncio.create_task() to avoid blocking the current validation. The cache requires an active event loop.
VetoCloudClient
Standalone client for direct cloud API interaction:
from veto.cloud.client import VetoCloudClient, VetoCloudConfig
client = VetoCloudClient(
config=VetoCloudConfig(
api_key="veto_abc123...",
base_url="https://api.runveto.com",
timeout=30000,
retries=2,
),
)await client.validate(tool_name, arguments, context?)
result = await client.validate("send_email", {
"to": "user@example.com",
"subject": "Hello",
})
# ValidationResponse(decision="allow"|"deny"|"require_approval", reason=..., approval_id=...)await client.poll_approval(approval_id, options?)
from veto.cloud.types import ApprovalPollOptions
approval = await client.poll_approval("apr_abc123", ApprovalPollOptions(
poll_interval=2.0, # seconds
timeout=300.0, # seconds
))
# ApprovalData(id=..., status="approved"|"denied"|"expired", resolved_by=...)await client.fetch_policy(tool_name)
Fetch a tool's policy from the server. Used by PolicyCache for background refreshes.
policy = await client.fetch_policy("send_email")
# dict with toolName, mode, constraints, etc.
# Returns None on errorclient.log_decision(request)
Log a client-side validation decision. Fire-and-forget — errors are silently swallowed.
client.log_decision({
"tool_name": "send_email",
"arguments": {"to": "user@example.com"},
"decision": "allow",
"mode": "deterministic",
"latency_ms": 2,
"source": "client",
})await client.register_tools(tools)
from veto.cloud.types import ToolRegistration, ToolParameter
await client.register_tools([
ToolRegistration(
name="send_email",
description="Send an email",
parameters=[
ToolParameter(name="to", type="string", required=True),
ToolParameter(name="subject", type="string", required=True),
ToolParameter(name="body", type="string", required=True),
],
),
])veto.export_decisions(options?)
Export the local decision history as JSON or CSV. Added in v0.6.0.
json_data = veto.export_decisions() # JSON by default
csv_data = veto.export_decisions(format="csv")| Option | Type | Default | Description |
|---|---|---|---|
format | str | "json" | "json" or "csv" |
Each record includes normalized audit fields: timestamp, tool_name, arguments, policy_version, rule_id, decision, reason.
Framework integrations
| Framework | Import | Guide |
|---|---|---|
| LangChain | veto.integrations.langchain | LangChain Integration |
| browser-use | veto.integrations.browser_use | Browser-Use Integration |
CLI commands
| Command | Description |
|---|---|
veto init | Initialize Veto in current directory |
veto version | Show version |