Vellaveto Python SDK

Python SDK for the Vellaveto agent interaction firewall.

Installation

# Basic installation
pip install vellaveto-sdk

# With LangChain support
pip install vellaveto-sdk[langchain]

# With LangGraph support
pip install vellaveto-sdk[langgraph]

# With CrewAI support
pip install vellaveto-sdk[crewai]

# With Google ADK support
pip install vellaveto-sdk[google-adk]

# With OpenAI Agents SDK support
pip install vellaveto-sdk[openai-agents]

# With Composio support
pip install vellaveto-sdk[composio]

# Full installation (all integrations)
pip install vellaveto-sdk[all]

Quick Start

Direct API Usage

from vellaveto import VellavetoClient, Verdict

client = VellavetoClient(url="http://localhost:3000", api_key="your-key")

# Evaluate a tool call
result = client.evaluate(
    tool="filesystem",
    function="read_file",
    parameters={"path": "/etc/passwd"}
)

if result.verdict == Verdict.ALLOW:
    # Proceed with tool call
    content = read_file("/etc/passwd")
elif result.verdict == Verdict.DENY:
    print(f"Blocked: {result.reason}")
elif result.verdict == Verdict.REQUIRE_APPROVAL:
    print(f"Needs approval: {result.approval_id}")

With Exception Handling

from vellaveto import VellavetoClient, PolicyDenied, ApprovalRequired

client = VellavetoClient(url="http://localhost:3000")

try:
    client.evaluate_or_raise(
        tool="http",
        function="fetch",
        parameters={"url": "https://evil.com/exfil"}
    )
    # Allowed - proceed
except PolicyDenied as e:
    print(f"Blocked by policy: {e.reason}")
except ApprovalRequired as e:
    print(f"Needs approval: {e.approval_id}")

Async Usage

from vellaveto import AsyncVellavetoClient

async with AsyncVellavetoClient(url="http://localhost:3000") as client:
    result = await client.evaluate(
        tool="bash",
        function="execute",
        parameters={"command": "rm -rf /"}
    )

LangChain Integration

Callback Handler

The VellavetoCallbackHandler automatically intercepts and evaluates all tool calls:

from langchain.agents import create_react_agent
from langchain_openai import ChatOpenAI
from vellaveto import VellavetoClient
from vellaveto.langchain import VellavetoCallbackHandler

client = VellavetoClient(url="http://localhost:3000")
handler = VellavetoCallbackHandler(
    client=client,
    session_id="my-session",
    raise_on_deny=True,
)

llm = ChatOpenAI()
agent = create_react_agent(llm, tools, callbacks=[handler])

# All tool calls will be evaluated by Vellaveto
result = agent.invoke({"input": "Read the file /etc/passwd"})

Tool Decorator

Guard individual tools with the @guard decorator:

from langchain.tools import tool
from vellaveto import VellavetoClient
from vellaveto.langchain import VellavetoToolGuard

client = VellavetoClient(url="http://localhost:3000")
guard = VellavetoToolGuard(client)

@tool
@guard("filesystem", "read_file")
def read_file(path: str) -> str:
    """Read a file from the filesystem."""
    with open(path) as f:
        return f.read()

Wrap Existing Toolkits

from langchain_community.agent_toolkits import FileManagementToolkit
from vellaveto import VellavetoClient
from vellaveto.langchain import create_guarded_toolkit

client = VellavetoClient(url="http://localhost:3000")
toolkit = FileManagementToolkit()
guarded_tools = create_guarded_toolkit(client, toolkit)

LangGraph Integration

Vellaveto Node

Add a vellaveto evaluation node to your graph:

from langgraph.graph import StateGraph, END
from vellaveto import VellavetoClient
from vellaveto.langgraph import create_vellaveto_node, VellavetoState

client = VellavetoClient(url="http://localhost:3000")
vellaveto_node = create_vellaveto_node(client, on_deny="block")

class MyState(VellavetoState):
    messages: list
    # ... your fields

graph = StateGraph(MyState)
graph.add_node("plan", plan_node)
graph.add_node("vellaveto", vellaveto_node)
graph.add_node("tools", tool_node)

graph.add_edge("plan", "vellaveto")
graph.add_conditional_edges(
    "vellaveto",
    lambda s: "blocked" if s.get("tool_blocked") else "allowed",
    {"blocked": END, "allowed": "tools"},
)

Guarded Tool Node

Combine tool execution with policy evaluation:

from langgraph.graph import StateGraph
from vellaveto import VellavetoClient
from vellaveto.langgraph import create_vellaveto_tool_node

client = VellavetoClient(url="http://localhost:3000")
tool_node = create_vellaveto_tool_node(client, [read_file, write_file])

graph = StateGraph(MyState)
graph.add_node("tools", tool_node)  # Policy evaluation + execution

CrewAI Integration

Guard tool calls in CrewAI crews with VellavetoCrewGuard:

from crewai import Agent, Crew, Task
from vellaveto import VellavetoClient
from vellaveto.crewai import VellavetoCrewGuard

client = VellavetoClient(url="http://localhost:3000")
guard = VellavetoCrewGuard(client)

agent = Agent(role="researcher", tools=[web_search])
task = Task(description="Research topic", agent=agent)
crew = Crew(agents=[agent], tasks=[task])

# Option 1: Wrap entire crew execution
result = guard.kickoff(crew)

# Option 2: Wrap individual tools
guarded_tool = guard.wrap_tool(web_search)

Google ADK Integration

Guard tools in Google Agent Development Kit applications with VellavetoADKGuard:

from google.adk import Agent
from vellaveto import VellavetoClient
from vellaveto.google_adk import VellavetoADKGuard

