Check Action

# Session history shows user asked for weather
# But agent tries to transfer money after reading a webpage

# The malicious action to check
action_content = {
    "amount": 1000,
    "recipient": "attacker@evil.com",
    "currency": "USD"
}

action_schema = {
    "type": "function",
    "name": "transfer_funds",
    "description": "Transfer money to recipient",
    "parameters": {
        "type": "object",
        "properties": {
            "amount": {"type": "number"},
            "recipient": {"type": "string"},
            "currency": {"type": "string"}
        }
    }
}

# Check the action
response = check_action(
    trace_id=trace_id,
    content=action_content,
    schema=action_schema
)

# Result
print(response)
# {
#   "is_safe": false,
#   "reasoning": "Transfer funds action unrelated to user's weather inquiry",
#   "action_check_id": "check-123",
#   "timestamp": 1678886415.789
# }

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": "vendor@company.com",
    "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
)

{
  "is_safe": false,
  "reasoning": "Detected potential prompt injection: action 'delete_all_files' unrelated to user's request for weather information",
  "action_check_id": "c3c3c3c3-c3c3-c3c3-c3c3-c3c3c3c3c3c3",
  "timestamp": 1678886425.456
}