Python SDK - Veto

The Veto Python SDK provides guardrails for AI agent tool calls. It intercepts and validates tool calls before execution, blocking, allowing, or routing to human approval based on your policies.

Installation

pip install veto-sdk

Quick Start

import asyncio
from veto import Veto

async def main():
    # Initialize Veto
    veto = await Veto.init()
    
    # Wrap your tools
    wrapped_tools = veto.wrap(my_tools)
    
    # Use with your agent framework
    agent = create_agent(tools=wrapped_tools)

if __name__ == "__main__":
    asyncio.run(main())

Core API

Veto.init()

Initialize a Veto instance by loading configuration and rules.

@classmethod
async def init(cls, options: Optional[VetoOptions] = None) -> Veto

options

VetoOptions

Optional configuration options

api_key

str

API key for Veto Cloud (can also use VETO_API_KEY env var)

base_url

str

default:"https://api.runveto.com"

Base URL for Veto Cloud API

mode

'strict' | 'log' | 'shadow'

default:"strict"

Operating mode:

strict: Block denied tool calls
log: Only log denials, allow execution
shadow: Compute decisions but never block

log_level

'debug' | 'info' | 'warn' | 'error' | 'silent'

default:"info"

Log verbosity level

session_id

str

Session ID for tracking (can also use VETO_SESSION_ID env var)

agent_id

str

Agent ID for tracking (can also use VETO_AGENT_ID env var)

user_id

str

User ID for tracking (can also use VETO_USER_ID env var)

role

str

Role for tracking (can also use VETO_ROLE env var)

config_dir

str

default:"./veto"

Path to the veto directory containing config and rules

validation_mode

'cloud' | 'local'

default:"cloud"

Validation mode:

cloud: Use Veto Cloud API
local: Local YAML-based evaluation only

validators

List[Union[Validator, NamedValidator]]

Additional custom validators to run

timeout

int

default:"30000"

API timeout in milliseconds

retries

int

default:"2"

Number of retries for API calls

on_approval_required

Callable

Callback fired when a tool call requires human approval

approval_poll_interval

float

default:"2.0"

Seconds between approval poll requests

approval_timeout

float

default:"300.0"

Max seconds to wait for approval resolution

Examples

veto = await Veto.init()

veto.wrap()

Wrap a list of tools with Veto validation. Types are preserved.

def wrap(self, tools: List[T]) -> List[T]

tools

List[T]

required

List of tools to wrap. Works with any tool format (LangChain, custom).

wrapped

List[T]

Wrapped tools with validation injected. Same type as input.

Examples

from langchain.tools import tool
from veto import Veto

@tool
def search_database(query: str) -> str:
    """Search the database."""
    return perform_search(query)

tools = [search_database]

veto = await Veto.init()
wrapped_tools = veto.wrap(tools)

# Use with LangChain agent
agent = create_react_agent(llm, wrapped_tools)

veto.wrap_tool()

Wrap a single tool with Veto validation.

def wrap_tool(self, tool: T) -> T

tool

required

Single tool to wrap

wrapped

Wrapped tool with validation injected

safe_tool = veto.wrap_tool(my_tool)

veto.guard()

Standalone validation without execution. Useful for checking if a call would be allowed.

async def guard(
    self,
    tool_name: str,
    args: dict[str, Any],
    *,
    session_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    user_id: Optional[str] = None,
    role: Optional[str] = None
) -> GuardResult

tool_name

str

required

Name of the tool to validate

args

dict[str, Any]

required

Arguments to validate

session_id

str

Session ID for this check

agent_id

str

Agent ID for this check

user_id

str

User ID for this check

role

str

Role for this check

result

GuardResult

decision

'allow' | 'deny' | 'require_approval'

required

Validation decision

reason

str

Human-readable reason for the decision

rule_id

str

ID of the matched rule

severity

'critical' | 'high' | 'medium' | 'low' | 'info'

Severity level from the matched rule

approval_id

str

Approval ID if decision is require_approval

shadow

bool

Whether this was evaluated in shadow mode

result = await veto.guard('transfer_funds', {
    'amount': 5000,
    'recipient': 'vendor-123'
})

if result.decision == 'deny':
    print(f'Would be blocked: {result.reason}')

History & Export

veto.get_history_stats()

Get statistics about tool call history.

def get_history_stats(self) -> HistoryStats

stats

HistoryStats

total_calls

int

Total number of tool calls validated

allowed_calls

int

Number of allowed calls

denied_calls

int

Number of denied calls

calls_by_tool

dict[str, int]

Breakdown by tool name

stats = veto.get_history_stats()
print(f'Blocked {stats.denied_calls} out of {stats.total_calls} calls')

veto.clear_history()

Clear the call history.

def clear_history(self) -> None

veto.export_decisions()

Export decision history in JSON or CSV format.

def export_decisions(self, format: DecisionExportFormat) -> str

format

'json' | 'csv'

required

Export format

data

str

Serialized decision history

json_data = veto.export_decisions('json')
csv_data = veto.export_decisions('csv')

