Installation
Install the AgentWarden SDK using pip:
Or add to your requirements.txt:
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
)
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)
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
The unique ID of the agent
The action identifier (e.g., “stripe.refund”)
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
The unique ID of the agent
The action that was performed
Status of the action: “success”, “failed”, “pending”, “denied”
Additional details about the action
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
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:
True if the action can proceed, False if blocked
True if human approval is required before proceeding
Explanation if action is blocked (e.g., “No permission found”, “Amount exceeds limit”)
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:
Unique identifier for the log entry
The agent that performed the action
The action that was logged
Status: “success”, “failed”, “pending”, or “denied”
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
Base exception class for all SDK errors
Invalid or missing API key
Network connectivity issue
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
1. Reuse AgentWarden Instance
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)
3. Cache Permission Checks
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