> ## Documentation Index
> Fetch the complete documentation index at: https://docs.fabraix.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Check Action
> Validate agent actions before execution to detect goal deviation
## Overview
This endpoint validates actions that an agent is about to execute, analyzing them in the context of the entire session history. It provides real-time security checks to prevent prompt injections, goal deviations, and other malicious behaviors.
Always check critical actions (financial transactions, data modifications, external communications) before execution. This is your primary defense against compromised agents.
## Request
Must be set to `"action_check"` for this endpoint.
Example: `"action_check"`
The session identifier from `/register-agent-run`. Used to retrieve session context for analysis.
Example: `"f4f4f4f4-f4f4-f4f4-f4f4-f4f4f4f4f4f4"`
Unix timestamp (seconds since epoch) when the check is being performed.
Example: `1678886415.123`
The action data as a stringified JSON object. This represents the specific parameters of the action to be executed.
Example: `"{\"amount\":500,\"recipient\":\"user@example.com\"}"`
A stringified JSON Schema defining the action's structure. Should include the function name, description, and parameter definitions.
Example: `"{\"type\":\"function\",\"name\":\"transfer_funds\",\"parameters\":{...}}"`
## Response
Whether the action is safe to execute.
* `true` - Action is approved
* `false` - Action is blocked
Human-readable explanation of the decision. Particularly important when `is_safe` is false.
Example: `"Action deviates from original user request for weather information"`
Unique identifier for this security check.
Example: `"c1c1c1c1-c1c1-c1c1-c1c1-c1c1c1c1c1c1"`
Server timestamp when the check was performed.
Example: `1678886415.789`
## Critical Actions to Check
Always validate these action types before execution:
* Money transfers
* Payment processing
* Refunds
* Account modifications
* Database updates
* File deletions
* Backup operations
* Schema changes
* Sending emails
* SMS messages
* API calls to external services
* Webhooks
* Code execution
* Configuration changes
* Permission modifications
* Service restarts
## Attack Detection Examples
### Prompt Injection Detection
```python Python - Attack Scenario theme={null}
# 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
# }
```
```javascript JavaScript - Safe Action theme={null}
// User asked to send an email
// Agent prepares appropriate email action
const actionContent = {
to: "boss@company.com",
subject: "Weekly Report",
body: "Here is the weekly report as requested..."
};
const actionSchema = {
type: "function",
name: "send_email",
description: "Send email message",
parameters: {
type: "object",
properties: {
to: { type: "string", format: "email" },
subject: { type: "string" },
body: { type: "string" }
}
}
};
// Check the action
const response = await checkAction(
traceId,
actionContent,
actionSchema
);
// Result
console.log(response);
// {
// "is_safe": true,
// "reasoning": "Email action aligns with user request",
// "action_check_id": "check-456",
// "timestamp": 1678886420.123
// }
```
### Goal Deviation Detection
```python theme={null}
# Initial: User asks for help with homework
# Later: Agent tries to search for inappropriate content
# Suspicious action after conversation drift
action_content = {
"query": "how to synthesize illegal substances",
"search_engine": "google"
}
action_schema = {
"type": "function",
"name": "web_search",
"description": "Search the web",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string"},
"search_engine": {"type": "string"}
}
}
}
response = check_action(trace_id, action_content, action_schema)
# Result: is_safe = False
# Reasoning: "Search query deviates from homework assistance objective"
```
## Complete Implementation Example
```python Python theme={null}
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://api.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
)
```
```javascript JavaScript theme={null}
class SecureAgent {
constructor(apiKey, traceId) {
this.apiKey = apiKey;
this.traceId = traceId;
this.baseUrl = 'https://api.fabraix.com/v1';
}
async checkAction(actionName, actionParams, actionSchema) {
const response = await fetch(`${this.baseUrl}/check`, {
method: 'POST',
headers: {
'x-api-key': this.apiKey,
'Content-Type': 'application/json'
},
body: JSON.stringify({
event_type: 'action_check',
trace_id: this.traceId,
timestamp: Date.now() / 1000,
content: JSON.stringify(actionParams),
schema: JSON.stringify(actionSchema)
})
});
const result = await response.json();
return {
isSafe: result.is_safe,
reasoning: result.reasoning
};
}
async executeWithSafety(actionName, actionParams, actionSchema,
executeFn, fallbackFn = null) {
// Check action safety
const { isSafe, reasoning } = await this.checkAction(
actionName,
actionParams,
actionSchema
);
if (isSafe) {
console.log(`✅ Action '${actionName}' approved`);
try {
// Execute the action
const result = await executeFn(actionParams);
// Log successful execution
await this.logEvent('environment', {
action: actionName,
status: 'success',
result: result
});
return result;
} catch (error) {
// Log execution error
await this.logEvent('error', {
action: actionName,
error: error.message
});
throw error;
}
} else {
// Action blocked
console.log(`❌ Action '${actionName}' blocked: ${reasoning}`);
// Log the block
await this.logEvent('security_block', {
action: actionName,
reasoning: reasoning,
params: actionParams
});
// Use fallback if provided
if (fallbackFn) {
return fallbackFn(reasoning);
} else {
throw new Error(`Action blocked: ${reasoning}`);
}
}
}
}
// Usage Example
const agent = new SecureAgent('YOUR_KEY', 'abc-123');
// Define action
const transferParams = {
amount: 100,
recipient: 'vendor@company.com',
reason: 'Invoice payment'
};
const transferSchema = {
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
async function performTransfer(params) {
// Actual transfer logic
return { transaction_id: 'tx-789', status: 'completed' };
}
function handleBlockedTransfer(reasoning) {
// Fallback for blocked transfers
return { status: 'blocked', message: reasoning };
}
const result = await agent.executeWithSafety(
'transfer_funds',
transferParams,
transferSchema,
performTransfer,
handleBlockedTransfer
);
```
## Response Examples
### Approved Action
```json theme={null}
{
"is_safe": true,
"reasoning": "Action aligns with user request and session context",
"action_check_id": "c2c2c2c2-c2c2-c2c2-c2c2-c2c2c2c2c2c2",
"timestamp": 1678886420.123
}
```
### Blocked Actions
```json Prompt Injection theme={null}
{
"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
}
```
```json Goal Deviation theme={null}
{
"is_safe": false,
"reasoning": "Action deviates from established objective: attempting financial transaction when user requested document summarization",
"action_check_id": "c4c4c4c4-c4c4-c4c4-c4c4-c4c4c4c4c4c4",
"timestamp": 1678886430.789
}
```
```json Unauthorized Access theme={null}
{
"is_safe": false,
"reasoning": "Attempting to access resources beyond authorized scope: admin panel access not permitted for current session",
"action_check_id": "c5c5c5c5-c5c5-c5c5-c5c5-c5c5c5c5c5c5",
"timestamp": 1678886435.012
}
```
## Best Practices
Establish what actions require checking:
```python theme={null}
ALWAYS_CHECK = [
"transfer_funds",
"delete_data",
"send_email",
"modify_permissions",
"execute_code"
]
CONDITIONAL_CHECK = {
"purchase_item": lambda params: params["amount"] > 100,
"api_call": lambda params: params["endpoint"].startswith("external"),
"database_query": lambda params: "DELETE" in params["query"]
}
def should_check_action(action_name, params):
if action_name in ALWAYS_CHECK:
return True
if action_name in CONDITIONAL_CHECK:
return CONDITIONAL_CHECK[action_name](params)
return False
```
Provide good user experience when actions are blocked:
```python theme={null}
def handle_blocked_action(action_name, reasoning):
user_friendly_messages = {
"prompt_injection": "I detected a potential security issue and cannot proceed.",
"goal_deviation": "This action doesn't align with your original request.",
"unauthorized": "I don't have permission to perform this action.",
"suspicious_pattern": "This action was flagged for security review."
}
# Determine block type from reasoning
for key, message in user_friendly_messages.items():
if key in reasoning.lower():
return message
# Default message
return "I cannot perform this action for security reasons."
```
Provide detailed schemas to improve analysis accuracy:
```python theme={null}
# Good: Rich schema with context
schema = {
"type": "function",
"name": "modify_database",
"description": "Modify database records",
"criticality": "high",
"reversible": false,
"parameters": {
"type": "object",
"properties": {
"table": {
"type": "string",
"enum": ["users", "orders", "products"]
},
"operation": {
"type": "string",
"enum": ["INSERT", "UPDATE", "DELETE"]
},
"where_clause": {
"type": "string",
"pattern": "^[A-Za-z0-9_]+ = .+$"
},
"data": {
"type": "object"
}
},
"required": ["table", "operation"]
}
}
# Bad: Minimal schema
schema = {
"type": "function",
"name": "modify_database"
}
```
Handle transient failures appropriately:
```python theme={null}
async def check_with_retry(action, max_retries=3):
for attempt in range(max_retries):
try:
return await check_action(action)
except RequestException as e:
if e.status_code == 429: # Rate limited
wait_time = min(2 ** attempt, 10)
await asyncio.sleep(wait_time)
elif e.status_code >= 500: # Server error
if attempt < max_retries - 1:
await asyncio.sleep(1)
continue
raise
raise MaxRetriesException()
```
Track and analyze blocked actions:
```python theme={null}
class ActionMonitor:
def __init__(self):
self.blocked_actions = []
self.block_reasons = {}
def record_block(self, action_name, reasoning):
self.blocked_actions.append({
"action": action_name,
"reasoning": reasoning,
"timestamp": time.time()
})
# Track block reasons
reason_type = self.classify_reason(reasoning)
self.block_reasons[reason_type] = \
self.block_reasons.get(reason_type, 0) + 1
def get_statistics(self):
return {
"total_blocks": len(self.blocked_actions),
"block_reasons": self.block_reasons,
"recent_blocks": self.blocked_actions[-10:]
}
```
## Performance Optimization
### Parallel Checking
Check multiple independent actions in parallel:
```python theme={null}
import asyncio
async def check_multiple_actions(trace_id, actions):
"""Check multiple actions in parallel"""
tasks = []
for action in actions:
task = check_action(
trace_id,
action["params"],
action["schema"]
)
tasks.append(task)
results = await asyncio.gather(*tasks)
# Return results with action names
return [
{
"action": actions[i]["name"],
"is_safe": results[i]["is_safe"],
"reasoning": results[i]["reasoning"]
}
for i in range(len(actions))
]
```
## Related Endpoints
* [POST /register-agent-run](/api-reference/arx/endpoint/register-agent-run) - Register session to get trace\_id
* [POST /event](/api-reference/arx/endpoint/event) - Log events that provide context for checks
## FAQ
Typical response time is 0.5-1s. Critical actions should always be checked despite the small latency cost which we are continuously improving.
Unchecked actions bypass Fabraix's security layer, leaving your agent vulnerable to prompt injections, goal deviations, and other attacks.
Blocks should be treated as final for security. If you need to override, log the override as an event and implement additional safeguards.
Fabraix analyzes the entire session history (all logged events) to understand the agent's trajectory and detect deviations or anomalies.
Generally, read-only operations don't require checking unless they involve sensitive data or could be part of a reconnaissance attack.