with open('decisions.json', 'w') as f:
    f.write(json_data)

with open('decisions.csv', 'w') as f:
    f.write(csv_data)

protect() Function

One-step wrapper for simple use cases. Automatically detects configuration source and applies appropriate policies.

from veto import protect

async def protect(
    tools: Union[T, List[T]],
    *,
    config_dir: Optional[str] = None,
    pack: Optional[str] = None,
    rules: Optional[List[dict]] = None,
    api_key: Optional[str] = None,
    endpoint: Optional[str] = None,
    mode: Optional[str] = None,
    log_level: Optional[str] = None,
    session_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    user_id: Optional[str] = None,
    role: Optional[str] = None,
    on_approval_required: Optional[Callable] = None
) -> Union[T, List[T]]

tools

T | List[T]

required

Tool or list of tools to protect

config_dir

str

Path to veto directory

pack

str

Built-in policy pack to apply (e.g., ‘financial’, ‘browser-automation’)

rules

List[dict]

Inline rules array

api_key

str

Cloud API key

endpoint

str

Self-hosted endpoint

mode

'strict' | 'log' | 'shadow'

Operating mode

log_level

str

Log level

session_id

str

Session ID

agent_id

str

Agent ID

user_id

str

User ID

role

str

Role

on_approval_required

Callable

Approval callback

Examples

from veto import protect

tools = [
    {'name': 'transfer_funds', 'handler': transfer_funds_handler},
    {'name': 'navigate', 'handler': navigate_handler}
]

# Automatically applies @veto/financial and @veto/browser-automation packs
protected = await protect(tools)

Framework Integrations

LangChain

Middleware
Tool Node
Callback

from veto.integrations.langchain import create_veto_middleware
from veto import Veto

veto = await Veto.init()
middleware = create_veto_middleware(
    veto,
    throw_on_deny=True,
    on_allow=lambda name, args: print(f'Allowed: {name}'),
    on_deny=lambda name, args, reason: print(f'Denied: {name}')
)

# Use with LangChain agent
agent = create_react_agent(llm, tools=[middleware])

from veto.integrations.langchain import create_veto_tool_node
from langgraph.prebuilt import ToolNode

veto = await Veto.init()
tool_node = ToolNode(tools)
veto_node = create_veto_tool_node(veto, tool_node)

# Use in LangGraph
from langgraph.graph import StateGraph

graph = StateGraph()
graph.add_node('tools', veto_node)
graph.add_edge('agent', 'tools')

from veto.integrations.langchain import create_veto_callback_handler

handler = create_veto_callback_handler(
    on_tool_start=lambda name, input: print(f'Starting: {name}'),
    on_tool_end=lambda name, output: print(f'Completed: {name}'),
    on_tool_error=lambda name, error: print(f'Error: {name}')
)

result = await agent.ainvoke(
    {'input': 'Transfer funds'},
    config={'callbacks': [handler]}
)

PydanticAI

from pydantic_ai import Agent
from veto.integrations.pydanticai import wrap_pydanticai
from veto import Veto

agent = Agent('openai:gpt-4o', tools=[transfer_funds, send_email])
veto = await Veto.init()

# Wrap the agent
safe_agent = wrap_pydanticai(agent, veto)

# Tool calls are now validated
result = await safe_agent.run('Transfer $5000 to vendor-123')

CrewAI

from crewai import Agent, Task, Crew
from veto.integrations.crewai import wrap_crewai_agent
from veto import Veto

veto = await Veto.init()

# Create agent with tools
agent = Agent(
    role='Financial Assistant',
    goal='Manage financial operations',
    tools=[transfer_funds, check_balance]
)

# Wrap agent with Veto
safe_agent = wrap_crewai_agent(agent, veto)

# Use in Crew
crew = Crew(agents=[safe_agent], tasks=[...])
result = crew.kickoff()

OpenAI Agents SDK

from openai import OpenAI
from veto.integrations.openai_agents import wrap_openai_agent
from veto import Veto

client = OpenAI()
veto = await Veto.init()

# Wrap the client
safe_client = wrap_openai_agent(client, veto)

# Tool calls are now validated
response = await safe_client.agents.run(
    agent_id='asst_123',
    messages=[{'role': 'user', 'content': 'Transfer funds'}]
)

Provider Adapters

Convert between Veto’s tool format and provider-specific formats.

OpenAI

from veto import to_openai, from_openai, to_openai_tools

# Convert to OpenAI format
openai_tools = to_openai_tools(veto_definitions)

# Convert from OpenAI tool call
veto_call = from_openai(openai_tool_call)

Anthropic

from veto import to_anthropic, from_anthropic, to_anthropic_tools

anthropic_tools = to_anthropic_tools(veto_definitions)
veto_call = from_anthropic(anthropic_tool_use)

Google

from veto import to_google_tool, from_google_function_call

google_tool = to_google_tool(veto_definition)
veto_call = from_google_function_call(google_call)

Error Handling

ToolCallDeniedError

Raised when a tool call is blocked in strict mode.

