import json
import time
import requests
class SecureAgent:
def __init__(self, api_key, trace_id):
self.api_key = api_key
self.trace_id = trace_id
self.base_url = "https://dev.fabraix.com/v1"
def check_action(self, action_name, action_params, action_schema):
"""Check if an action is safe before execution"""
response = requests.post(
f"{self.base_url}/check",
headers={
"x-api-key": self.api_key,
"Content-Type": "application/json"
},
json={
"event_type": "action_check",
"trace_id": self.trace_id,
"timestamp": time.time(),
"content": json.dumps(action_params),
"schema": json.dumps(action_schema)
}
)
result = response.json()
return result["is_safe"], result["reasoning"]
def execute_with_safety(self, action_name, action_params, action_schema,
execute_fn, fallback_fn=None):
"""Execute an action only if it passes safety checks"""
# Check action safety
is_safe, reasoning = self.check_action(
action_name,
action_params,
action_schema
)
if is_safe:
# Log approval
print(f"✅ Action '{action_name}' approved")
# Execute the action
try:
result = execute_fn(action_params)
# Log successful execution
self.log_event("environment", {
"action": action_name,
"status": "success",
"result": result
})
return result
except Exception as e:
# Log execution error
self.log_event("error", {
"action": action_name,
"error": str(e)
})
raise
else:
# Action blocked
print(f"❌ Action '{action_name}' blocked: {reasoning}")
# Log the block
self.log_event("security_block", {
"action": action_name,
"reasoning": reasoning,
"params": action_params
})
# Use fallback if provided
if fallback_fn:
return fallback_fn(reasoning)
else:
raise SecurityException(f"Action blocked: {reasoning}")
# Usage Example
agent = SecureAgent(api_key="YOUR_KEY", trace_id="abc-123")
# Define action
transfer_params = {
"amount": 100,
"recipient": "[email protected]",
"reason": "Invoice payment"
}
transfer_schema = {
"type": "function",
"name": "transfer_funds",
"description": "Transfer funds to recipient",
"parameters": {
"type": "object",
"properties": {
"amount": {
"type": "number",
"minimum": 0,
"maximum": 10000
},
"recipient": {
"type": "string",
"format": "email"
},
"reason": {
"type": "string"
}
},
"required": ["amount", "recipient"]
}
}
# Execute with safety check
def perform_transfer(params):
# Actual transfer logic
return {"transaction_id": "tx-789", "status": "completed"}
def handle_blocked_transfer(reasoning):
# Fallback for blocked transfers
return {"status": "blocked", "message": reasoning}
result = agent.execute_with_safety(
action_name="transfer_funds",
action_params=transfer_params,
action_schema=transfer_schema,
execute_fn=perform_transfer,
fallback_fn=handle_blocked_transfer
)