Installation

Install the AgentWarden SDK using pip:
pip install agentwarden
Or add to your requirements.txt:
agentwarden>=1.0.0

Requirements

  • Python 3.8 or higher
  • requests library (installed automatically)

Quick Start

from agentwarden import AgentWarden

# Initialize the SDK
guard = AgentWarden(api_key="your-api-key-here")

# Check permission
result = guard.check(
    agent_id="customer-support-bot",
    action="stripe.refund",
    context={"amount": 50.00}
)

if result.allowed:
    # Execute action
    process_refund()
    
    # Log success
    guard.log(
        agent_id="customer-support-bot",
        action="stripe.refund",
        status="success",
        context={"amount": 50.00}
    )

AgentWarden Class

Constructor

AgentWarden(
    api_key: str,
    base_url: str = "https://api.agentwarden.io",
    timeout: int = 30
)
api_key
string
required
Your organization’s API key from the dashboard
base_url
string
default:"https://api.agentwarden.io"
Base URL for the API (change for self-hosted instances)
timeout
int
default:"30"
Request timeout in seconds
Example:
from agentwarden import AgentWarden

# Standard usage
guard = AgentWarden(api_key="ak_1234567890abcdef")

# Custom configuration
guard = AgentWarden(
    api_key="ak_1234567890abcdef",
    base_url="https://custom.agentwarden.io",
    timeout=60
)

Methods

check()

Check if an agent has permission to perform an action.
check(
    agent_id: str,
    action: str,
    context: dict = None
) -> CheckResponse
agent_id
string
required
The unique ID of the agent
action
string
required
The action identifier (e.g., “stripe.refund”)
context
dict
Additional context for the permission check (e.g., amount, resource ID)
Returns: CheckResponse object Example:
result = guard.check(
    agent_id="support-bot",
    action="stripe.refund",
    context={
        "amount": 150.00,
        "customer_id": "cus_123",
        "reason": "defective_product"
    }
)

print(result.allowed)            # True or False
print(result.requires_approval)  # True or False
print(result.reason)             # Reason if blocked

log()

Log an agent action for audit trail.
log(
    agent_id: str,
    action: str,
    status: str,
    context: dict = None,
    ip_address: str = None
) -> LogResponse
agent_id
string
required
The unique ID of the agent
action
string
required
The action that was performed
status
string
required
Status of the action: “success”, “failed”, “pending”, “denied”
context
dict
Additional details about the action
ip_address
string
IP address where the action originated
Returns: LogResponse object Example:
# Log successful action
log_result = guard.log(
    agent_id="support-bot",
    action="stripe.refund",
    status="success",
    context={
        "amount": 150.00,
        "customer_id": "cus_123",
        "refund_id": "re_abc123"
    },
    ip_address="192.168.1.1"
)

print(log_result.id)         # Log entry ID
print(log_result.created_at) # Timestamp

get_approval_status()

Get the current status of an approval request.
get_approval_status(
    approval_id: str
) -> str
approval_id
string
required
The unique ID of the approval request
Returns: String - “pending”, “approved”, “denied” Example:
status = guard.get_approval_status("apr_xyz789")

if status == "approved":
    print("Approval granted!")
elif status == "denied":
    print("Approval denied")
else:
    print("Still pending...")

Response Objects

CheckResponse

Returned by the check() method.
class CheckResponse:
    allowed: bool           # Whether action is allowed
    requires_approval: bool # Whether action needs approval
    reason: str | None      # Reason if blocked
    approval_id: str | None # Approval ID if requires_approval
Attributes:
allowed
boolean
True if the action can proceed, False if blocked
requires_approval
boolean
True if human approval is required before proceeding
reason
string | null
Explanation if action is blocked (e.g., “No permission found”, “Amount exceeds limit”)
approval_id
string | null
Unique ID of the approval request (only if requires_approval is True)
Usage:
result = guard.check(agent_id, action, context)

if result.allowed:
    execute_action()
elif result.requires_approval:
    wait_for_approval(result.approval_id)
else:
    handle_blocked_action(result.reason)

LogResponse

Returned by the log() method.
class LogResponse:
    id: str              # Unique log entry ID
    agent_id: str        # Agent that performed the action
    action: str          # Action that was performed
    status: str          # Status of the action
    created_at: datetime # When the log was created