from veto import ToolCallDeniedError

try:
    await wrapped_tool.handler({'amount': 10000})
except ToolCallDeniedError as e:
    print(f'Tool call denied: {e}')
    print(f'Reason: {e.reason}')
    print(f'Tool: {e.tool_name}')

ApprovalTimeoutError

Raised when approval flow times out.

from veto import ApprovalTimeoutError

try:
    await wrapped_tool.handler(args)
except ApprovalTimeoutError:
    print('Approval timed out')

Configuration Types

VetoOptions

from dataclasses import dataclass
from typing import Optional, List, Union, Callable

@dataclass
class VetoOptions:
    api_key: Optional[str] = None
    base_url: Optional[str] = None
    mode: Optional[str] = None
    log_level: Optional[str] = None
    session_id: Optional[str] = None
    agent_id: Optional[str] = None
    user_id: Optional[str] = None
    role: Optional[str] = None
    config_dir: Optional[str] = None
    validation_mode: Optional[str] = None
    validators: Optional[List[Union[Validator, NamedValidator]]] = None
    timeout: Optional[int] = None
    retries: Optional[int] = None
    on_approval_required: Optional[Callable] = None
    approval_poll_interval: Optional[float] = None
    approval_timeout: Optional[float] = None

Validator

Custom validation function.

from typing import Callable, Union
from dataclasses import dataclass

Validator = Callable[[ValidationContext], Union[ValidationResult, Awaitable[ValidationResult]]]

@dataclass
class NamedValidator:
    name: str
    validate: Validator
    description: Optional[str] = None
    priority: Optional[int] = None
    tool_filter: Optional[List[str]] = None

Example

from veto import Veto, VetoOptions, NamedValidator

async def no_delete_validator(context):
    if 'delete' in context.tool_name:
        return {
            'decision': 'deny',
            'reason': 'Delete operations are not allowed'
        }
    return {'decision': 'allow'}

validator = NamedValidator(
    name='no-delete',
    description='Block all delete operations',
    priority=10,
    validate=no_delete_validator
)

veto = await Veto.init(VetoOptions(
    validators=[validator]
))

Advanced Features

Async/Await Patterns

The Python SDK is fully async. All validation and execution happens asynchronously.

import asyncio
from veto import Veto

async def main():
    veto = await Veto.init()
    wrapped_tools = veto.wrap(tools)
    
    # Execute tool calls
    results = await asyncio.gather(
        wrapped_tools[0].handler({'query': 'test1'}),
        wrapped_tools[1].handler({'query': 'test2'})
    )
    
    print(results)

if __name__ == '__main__':
    asyncio.run(main())

Event Webhooks

# In veto/veto.config.yaml:
# events:
#   webhook:
#     url: https://example.com/veto-events
#     on:
#       - decision
#       - approval_required
#     min_severity: high
#     format: slack

veto = await Veto.init()
# Events automatically sent to webhook

Output Validation

Validate tool outputs against patterns.

# In veto/rules/output.yaml
output_rules:
  - id: redact-ssn
    name: Redact SSNs from output
    action: redact
    tools:
      - fetch_user_data
    patterns:
      - '\d{3}-\d{2}-\d{4}'

Type Hints

The SDK provides full type hints for better IDE support:

from veto import (
    Veto,
    VetoOptions,
    VetoMode,
    ValidationMode,
    WrappedTools,
    WrappedHandler,
    GuardResult,
    ToolDefinition,
    ToolCall,
    ToolResult,
    ValidationContext,
    ValidationResult,
    ValidationDecision,
    LogLevel
)

Get Started

Core Concepts

SDKs

Integrations

Policy Packs

Guides

​Installation

​Quick Start

​Core API

​Veto.init()

​Examples

​veto.wrap()

​Examples

​veto.wrap_tool()

​veto.guard()

​History & Export

​veto.get_history_stats()

​veto.clear_history()

​veto.export_decisions()

​protect() Function

​Examples

​Framework Integrations

​LangChain

​PydanticAI

​CrewAI

​OpenAI Agents SDK

​Provider Adapters

​OpenAI

​Anthropic

​Google

​Error Handling

​ToolCallDeniedError

​ApprovalTimeoutError

​Configuration Types

​VetoOptions

​Validator

​Example

​Advanced Features

​Async/Await Patterns

​Event Webhooks

​Output Validation

​Type Hints

Build docs developers (and LLMs) love

Installation

Quick Start

Core API

Veto.init()

Examples

veto.wrap()

Examples

veto.wrap_tool()

veto.guard()

History & Export

veto.get_history_stats()

veto.clear_history()

veto.export_decisions()

protect() Function

Examples

Framework Integrations

LangChain

PydanticAI

CrewAI

OpenAI Agents SDK

Provider Adapters

OpenAI

Anthropic

Google

Error Handling

ToolCallDeniedError

ApprovalTimeoutError

Configuration Types

VetoOptions

Validator

Example

Advanced Features

Async/Await Patterns

Event Webhooks

Output Validation

Type Hints