Overview
This guide provides complete, production-ready examples of integrating AgentWarden into various types of applications.E-commerce Customer Support Bot
A complete example of an AI customer support agent with tiered refund permissions.Requirements
Copy
pip install agentwarden stripe openai
Complete Implementation
Copy
import os
from agentwarden import AgentWarden
import stripe
from openai import OpenAI
# Initialize clients
guard = AgentWarden(api_key=os.getenv('AGENTWARDEN_API_KEY'))
stripe.api_key = os.getenv('STRIPE_API_KEY')
openai_client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
AGENT_ID = "customer-support-bot"
class CustomerSupportAgent:
"""AI-powered customer support with permission controls"""
def __init__(self):
self.guard = guard
self.agent_id = AGENT_ID
def process_refund_request(self, customer_id: str, order_id: str,
amount: float, reason: str) -> dict:
"""
Process a refund request with permission checking
Returns:
dict with status: 'approved', 'pending_approval', or 'denied'
"""
# Determine action based on amount
if amount <= 50:
action = "stripe.refund.small"
elif amount <= 200:
action = "stripe.refund.medium"
else:
action = "stripe.refund.large"
# Check permission with AgentWarden
result = self.guard.check(
agent_id=self.agent_id,
action=action,
context={
"amount": amount,
"customer_id": customer_id,
"order_id": order_id,
"reason": reason
}
)
if result.allowed:
# Process refund immediately
try:
refund = stripe.Refund.create(
charge=order_id,
amount=int(amount * 100), # Stripe uses cents
reason=reason
)
# Log success
self.guard.log(
agent_id=self.agent_id,
action=action,
status="success",
context={
"amount": amount,
"refund_id": refund.id,
"customer_id": customer_id,
"order_id": order_id
}
)
return {
"status": "approved",
"refund_id": refund.id,
"message": f"Refund of ${amount} processed successfully"
}
except stripe.error.StripeError as e:
# Log failure
self.guard.log(
agent_id=self.agent_id,
action=action,
status="failed",
context={
"amount": amount,
"error": str(e),
"error_type": type(e).__name__
}
)
return {
"status": "failed",
"message": f"Refund failed: {str(e)}"
}
elif result.requires_approval:
# Requires human approval
self.notify_admin_for_approval(
approval_id=result.approval_id,
amount=amount,
customer_id=customer_id,
order_id=order_id,
reason=reason
)
return {
"status": "pending_approval",
"approval_id": result.approval_id,
"message": f"Refund of ${amount} requires manager approval"
}
else:
# Denied
self.guard.log(
agent_id=self.agent_id,
action=action,
status="denied",
context={
"amount": amount,
"reason": result.reason
}
)
return {
"status": "denied",
"message": result.reason
}
def notify_admin_for_approval(self, approval_id: str, amount: float,
customer_id: str, order_id: str, reason: str):
"""Send notification to admin for approval"""
# Send email, Slack message, etc.
print(f"""
🔔 APPROVAL NEEDED
Refund Request: ${amount}
Customer: {customer_id}
Order: {order_id}
Reason: {reason}
Review at: https://app.agentwarden.io/approvals/{approval_id}
""")
def handle_customer_query(self, customer_message: str,
customer_id: str) -> str:
"""
Use AI to understand customer query and take appropriate action
"""
# Get AI response
response = openai_client.chat.completions.create(
model="gpt-4",
messages=[
{
"role": "system",
"content": """You are a helpful customer support agent.
If the customer requests a refund, extract:
- order_id
- amount
- reason
Respond in JSON format: {"action": "refund", "data": {...}}
"""
},
{"role": "user", "content": customer_message}
]
)
ai_response = response.choices[0].message.content
# Parse AI response and execute actions
# (simplified - add proper JSON parsing)
if "refund" in ai_response.lower():
# Extract details and process refund
result = self.process_refund_request(
customer_id=customer_id,
order_id="ch_example123", # Extract from AI
amount=50.00, # Extract from AI
reason="customer_request"
)
return f"Refund status: {result['status']}"
return ai_response
# Usage
agent = CustomerSupportAgent()
# Scenario 1: Small refund (auto-approved)
result = agent.process_refund_request(
customer_id="cus_ABC123",
order_id="ch_1234567890",
amount=25.00,
reason="defective_product"
)
print(result) # {'status': 'approved', ...}
# Scenario 2: Large refund (requires approval)
result = agent.process_refund_request(
customer_id="cus_XYZ789",
order_id="ch_9876543210",
amount=500.00,
reason="not_as_described"
)
print(result) # {'status': 'pending_approval', ...}
Permission Setup
In the AgentWarden dashboard, create these permissions:Copy
Action: stripe.refund.small
Max Amount: $50
Requires Approval: No
Action: stripe.refund.medium
Max Amount: $200
Requires Approval: No
Action: stripe.refund.large
Max Amount: $1000
Requires Approval: Yes
DevOps Deployment Agent
Automated deployments with approval for production.Copy
import os
import subprocess
from agentwarden import AgentWarden
guard = AgentWarden(api_key=os.getenv('AGENTWARDEN_API_KEY'))
AGENT_ID = "devops-deployment-agent"
class DeploymentAgent:
"""Automated deployment with safety controls"""
def deploy(self, environment: str, version: str,
commit_hash: str) -> dict:
"""
Deploy to specified environment with permission check
"""
action = f"deploy.{environment}"
# Run tests first
tests_passed = self.run_tests()
# Check permission
result = guard.check(
agent_id=AGENT_ID,
action=action,
context={
"environment": environment,
"version": version,
"commit": commit_hash,
"tests_passed": tests_passed,
"deployed_by": "automation"
}
)
if not result.allowed:
if result.requires_approval:
self.notify_team(
f"🚀 Production deployment of {version} requires approval",
result.approval_id
)
return {"status": "pending", "approval_id": result.approval_id}
else:
return {"status": "denied", "reason": result.reason}
# Execute deployment
try:
self.execute_deployment(environment, version, commit_hash)
guard.log(
agent_id=AGENT_ID,
action=action,
status="success",
context={
"version": version,
"commit": commit_hash,
"tests_passed": tests_passed
}
)
return {"status": "deployed", "version": version}
except Exception as e:
guard.log(
agent_id=AGENT_ID,
action=action,
status="failed",
context={"error": str(e), "version": version}
)
raise
def run_tests(self) -> bool:
"""Run test suite"""
result = subprocess.run(['pytest'], capture_output=True)
return result.returncode == 0
def execute_deployment(self, env: str, version: str, commit: str):
"""Execute the actual deployment"""
print(f"Deploying {version} to {env}...")
# Your deployment logic here
pass
def notify_team(self, message: str, approval_id: str = None):
"""Send Slack/email notification"""
print(f"📢 {message}")
if approval_id:
print(f"Approve at: https://app.agentwarden.io/approvals/{approval_id}")
# Usage
deployer = DeploymentAgent()
# Staging - auto-deploys
deployer.deploy("staging", "v2.1.0", "a1b2c3d")
# Production - requires approval
deployer.deploy("production", "v2.1.0", "a1b2c3d")
Permission Setup
Copy
Action: deploy.staging
Requires Approval: No
Action: deploy.production
Requires Approval: Yes
Data Processing Agent
Automated data operations with safety controls.Copy
import pandas as pd
from agentwarden import AgentWarden
guard = AgentWarden(api_key=os.getenv('AGENTWARDEN_API_KEY'))
AGENT_ID = "data-processing-agent"
class DataAgent:
"""Data operations with permission controls"""
def delete_user_data(self, user_id: str, reason: str,
ticket_id: str = None) -> dict:
"""
Delete user data (GDPR compliance)
"""
# Check permission
result = guard.check(
agent_id=AGENT_ID,
action="database.users.delete",
context={
"user_id": user_id,
"reason": reason,
"ticket_id": ticket_id,
"gdpr_request": reason == "gdpr"
}
)
if not result.allowed:
if result.requires_approval:
return {
"status": "pending_approval",
"approval_id": result.approval_id
}
else:
return {"status": "denied", "reason": result.reason}
try:
# Execute deletion
self.execute_deletion(user_id)
guard.log(
agent_id=AGENT_ID,
action="database.users.delete",
status="success",
context={
"user_id": user_id,
"reason": reason,
"records_deleted": 1
}
)
return {"status": "deleted", "user_id": user_id}
except Exception as e:
guard.log(
agent_id=AGENT_ID,
action="database.users.delete",
status="failed",
context={"error": str(e), "user_id": user_id}
)
raise
def execute_deletion(self, user_id: str):
"""Execute the actual deletion"""
# Your database deletion logic
print(f"Deleting user {user_id} from database...")
pass
# Usage
agent = DataAgent()
# GDPR deletion - may auto-approve depending on permission
agent.delete_user_data(
user_id="usr_12345",
reason="gdpr",
ticket_id="GDPR-4567"
)
Best Practices Demonstrated
1. Always Check Before Acting
1. Always Check Before Acting
All examples check permissions BEFORE executing sensitive operations.
2. Provide Rich Context
2. Provide Rich Context
Context includes all relevant information for approval decisions.
3. Log Everything
3. Log Everything
Success, failures, and denials are all logged for audit trail.
4. Handle All Scenarios
4. Handle All Scenarios
Code handles allowed, requires_approval, and denied cases.
5. Structured Error Handling
5. Structured Error Handling
Errors are caught, logged, and reported with context.