client = VellavetoClient(url="http://localhost:3000")
guard = VellavetoADKGuard(client)

# Option 1: Decorator
@guard.protect
def search_web(query: str) -> str:
    return do_search(query)

# Option 2: Before-tool callback
agent = Agent(
    tools=[search_web],
    before_tool_call=guard.before_tool_callback(),
)

OpenAI Agents SDK Integration

Enforce policies on OpenAI Agents SDK function calls with VellavetoAgentGuard:

from agents import Agent, Runner
from vellaveto import VellavetoClient
from vellaveto.openai_agents import VellavetoAgentGuard

client = VellavetoClient(url="http://localhost:3000")
guard = VellavetoAgentGuard(client)

# Wrap tools with policy enforcement
agent = Agent(
    name="assistant",
    tools=[guard.wrap_function(read_file), guard.wrap_function(web_search)],
)
result = Runner.run_sync(agent, "Read the config file")

Composio Integration

Guard Composio tool calls across any provider (OpenAI, LangChain, CrewAI, AutoGen) with ComposioGuard:

from composio import Composio
from vellaveto import VellavetoClient
from vellaveto.composio import ComposioGuard

client = VellavetoClient(url="http://localhost:3000", api_key="key")
guard = ComposioGuard(client, session_id="sess-1")

composio = Composio(api_key="...")
tools = composio.tools.get(
    user_id="default",
    toolkits=["GITHUB"],
    modifiers=[guard.before_execute_modifier(), guard.after_execute_modifier()],
)

Claude Agent SDK Integration

Enforce tool permissions in Anthropic Claude Agent SDK applications with VellavetoToolPermission:

from claude_agent_sdk import Agent, tool
from vellaveto import VellavetoClient
from vellaveto.claude_agent import VellavetoToolPermission

client = VellavetoClient(url="http://localhost:3000")
permission = VellavetoToolPermission(client)

@tool
def read_file(path: str) -> str:
    return open(path).read()

agent = Agent(
    tools=[read_file],
    tool_permission_callback=permission.check,
)

AWS Strands Agents Integration

Guard tool calls in AWS Strands Agents applications with VellavetoStrandsGuard:

from strands import Agent
from strands.tools import tool
from vellaveto import VellavetoClient
from vellaveto.strands import VellavetoStrandsGuard

client = VellavetoClient(url="http://localhost:3000")
guard = VellavetoStrandsGuard(client)

@tool
def read_file(path: str) -> str:
    return open(path).read()

agent = Agent(
    tools=[guard.wrap_tool(read_file)],
)

Microsoft Agent Framework Integration

Intercept tool calls in Microsoft Agent Framework (AutoGen + Semantic Kernel) with VellavetoAgentMiddleware:

from vellaveto import VellavetoClient
from vellaveto.microsoft_agents import VellavetoAgentMiddleware

client = VellavetoClient(url="http://localhost:3000")
middleware = VellavetoAgentMiddleware(client)

# Use as middleware in Microsoft Agent Framework
agent = Agent(
    middleware=[middleware],
    tools=[read_file, web_search],
)

Response Attestation

Verify that proxied responses haven't been tampered with:

import os
key = os.environ["VELLAVETO_ATTESTATION_SECRET"].encode()
result = client.evaluate(action)
is_valid = VellavetoClient.verify_attestation(result.raw_response, key)

Security guarantees (R260): verify_attestation() validates that the signature is exactly 64 hex characters (SHA-256 HMAC), rejects non-integer version/scan_passes, non-string trust_tier, and non-numeric timestamp fields before constructing the signing content. This prevents type coercion attacks where malformed fields produce incorrect HMAC inputs.

Parameter Redaction

Strip sensitive parameter values before they transit the network to Vellaveto:

from vellaveto import VellavetoClient, ParameterRedactor

# Keys-only (default): redacts values for known sensitive parameter names
redactor = ParameterRedactor()

# Values: also scans string values for secret patterns (sk-..., ghp_..., JWTs)
redactor = ParameterRedactor(mode="values")

# All: redacts every parameter value (sends only keys to Vellaveto)
redactor = ParameterRedactor(mode="all")

# Custom sensitive keys (added to defaults)
redactor = ParameterRedactor(extra_keys={"internal_token", "vault_path"})

client = VellavetoClient(url="http://localhost:3000", redactor=redactor)

# api_key value is replaced with "[REDACTED]" before sending
client.evaluate(
    tool="http",
    function="fetch",
    parameters={"url": "https://api.example.com", "api_key": "sk-secret123"},
)

Configuration

Client Options

client = VellavetoClient(
    url="http://localhost:3000",  # Vellaveto server URL
    api_key="your-api-key",       # API key for authentication
    timeout=30.0,                  # Request timeout in seconds
    verify_ssl=True,               # Verify SSL certificates
    redactor=ParameterRedactor(),  # Client-side parameter redaction (optional)
)

Evaluation Context

Provide context for stateful policy evaluation:

from vellaveto.types import EvaluationContext

context = EvaluationContext(
    session_id="user-session-123",
    agent_id="my-agent",
    tenant_id="tenant-abc",
    call_chain=["tool1", "tool2"],
    metadata={"user": "alice"},
)

result = client.evaluate(
    tool="filesystem",
    function="write_file",
    parameters={"path": "/tmp/output.txt", "content": "data"},
    context=context,
)

API Reference

VellavetoClient

Types

Exceptions

Development

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Type checking
mypy vellaveto

# Linting
ruff check vellaveto

License

See LICENSE and LICENSING.md for package and repository licensing details.