Attributes:
id
string
Unique identifier for the log entry
agent_id
string
The agent that performed the action
action
string
The action that was logged
status
string
Status: “success”, “failed”, “pending”, or “denied”
created_at
datetime
Timestamp when the log was created

Error Handling

The SDK raises exceptions for various error conditions:
from agentwarden import AgentWarden, AgentWardenError

try:
    guard = AgentWarden(api_key="invalid_key")
    result = guard.check("agent-id", "action")
    
except AgentWardenError as e:
    print(f"AgentWarden error: {e}")
    # Handle error (invalid API key, network issue, etc.)
    
except Exception as e:
    print(f"Unexpected error: {e}")

Exception Types

AgentWardenError
exception
Base exception class for all SDK errors
AuthenticationError
exception
Invalid or missing API key
RateLimitError
exception
API rate limit exceeded
NetworkError
exception
Network connectivity issue
ValidationError
exception
Invalid parameters passed to method

Common Patterns

Pattern 1: Check-Execute-Log

The recommended pattern for all agent actions:
def execute_agent_action(agent_id, action, context, execute_fn):
    """
    Generic pattern for agent actions with permission checks and logging
    
    Args:
        agent_id: Agent identifier
        action: Action name
        context: Action context
        execute_fn: Function to execute if allowed
    """
    # 1. Check permission
    result = guard.check(agent_id, action, context)
    
    if not result.allowed:
        # Log denial
        guard.log(agent_id, action, "denied", context)
        
        if result.requires_approval:
            return {"status": "pending_approval", "approval_id": result.approval_id}
        else:
            return {"status": "denied", "reason": result.reason}
    
    # 2. Execute action
    try:
        outcome = execute_fn()
        
        # 3. Log success
        guard.log(agent_id, action, "success", {**context, "result": outcome})
        
        return {"status": "success", "result": outcome}
        
    except Exception as e:
        # 3. Log failure
        guard.log(agent_id, action, "failed", {**context, "error": str(e)})
        
        return {"status": "failed", "error": str(e)}

# Usage
result = execute_agent_action(
    agent_id="support-bot",
    action="stripe.refund",
    context={"amount": 50},
    execute_fn=lambda: process_refund(50)
)

Pattern 2: Decorator for Agent Actions

Create a decorator to automatically check permissions:
from functools import wraps

def requires_permission(action):
    """
    Decorator that checks permissions before executing function
    """
    def decorator(func):
        @wraps(func)
        def wrapper(agent_id, *args, **kwargs):
            # Extract context from kwargs
            context = kwargs.pop('context', {})
            
            # Check permission
            result = guard.check(agent_id, action, context)
            
            if not result.allowed:
                if result.requires_approval:
                    raise PermissionError(f"Approval required: {result.approval_id}")
                else:
                    raise PermissionError(f"Permission denied: {result.reason}")
            
            # Execute function
            try:
                outcome = func(agent_id, *args, **kwargs)
                
                # Log success
                guard.log(agent_id, action, "success", context)
                
                return outcome
                
            except Exception as e:
                # Log failure
                guard.log(agent_id, action, "failed", {**context, "error": str(e)})
                raise
        
        return wrapper
    return decorator

# Usage
@requires_permission("stripe.refund")
def process_refund(agent_id, customer_id, amount):
    stripe.Refund.create(
        customer=customer_id,
        amount=int(amount * 100)
    )
    return {"refund_id": "re_123"}

# Call the function
try:
    result = process_refund(
        "support-bot",
        "cus_123",
        50.00,
        context={"customer_id": "cus_123", "amount": 50.00}
    )
except PermissionError as e:
    print(f"Permission denied: {e}")

Pattern 3: Batch Permission Checks

Check multiple permissions at once:
def check_multiple_permissions(agent_id, actions_with_context):
    """
    Check multiple permissions in one go
    
    Args:
        agent_id: Agent identifier
        actions_with_context: List of (action, context) tuples
    
    Returns:
        Dict of action -> CheckResponse
    """
    results = {}
    
    for action, context in actions_with_context:
        results[action] = guard.check(agent_id, action, context)
    
    return results

# Usage
permissions = check_multiple_permissions(
    "support-bot",
    [
        ("stripe.refund", {"amount": 50}),
        ("email.send", {"count": 100}),
        ("zendesk.ticket.update", {})
    ]
)

