useveto

Runtime authorization for AI agents — Python SDK.

What is Veto?

Veto intercepts every tool call your AI agent makes, evaluates it against your policies, and decides: allow, deny, or escalate. Sub-10ms. Default deny.

• Default deny — no matching policy = blocked
• Full audit trail — every decision logged
• Edge-first — powered by Cloudflare Workers
• Sync + async — both clients included

Install

pip install useveto

Quick Start

from veto import VetoClient

client = VetoClient(api_key="veto_...")

# Check if an agent can perform an action
result = client.authorize("agent-uuid", "file.write", {"path": "/etc/passwd"})

if not result.allowed:
    print(f"Blocked: {result.reason}")

Async

from veto import AsyncVetoClient

async with AsyncVetoClient(api_key="veto_...") as client:
    result = await client.authorize("agent-uuid", "send_email", {
        "to": "user@example.com",
        "subject": "Refund confirmation",
    })

LangChain Integration

from veto import VetoClient, VetoError

veto = VetoClient(api_key="veto_...")

def veto_tool_wrapper(tool_func, tool_name: str, agent_id: str):
    """Wrap any LangChain tool with Veto authorization."""
    def wrapper(*args, **kwargs):
        result = veto.authorize(agent_id, tool_name, kwargs)
        if not result.allowed:
            return f"Authorization denied: {result.reason}"
        return tool_func(*args, **kwargs)
    wrapper.__name__ = tool_func.__name__
    wrapper.__doc__ = tool_func.__doc__
    return wrapper

CrewAI Integration

from veto import VetoClient

veto = VetoClient(api_key="veto_...")

def before_tool_callback(agent_id: str):
    def callback(tool_name: str, tool_input: dict) -> bool:
        result = veto.authorize(agent_id, tool_name, tool_input)
        return result.allowed
    return callback

API Reference

VetoClient / AsyncVetoClient

client = VetoClient(
    api_key="veto_...",       # required
    endpoint="https://api.veto.tools",  # optional
    timeout=5.0,              # optional, seconds
)

Authorization

client.authorize(agent_id, tool_name, parameters={}) -> AuthorizationResult

Agents

client.create_agent(CreateAgentInput(name="Bot", description="...")) -> Agent
client.list_agents() -> List[Agent]
client.get_agent(agent_id) -> Agent
client.delete_agent(agent_id) -> None

Policies

client.create_policy(CreatePolicyInput(
    agent_id="...",
    name="Allow reads",
    rules=[PolicyRule(type="tool_allowlist", tools=["file.read"])],
    priority=10,
)) -> Policy

client.list_policies(agent_id=None) -> List[Policy]
client.get_policy(policy_id) -> Policy
client.update_policy(policy_id, UpdatePolicyInput(name="New name")) -> Policy
client.delete_policy(policy_id) -> None

Audit Logs

client.query_audit_log(
    agent_id=None,
    tool_name=None,
    result=None,  # "allowed" | "denied" | "escalated"
    limit=100,
    offset=0,
) -> PaginatedAuditLogResponse

Error Handling

from veto import VetoError, UnauthorizedError, RateLimitError

try:
    result = client.authorize("agent", "tool")
except RateLimitError as e:
    print(f"Retry after {e.retry_after_ms}ms")
except UnauthorizedError:
    print("Invalid API key")
except VetoError as e:
    print(f"{e.code}: {e}")

Context Manager

# Sync
with VetoClient(api_key="veto_...") as client:
    result = client.authorize("agent-1", "file.read")

# Async
async with AsyncVetoClient(api_key="veto_...") as client:
    result = await client.authorize("agent-1", "file.read")

Links

• Website: veto.tools
• Dashboard: app.veto.tools
• API Docs: docs.veto.tools
• GitHub: github.com/useveto/python

License

MIT