Vaikora Python SDK

The official Python SDK for the Vaikora AI Agent Security Platform.

Installation

pip install vaikora

Quick Start

Simple Configuration

import vaikora

# Configure the SDK (do once at startup)
vaikora.configure(api_key="vk_your_api_key")

# Register your AI agent
agent = await vaikora.register(
    name="my-ai-agent",
    agent_type="autonomous",
    capabilities=["database_write", "api_call"]
)

print(f"Agent registered: {agent.id}")

Using the Interceptor Decorator

The SDK provides a powerful decorator for securing AI agent actions:

import vaikora

vaikora.configure(api_key="vk_your_api_key")

# Register agent first
await vaikora.register(name="my-agent", agent_type="autonomous")

@vaikora.interceptor(
    action_type="database.write",
    resource="production",
    require_approval=True,
    validate_inputs=True,
)
async def write_to_database(data: dict) -> bool:
    # Your code here - Vaikora validates and checks policies first
    await database.insert(data)
    return True

# Usage - automatically secured
result = await write_to_database({"user": "john", "action": "update"})

Data Validation

Validate data for PII, anomalies, and toxicity:

import vaikora

vaikora.configure(api_key="vk_your_api_key")

# Validate data before processing
result = await vaikora.validate_data(
    data={"email": "user@example.com", "ssn": "123-45-6789"},
    check_pii=True,
    check_anomalies=True,
    check_toxicity=True,
)

if result.pii_detected:
    print(f"PII found: {[p.pii_type for p in result.pii_detections]}")
    # Use cleaned data
    clean_data = result.cleaned_data

Full Client Usage

For more control, use the client directly:

from vaikora import VaikoraClient

# Initialize the client
client = VaikoraClient(api_key="vk_your_api_key")

# Register an agent
agent = await client.agents.register(
    name="my-ai-agent",
    agent_type="autonomous",
    capabilities=["database_write", "api_call", "file_access"]
)

# Submit an action for policy evaluation
result = await client.actions.submit(
    agent_id=str(agent.id),
    action_type="database_write",
    resource="users_table",
    payload={"user_id": 123, "data": {"name": "John"}}
)

if result.approved:
    # Proceed with the action
    pass
else:
    print(f"Action blocked: {result.denial_reason}")

Synchronous Usage

For synchronous code, use the sync client:

from vaikora import VaikoraClientSync

client = VaikoraClientSync(api_key="your-api-key")

# All methods are synchronous
agent = client.agents.register(
    name="my-sync-agent",
    agent_type="autonomous"
)

Features

• Action Control: Submit actions for policy evaluation before execution
• Anomaly Detection: Automatic detection of unusual agent behavior
• Policy Enforcement: Server-side policy evaluation with local caching
• Alerting: Real-time alerts for security events
• Audit Logging: Immutable action logs with hash chains

Configuration

from vaikora import VaikoraClient

client = VaikoraClient(
    api_key="your-api-key",
    base_url="https://api.vaikora.com",  # Default
    timeout=30.0,  # Request timeout in seconds
    retry_count=3,  # Number of retries for failed requests
    agent_id="default-agent-id",  # Default agent for actions
)

API Reference

Agents

# Register a new agent
agent = await client.agents.register(
    name="my-agent",
    agent_type="autonomous",  # autonomous, semi_autonomous, or supervised
    capabilities=["database_write", "api_call"],
    metadata={"version": "1.0"}
)

# Get agent by ID
agent = await client.agents.get(agent_id)

# List all agents
agents = await client.agents.list(page=1, page_size=20)

# Update agent
agent = await client.agents.update(agent_id, name="new-name")

# Deactivate agent
await client.agents.deactivate(agent_id)

Actions

# Submit action for evaluation
result = await client.actions.submit(
    agent_id="agent-uuid",
    action_type="database_write",
    resource="users_table",
    payload={"query": "UPDATE users SET name='John' WHERE id=123"},
    metadata={"source": "user_request"}
)

# Check result
if result.approved:
    print(f"Action approved: {result.action_id}")
else:
    print(f"Action denied: {result.denial_reason}")

# Get action details
action = await client.actions.get(action_id)

# List actions with filters
actions = await client.actions.list(
    agent_id="agent-uuid",
    action_type="database_write",
    status="approved",
    start_date=datetime(2024, 1, 1),
    end_date=datetime.now()
)

Policies

# List policies
policies = await client.policies.list()

# Get policy details
policy = await client.policies.get(policy_id)

Alerts

# List alerts
alerts = await client.alerts.list(status="open", severity="high")

# Get alert details
alert = await client.alerts.get(alert_id)

# Acknowledge alert
await client.alerts.acknowledge(alert_id)

# Resolve alert
await client.alerts.resolve(alert_id, notes="Issue addressed")

Error Handling

from vaikora import VaikoraClient
from vaikora.exceptions import (
    VaikoraError,
    AuthenticationError,
    RateLimitError,
    PolicyViolationError,
    NetworkError,
)

client = VaikoraClient(api_key="your-api-key")

try:
    result = await client.actions.submit(...)
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except PolicyViolationError as e:
    print(f"Policy violation: {e.policy_name} - {e.message}")
except NetworkError:
    print("Network error. Check your connection.")
except VaikoraError as e:
    print(f"Vaikora error: {e}")

License

MIT License - see LICENSE for details.