# Check all permissions
all_allowed = all(r.allowed for r in permissions.values())
if all_allowed:
    execute_batch_operation()

Pattern 4: Retry with Approval

Wait for approval and retry:
import time

def execute_with_approval_retry(agent_id, action, context, execute_fn, max_wait=3600):
    """
    Execute action, wait for approval if needed, then retry
    
    Args:
        agent_id: Agent identifier
        action: Action name
        context: Context
        execute_fn: Function to execute
        max_wait: Max seconds to wait for approval
    
    Returns:
        Result of execution
    """
    result = guard.check(agent_id, action, context)
    
    if result.allowed:
        # Execute immediately
        return execute_fn()
    
    elif result.requires_approval:
        print(f"⏳ Waiting for approval: {result.approval_id}")
        
        # Poll for approval
        start_time = time.time()
        while time.time() - start_time < max_wait:
            status = guard.get_approval_status(result.approval_id)
            
            if status == "approved":
                print("✅ Approved! Executing...")
                return execute_fn()
            
            elif status == "denied":
                print("❌ Denied")
                raise PermissionError("Action was denied by approver")
            
            # Wait 30 seconds before checking again
            time.sleep(30)
        
        raise TimeoutError("Approval timeout reached")
    
    else:
        raise PermissionError(f"Action blocked: {result.reason}")

# Usage
try:
    result = execute_with_approval_retry(
        "devops-agent",
        "deploy.production",
        {"version": "v2.1.0"},
        lambda: deploy_to_production("v2.1.0")
    )
except (PermissionError, TimeoutError) as e:
    print(f"Failed: {e}")

Environment Variables

You can configure the SDK using environment variables:
export AGENTWARDEN_API_KEY="your-api-key"
export AGENTWARDEN_BASE_URL="https://api.agentwarden.io"
export AGENTWARDEN_TIMEOUT="30"
Then initialize without parameters:
from agentwarden import AgentWarden

# Reads from environment variables
guard = AgentWarden()

Testing

Mocking in Tests

from unittest.mock import Mock, patch
import pytest

def test_agent_action():
    # Mock the AgentWarden SDK
    with patch('agentwarden.AgentWarden') as MockGuard:
        mock_guard = MockGuard.return_value
        
        # Mock check response
        mock_guard.check.return_value = Mock(
            allowed=True,
            requires_approval=False,
            reason=None
        )
        
        # Your code that uses the SDK
        result = guard.check("agent-id", "action", {})
        
        assert result.allowed == True
        mock_guard.check.assert_called_once()

Test Mode

AgentWarden provides a test mode for development:
# In development/testing
guard = AgentWarden(
    api_key="test_key",
    base_url="https://test.agentwarden.io"
)
Test API keys start with test_ and don’t count against your plan limits.

Advanced Configuration

Custom HTTP Client

import requests
from agentwarden import AgentWarden

# Create custom session
session = requests.Session()
session.headers.update({'User-Agent': 'MyApp/1.0'})

# Pass to AgentWarden
guard = AgentWarden(
    api_key="your-key",
    http_client=session
)

Logging

Enable SDK logging for debugging:
import logging

# Enable AgentWarden SDK logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger('agentwarden')
logger.setLevel(logging.DEBUG)

guard = AgentWarden(api_key="your-key")
# All SDK calls will now be logged

Performance Tips

Create one instance and reuse it across your application. Don’t create a new instance for each request.
    # Good
    guard = AgentWarden(api_key="...")
    
    def action1():
        guard.check(...)
    
    def action2():
        guard.check(...)
If logging many actions, consider batching them:
    logs = []
    for item in items:
        logs.append({
            "agent_id": "bot",
            "action": "process",
            "status": "success",
            "context": {"item": item}
        })
    
    # Batch log (future feature)
    guard.log_batch(logs)
For identical actions, cache permission check results (with short TTL):
    from functools import lru_cache
    import time
    
    @lru_cache(maxsize=1000)
    def cached_check(agent_id, action, context_hash):
        return guard.check(agent_id, action, context)
    
    # Use with care - permissions can change!
For async applications, use the async SDK (coming soon):
    from agentwarden import AsyncAgentWarden
    
    guard = AsyncAgentWarden(api_key="...")
    result = await guard.check(...)

Next Steps