Python preview package for local tool-call checks

from fastapi import FastAPI, HTTPException
import os
from veto_sdk import Veto, ToolCallDeniedError
from pydantic import BaseModel

app = FastAPI()
veto: Veto | None = None

@app.on_event("startup")
async def startup():
    global veto
    veto = await Veto.init(api_key=os.environ["VETO_API_KEY"])

class AgentRequest(BaseModel):
    tool_name: str
    arguments: dict

@app.post("/agent/execute")
async def execute_tool(req: AgentRequest):
    decision = await veto.guard(req.tool_name, req.arguments)

    if decision.decision == "deny":
        raise HTTPException(status_code=403, detail=decision.reason)

    if decision.decision == "require_approval":
        return {
            "status": "pending_approval",
            "approval_id": decision.approval_id,
            "message": decision.reason,
        }

    result = await run_tool(req.tool_name, req.arguments)
    return {"status": "executed", "result": result}

# views.py
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
import os
from veto_sdk import Veto
import asyncio
import json

_veto = None

async def get_veto():
    global _veto
    if _veto is None:
        _veto = await Veto.init()
    return _veto

@csrf_exempt
def agent_execute(request):
    if request.method != "POST":
        return JsonResponse({"error": "Method not allowed"}, status=405)

    body = json.loads(request.body)
    veto = asyncio.run(get_veto())

    decision = asyncio.run(
        veto.guard(body["tool_name"], body["arguments"])
    )

    if decision.decision == "deny":
        return JsonResponse({"error": decision.reason}, status=403)

    result = execute_tool(body["tool_name"], body["arguments"])
    return JsonResponse({"result": result})

import os
from veto_sdk import Veto, to_openai_tools, from_openai_tool_call
from openai import AsyncOpenAI

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

# Define tools and wrap with Veto
my_tools = [search_web, send_email, read_database]
wrapped_tools = veto.wrap(my_tools)

# Convert to OpenAI format
openai_tools = to_openai_tools(wrapped_tools)

response = await client.chat.completions.create(
    model=os.environ["OPENAI_MODEL"],
    messages=[{"role": "user", "content": "Search for Q3 revenue data"}],
    tools=openai_tools,
)

# When OpenAI returns a tool call, convert and validate
for tool_call in response.choices[0].message.tool_calls:
    veto_call = from_openai_tool_call(tool_call)
    # Guardrails already enforced by governed tools
    result = await execute_wrapped(veto_call)

import os
from veto_sdk import Veto, to_anthropic_tools, from_anthropic_tool_use
import anthropic

client = anthropic.AsyncAnthropic()
veto = await Veto.init()

wrapped_tools = veto.wrap([search_web, send_email])
claude_tools = to_anthropic_tools(wrapped_tools)

response = await client.messages.create(
    model=os.environ["ANTHROPIC_MODEL"],
    max_tokens=1024,
    messages=[{"role": "user", "content": "Send the report to the team"}],
    tools=claude_tools,
)

for block in response.content:
    if block.type == "tool_use":
        veto_call = from_anthropic_tool_use(block)
        result = await execute_wrapped(veto_call)

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from pydantic import BaseModel, Field
import os
from veto_sdk import Veto

class TransferInput(BaseModel):
    amount: float = Field(description="Amount to transfer")
    to_account: str = Field(description="Destination account ID")

@tool("transfer_money", args_schema=TransferInput)
def transfer_money(amount: float, to_account: str) -> str:
    """Transfer money to a bank account."""
    return f"Transferred $" + str(amount) + f" to {to_account}"

veto = await Veto.init()
wrapped_tools = veto.wrap([transfer_money])

llm = ChatOpenAI(model=os.environ["OPENAI_MODEL"])
agent = create_react_agent(llm, wrapped_tools)

from dataclasses import dataclass
from typing import Any

@dataclass
class DatabaseQuery:
    name: str = "query_database"
    description: str = "Run a SQL query against the production database"

    async def handler(self, args: dict[str, Any]) -> str:
        query = args["sql"]
        # Execute query here.
        return f"Query returned {len(rows)} rows"

@dataclass
class FileWriter:
    name: str = "write_file"
    description: str = "Write content to a file on disk"

    async def handler(self, args: dict[str, Any]) -> str:
        path, content = args["path"], args["content"]
        # Write file here.
        return f"Wrote {len(content)} bytes to {path}"

veto = await Veto.init()
guarded_tools = veto.wrap([DatabaseQuery(), FileWriter()])