# Check Action
Source: https://docs.fabraix.com/api-reference/arx/endpoint/check
POST /check
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.
# Log Event
Source: https://docs.fabraix.com/api-reference/arx/endpoint/event
POST /event
Log agent events for observability and security analysis
## Overview
This endpoint receives and stores individual events from AI agents throughout their execution. Events represent distinct steps in the agent's reasoning loop (user inputs, model outputs, tool calls, etc.) and are essential for security analysis and observability.
Events should be logged continuously throughout the agent's lifecycle to maintain a complete audit trail.
## Request
The category of event being logged. Must be one of:
* `user` - Input from human users
* `model_input` - Data sent to the LLM
* `model_output` - Responses from the LLM
* `tool` - Tool/function calls and results
* `environment` - External system interactions
* `memory` - Memory read/write operations
* `system` - System-level events
* `error` - Error conditions and failures
The session identifier obtained from `/register-agent-run`. This links the event to a specific agent session.
Example: `"f4f4f4f4-f4f4-f4f4-f4f4-f4f4f4f4f4f4"`
Unix timestamp (seconds since epoch) when the event occurred.
Example: `1678886405.123`
The event data as a stringified JSON object. Must conform to the structure defined in the `schema` field.
Example: `"{\"location\":\"London, UK\",\"units\":\"celsius\"}"`
A stringified JSON Schema object defining the structure of the `content` field. This enables dynamic validation and understanding of diverse event types.
Example: `"{\"type\":\"object\",\"properties\":{\"location\":{\"type\":\"string\"},\"units\":{\"type\":\"string\"}}}"`
## Response
Unique identifier for the logged event.
Example: `"e1e1e1e1-e1e1-e1e1-e1e1-e1e1e1e1e1e1"`
Server timestamp when the event was processed.
Example: `1678886405.456`
## Event Type Examples
### User Event
Log user inputs to the agent:
```python Python theme={null}
import json
import time
# User message event
user_content = {
"message": "Can you help me book a flight to Paris?",
"user_id": "user-123",
"channel": "web_chat"
}
user_schema = {
"type": "object",
"properties": {
"message": {
"type": "string",
"description": "User's message text"
},
"user_id": {
"type": "string",
"description": "Unique user identifier"
},
"channel": {
"type": "string",
"enum": ["web_chat", "mobile", "api"]
}
},
"required": ["message", "user_id"]
}
response = log_event(
event_type="user",
trace_id=trace_id,
timestamp=time.time(),
content=json.dumps(user_content),
schema=json.dumps(user_schema)
)
```
```javascript JavaScript theme={null}
// User message event
const userContent = {
message: "Can you help me book a flight to Paris?",
user_id: "user-123",
channel: "web_chat"
};
const userSchema = {
type: "object",
properties: {
message: {
type: "string",
description: "User's message text"
},
user_id: {
type: "string",
description: "Unique user identifier"
},
channel: {
type: "string",
enum: ["web_chat", "mobile", "api"]
}
},
required: ["message", "user_id"]
};
const response = await logEvent({
event_type: "user",
trace_id: traceId,
timestamp: Date.now() / 1000,
content: JSON.stringify(userContent),
schema: JSON.stringify(userSchema)
});
```
### Tool Event
Log tool/function calls:
```python Python theme={null}
# Tool call event
tool_content = {
"location": "Paris, France",
"check_in": "2024-03-15",
"check_out": "2024-03-20",
"guests": 2
}
tool_schema = {
"type": "function",
"name": "search_hotels",
"description": "Search for available hotels",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "Destination city"
},
"check_in": {
"type": "string",
"format": "date"
},
"check_out": {
"type": "string",
"format": "date"
},
"guests": {
"type": "integer",
"minimum": 1,
"maximum": 10
}
},
"required": ["location", "check_in", "check_out"]
}
}
response = log_event(
event_type="tool",
trace_id=trace_id,
timestamp=time.time(),
content=json.dumps(tool_content),
schema=json.dumps(tool_schema)
)
```
```javascript JavaScript theme={null}
// Tool call event
const toolContent = {
location: "Paris, France",
check_in: "2024-03-15",
check_out: "2024-03-20",
guests: 2
};
const toolSchema = {
type: "function",
name: "search_hotels",
description: "Search for available hotels",
parameters: {
type: "object",
properties: {
location: {
type: "string",
description: "Destination city"
},
check_in: {
type: "string",
format: "date"
},
check_out: {
type: "string",
format: "date"
},
guests: {
type: "integer",
minimum: 1,
maximum: 10
}
},
required: ["location", "check_in", "check_out"]
}
};
const response = await logEvent({
event_type: "tool",
trace_id: traceId,
timestamp: Date.now() / 1000,
content: JSON.stringify(toolContent),
schema: JSON.stringify(toolSchema)
});
```
### Model Output Event
Log LLM responses:
```python theme={null}
# Model output event
model_content = {
"response": "I'll help you search for hotels in Paris.",
"tool_calls": [
{
"id": "call_123",
"function": "search_hotels",
"arguments": {
"location": "Paris, France"
}
}
],
"confidence": 0.95,
"tokens_used": 45
}
model_schema = {
"type": "object",
"properties": {
"response": {"type": "string"},
"tool_calls": {
"type": "array",
"items": {
"type": "object",
"properties": {
"id": {"type": "string"},
"function": {"type": "string"},
"arguments": {"type": "object"}
}
}
},
"confidence": {"type": "number"},
"tokens_used": {"type": "integer"}
}
}
```
### Memory Event
Log memory operations:
```python theme={null}
# Memory write event
memory_content = {
"operation": "write",
"key": "user_preferences",
"value": {
"preferred_airline": "Air France",
"seat_preference": "window",
"meal_preference": "vegetarian"
},
"ttl": 86400 # 24 hours
}
memory_schema = {
"type": "object",
"properties": {
"operation": {
"type": "string",
"enum": ["read", "write", "delete", "update"]
},
"key": {"type": "string"},
"value": {"type": "object"},
"ttl": {
"type": "integer",
"description": "Time to live in seconds"
}
},
"required": ["operation", "key"]
}
```
## Complete Example
Here's a complete example of logging multiple events in sequence:
```python Python theme={null}
import json
import time
import requests
class FabraixLogger:
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 log_event(self, event_type, content, schema):
"""Log an event to Fabraix"""
response = requests.post(
f"{self.base_url}/event",
headers={
"x-api-key": self.api_key,
"Content-Type": "application/json"
},
json={
"event_type": event_type,
"trace_id": self.trace_id,
"timestamp": time.time(),
"content": json.dumps(content),
"schema": json.dumps(schema)
}
)
return response.json()
def log_user_message(self, message, user_id):
"""Log a user message"""
return self.log_event(
event_type="user",
content={
"message": message,
"user_id": user_id
},
schema={
"type": "object",
"properties": {
"message": {"type": "string"},
"user_id": {"type": "string"}
}
}
)
def log_tool_call(self, tool_name, arguments, result=None):
"""Log a tool call"""
content = {
"tool": tool_name,
"arguments": arguments
}
if result:
content["result"] = result
return self.log_event(
event_type="tool",
content=content,
schema={
"type": "object",
"properties": {
"tool": {"type": "string"},
"arguments": {"type": "object"},
"result": {"type": "object"}
}
}
)
# Usage
logger = FabraixLogger(api_key="YOUR_KEY", trace_id="abc-123")
# Log conversation flow
logger.log_user_message("Book a flight to Paris", "user-456")
logger.log_tool_call("search_flights", {"destination": "Paris"})
logger.log_tool_call(
"search_flights",
{"destination": "Paris"},
result={"flights": [{"id": "FL123", "price": 299}]}
)
```
```javascript JavaScript theme={null}
class FabraixLogger {
constructor(apiKey, traceId) {
this.apiKey = apiKey;
this.traceId = traceId;
this.baseUrl = 'https://api.fabraix.com/v1';
}
async logEvent(eventType, content, schema) {
const response = await fetch(`${this.baseUrl}/event`, {
method: 'POST',
headers: {
'x-api-key': this.apiKey,
'Content-Type': 'application/json'
},
body: JSON.stringify({
event_type: eventType,
trace_id: this.traceId,
timestamp: Date.now() / 1000,
content: JSON.stringify(content),
schema: JSON.stringify(schema)
})
});
return response.json();
}
async logUserMessage(message, userId) {
return this.logEvent(
'user',
{
message: message,
user_id: userId
},
{
type: 'object',
properties: {
message: { type: 'string' },
user_id: { type: 'string' }
}
}
);
}
async logToolCall(toolName, args, result = null) {
const content = {
tool: toolName,
arguments: args
};
if (result) {
content.result = result;
}
return this.logEvent(
'tool',
content,
{
type: 'object',
properties: {
tool: { type: 'string' },
arguments: { type: 'object' },
result: { type: 'object' }
}
}
);
}
}
// Usage
const logger = new FabraixLogger('YOUR_KEY', 'abc-123');
// Log conversation flow
await logger.logUserMessage('Book a flight to Paris', 'user-456');
await logger.logToolCall('search_flights', { destination: 'Paris' });
await logger.logToolCall(
'search_flights',
{ destination: 'Paris' },
{ flights: [{ id: 'FL123', price: 299 }] }
);
```
## Response Examples
### Success Response
```json theme={null}
{
"event_id": "e1e1e1e1-e1e1-e1e1-e1e1-e1e1e1e1e1e1",
"timestamp": 1678886405.789
}
```
### Error Responses
```json 400 Bad Request - Invalid Schema theme={null}
{
"error": {
"message": "Content does not match provided schema",
"type": "validation_error",
"code": "SCHEMA_MISMATCH",
"details": {
"missing_field": "user_id",
"path": "$.user_id"
}
}
}
```
```json 404 Not Found - Invalid Trace theme={null}
{
"error": {
"message": "Trace ID not found",
"type": "not_found_error",
"code": "TRACE_NOT_FOUND"
}
}
```
```json 400 Bad Request - Invalid JSON theme={null}
{
"error": {
"message": "Invalid JSON in content field",
"type": "parse_error",
"code": "INVALID_JSON",
"details": {
"position": 45,
"line": 2
}
}
}
```
## Best Practices
Log events as they occur rather than batching (unless using batch API):
```python theme={null}
# Good: Immediate logging
user_input = get_user_input()
log_event("user", user_input) # Log immediately
response = process_input(user_input)
log_event("model_output", response) # Log immediately
# Bad: Delayed logging
events = []
events.append(("user", user_input))
# ... much later ...
for event in events:
log_event(*event) # Context lost
```
Add contextual information that aids in debugging and analysis:
```python theme={null}
# Good: Rich context
content = {
"action": "delete_file",
"file_path": "/data/report.pdf",
"file_size": 1024000,
"reason": "User requested deletion",
"user_id": "user-123",
"ip_address": "192.168.1.1",
"session_id": "sess-456"
}
# Bad: Minimal context
content = {
"action": "delete_file",
"file": "report.pdf"
}
```
Provide comprehensive schemas with constraints and descriptions:
```python theme={null}
# Good: Detailed schema
schema = {
"type": "object",
"description": "Email sending event",
"properties": {
"to": {
"type": "string",
"format": "email",
"description": "Recipient email"
},
"subject": {
"type": "string",
"maxLength": 200,
"description": "Email subject line"
},
"priority": {
"type": "string",
"enum": ["low", "normal", "high"],
"default": "normal"
}
},
"required": ["to", "subject"],
"additionalProperties": false
}
# Bad: Minimal schema
schema = {"type": "object"}
```
Log errors as events for complete observability:
```python theme={null}
try:
result = risky_operation()
log_event("tool", {"result": result})
except Exception as e:
# Log the error as an event
log_event(
event_type="error",
content={
"error": str(e),
"error_type": e.__class__.__name__,
"operation": "risky_operation",
"traceback": traceback.format_exc()
},
schema={
"type": "object",
"properties": {
"error": {"type": "string"},
"error_type": {"type": "string"},
"operation": {"type": "string"},
"traceback": {"type": "string"}
}
}
)
```
Use async logging to minimize latency:
```python theme={null}
import asyncio
from concurrent.futures import ThreadPoolExecutor
class AsyncLogger:
def __init__(self, api_key, trace_id):
self.api_key = api_key
self.trace_id = trace_id
self.executor = ThreadPoolExecutor(max_workers=5)
async def log_event_async(self, event_type, content, schema):
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
self.executor,
self._log_event_sync,
event_type,
content,
schema
)
def _log_event_sync(self, event_type, content, schema):
# Synchronous API call
return log_event(
self.trace_id,
event_type,
content,
schema
)
```
## Performance Considerations
### Batch Logging
For high-volume applications, consider batching events:
```python theme={null}
class EventBatcher:
def __init__(self, api_key, trace_id, batch_size=50):
self.api_key = api_key
self.trace_id = trace_id
self.batch = []
self.batch_size = batch_size
def add_event(self, event_type, content, schema):
self.batch.append({
"event_type": event_type,
"trace_id": self.trace_id,
"timestamp": time.time(),
"content": json.dumps(content),
"schema": json.dumps(schema)
})
if len(self.batch) >= self.batch_size:
self.flush()
def flush(self):
if self.batch:
# Send batch to API (future feature)
send_batch(self.batch)
self.batch = []
```
### Content Size Limits
Be mindful of content size:
* Maximum content size: 1MB
* Maximum schema size: 64KB
* For large data, consider storing externally and logging references
```python theme={null}
# For large content, store externally
if len(content_str) > 900000: # Near 1MB limit
# Store in S3/database
reference_id = store_external(content_str)
# Log reference instead
log_event(
event_type="environment",
content={
"type": "external_reference",
"reference_id": reference_id,
"size_bytes": len(content_str)
},
schema=reference_schema
)
```
## Related Endpoints
* [POST /register-agent-run](/api-reference/arx/endpoint/register-agent-run) - Register a session before logging events
* [POST /check](/api-reference/arx/endpoint/check) - Validate actions based on logged events
## FAQ
Yes, events can arrive out of order. The timestamp field is used to establish the correct sequence. However, logging events as they occur is recommended for real-time analysis.
The event will be rejected with a 400 error detailing the validation failure. Fix the content to match the schema or update the schema to match the content.
Schemas should be as detailed as possible. Include types, constraints, enums, and descriptions. This helps Fabraix better understand your agent's behavior.
There's no hard limit on events per session, but extremely long sessions (>10,000 events) may experience degraded performance. Consider creating new sessions for long-running agents periodically.
# Register Agent Run
Source: https://docs.fabraix.com/api-reference/arx/endpoint/register-agent-run
POST /register-agent-run
Initialize a new agent session and receive a trace_id
## Overview
This endpoint registers the start of a new agent interaction or "run". It creates a new session with a unique `trace_id` that will be used to correlate all subsequent events for this specific agent session.
Always call this endpoint at the beginning of each new agent conversation or task to establish a tracking context.
## Request
The unique identifier for your agent. This should be consistent across all runs of the same agent.
Example: `"a1b2c3d4-e5f6-7890-1234-567890abcdef"`
The timestamp when the agent run is starting, in ISO 8601 format.
Format: `YYYY-MM-DDTHH:mm:ss.sssZ`
Example: `"2024-01-15T14:30:45.123Z"`
The system prompt or initial instructions given to the agent for this session. This helps establish the agent's intended behavior for security analysis.
Example: `"You are a helpful customer service assistant. You should be polite and professional. Never share customer personal information."`
## Response
The unique identifier for this agent session. Use this ID for all subsequent event logging and action checking within this session.
Example: `"f4f4f4f4-f4f4-f4f4-f4f4-f4f4f4f4f4f4"`
The server timestamp when the session was registered, in ISO 8601 format.
Example: `"2024-01-15T14:30:45.456Z"`
## Examples
```bash cURL theme={null}
curl -X POST https://api.fabraix.com/v1/register-agent-run \
-H "x-api-key: YOUR_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"agent_id": "a1b2c3d4-e5f6-7890-1234-567890abcdef",
"timestamp": "2024-01-15T14:30:45.123Z",
"system_prompt": "You are a helpful customer service assistant. You should be polite and professional."
}'
```
```python Python theme={null}
import requests
import uuid
from datetime import datetime
# Configuration
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://api.fabraix.com/v1"
# Create request
agent_id = uuid.uuid4()
request_data = {
"agent_id": str(agent_id),
"timestamp": datetime.now().isoformat(),
"system_prompt": "You are a helpful customer service assistant."
}
# Make API call
response = requests.post(
f"{BASE_URL}/register-agent-run",
headers={
"x-api-key": API_KEY,
"Content-Type": "application/json"
},
json=request_data
)
# Extract trace_id
result = response.json()
trace_id = result["trace_id"]
print(f"Session started with trace_id: {trace_id}")
```
```javascript JavaScript theme={null}
const axios = require('axios');
const { v4: uuidv4 } = require('uuid');
// Configuration
const API_KEY = 'YOUR_API_KEY';
const BASE_URL = 'https://api.fabraix.com/v1';
// Create request
const agentId = uuidv4();
const requestData = {
agent_id: agentId,
timestamp: new Date().toISOString(),
system_prompt: 'You are a helpful customer service assistant.'
};
// Make API call
async function registerAgentRun() {
try {
const response = await axios.post(
`${BASE_URL}/register-agent-run`,
requestData,
{
headers: {
'x-api-key': API_KEY,
'Content-Type': 'application/json'
}
}
);
const traceId = response.data.trace_id;
console.log(`Session started with trace_id: ${traceId}`);
return traceId;
} catch (error) {
console.error('Error registering agent run:', error);
throw error;
}
}
registerAgentRun();
```
```go Go theme={null}
package main
import (
"bytes"
"encoding/json"
"fmt"
"net/http"
"time"
"github.com/google/uuid"
)
type RegisterRequest struct {
AgentID string `json:"agent_id"`
Timestamp time.Time `json:"timestamp"`
SystemPrompt string `json:"system_prompt"`
}
type RegisterResponse struct {
TraceID string `json:"trace_id"`
Timestamp time.Time `json:"timestamp"`
}
func registerAgentRun(apiKey string) (string, error) {
url := "https://api.fabraix.com/v1/register-agent-run"
// Create request
agentID := uuid.New().String()
reqData := RegisterRequest{
AgentID: agentID,
Timestamp: time.Now(),
SystemPrompt: "You are a helpful customer service assistant.",
}
jsonData, err := json.Marshal(reqData)
if err != nil {
return "", err
}
// Make HTTP request
req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonData))
if err != nil {
return "", err
}
req.Header.Set("x-api-key", apiKey)
req.Header.Set("Content-Type", "application/json")
client := &http.Client{}
resp, err := client.Do(req)
if err != nil {
return "", err
}
defer resp.Body.Close()
// Parse response
var result RegisterResponse
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return "", err
}
fmt.Printf("Session started with trace_id: %s\n", result.TraceID)
return result.TraceID, nil
}
```
## Response Examples
### Success Response
```json theme={null}
{
"trace_id": "f4f4f4f4-f4f4-f4f4-f4f4-f4f4f4f4f4f4",
"timestamp": "2024-01-15T14:30:45.456Z"
}
```
### Error Responses
```json 400 Bad Request theme={null}
{
"error": {
"message": "Invalid agent_id format",
"type": "validation_error",
"code": "INVALID_AGENT_ID"
}
}
```
```json 401 Unauthorized theme={null}
{
"error": {
"message": "Invalid API key provided",
"type": "authentication_error",
"code": "INVALID_API_KEY"
}
}
```
```json 429 Rate Limited theme={null}
{
"error": {
"message": "Rate limit exceeded",
"type": "rate_limit_error",
"retry_after": 30
}
}
```
## Best Practices
Use the same `agent_id` for all runs of the same agent type. This helps with:
* Analytics and monitoring
* Identifying patterns across sessions
* Debugging and troubleshooting
```python theme={null}
# Good: Consistent agent ID
CUSTOMER_SERVICE_AGENT_ID = "cs-agent-prod-v1"
# Bad: Random ID each time
agent_id = str(uuid.uuid4()) # Different every run
```
Include clear constraints and objectives in your system prompt:
```python theme={null}
# Good: Clear boundaries
system_prompt = """
You are a customer service assistant for AcmeCorp.
- Help customers with product inquiries and orders
- Never share customer personal information
- Do not process refunds over $500 without manager approval
- Always be professional and courteous
"""
# Bad: Vague prompt
system_prompt = "You are helpful"
```
Store the `trace_id` properly for the entire session:
```python theme={null}
class AgentSession:
def __init__(self, agent_id, system_prompt):
self.agent_id = agent_id
self.trace_id = self.register_run(system_prompt)
self.event_count = 0
def register_run(self, system_prompt):
response = fabraix.register_agent_run(
agent_id=self.agent_id,
timestamp=datetime.now(),
system_prompt=system_prompt
)
return response["trace_id"]
def log_event(self, event_type, content, schema):
# Use stored trace_id for all events
return fabraix.log_event(
trace_id=self.trace_id,
event_type=event_type,
content=content,
schema=schema
)
```
Implement proper retry logic for transient failures:
```python theme={null}
import time
import random
def register_with_retry(agent_id, system_prompt, max_retries=3):
for attempt in range(max_retries):
try:
return register_agent_run(agent_id, system_prompt)
except HTTPError as e:
if e.response.status_code == 429:
# Rate limited - use exponential backoff
wait_time = (2 ** attempt) + random.uniform(0, 1)
time.sleep(wait_time)
elif e.response.status_code >= 500:
# Server error - retry
if attempt < max_retries - 1:
time.sleep(1)
continue
raise
raise Exception("Max retries exceeded")
```
## Use Cases
### Multi-Turn Conversation
```python theme={null}
# Start a conversation
trace_id = register_agent_run(
agent_id="chat-agent-v1",
timestamp=datetime.now(),
system_prompt="You are a helpful assistant"
)
# Use same trace_id for entire conversation
for message in conversation:
log_event(trace_id, "user", message)
response = process_message(message)
log_event(trace_id, "model_output", response)
```
### Task-Based Agent
```python theme={null}
# Start a task
trace_id = register_agent_run(
agent_id="task-agent-v1",
timestamp=datetime.now(),
system_prompt="Process customer orders"
)
# Use trace_id throughout task execution
for order in orders_to_process:
log_event(trace_id, "environment", order)
result = process_order(order)
log_event(trace_id, "tool", result)
```
### Autonomous Agent
```python theme={null}
# Start autonomous operation
trace_id = register_agent_run(
agent_id="auto-agent-v1",
timestamp=datetime.now(),
system_prompt="Monitor system health and respond to issues"
)
# Long-running agent with same trace_id
while running:
event = wait_for_event()
log_event(trace_id, "environment", event)
if requires_action(event):
action = determine_action(event)
if check_action(trace_id, action):
execute_action(action)
```
## Related Endpoints
* [POST /event](/api-reference/arx/endpoint/event) - Log events after registering a run
* [POST /check](/api-reference/arx/endpoint/check) - Validate actions using the trace\_id
## FAQ
Create a new trace\_id for:
* Each new conversation with a user
* Each distinct task or job
* When the agent's context is reset
* After a significant time gap (e.g., new day)
No, trace\_ids should be unique for each session. Reusing them will mix events from different sessions and compromise security analysis.
Events and checks submitted without a valid trace\_id will be rejected with a 404 error. Always register a run first.
Trace\_ids remain valid indefinitely for historical analysis, but should not be reused for new sessions after the original session ends.
# Arx API Reference
Source: https://docs.fabraix.com/api-reference/arx/introduction
Complete reference for the Arx run-time defence REST API
## Overview
Arx is Fabraix's run-time defence layer for AI agents, informed by what our offensive agent [Nyx](/api-reference/nyx/introduction) finds in the wild. The premise is simple: you can't verify what you haven't tried to break, and the same research that makes Nyx effective on offence is what makes Arx effective on defence.
Arx provides three primary capabilities:
* **Session tracking**: register agent runs and correlate every step under a `trace_id`
* **Event logging**: record user inputs, model outputs, tool calls, memory ops, and environment events
* **Action checks**: validate actions in the context of the full session before they execute
## Base URL
All API requests should be made to:
```
https://api.fabraix.com/v1
```
## Authentication
All requests must include your API key in the `x-api-key` header:
```bash theme={null}
curl -H "x-api-key: YOUR_API_KEY" https://api.fabraix.com/v1/endpoint
```
See the [Authentication guide](/essentials/authentication) for details.
## Available Endpoints
Register a new agent session and receive a trace\_id for tracking all subsequent events
Log events throughout your agent's reasoning loop for observability and analysis
Validate actions before execution to prevent malicious or unintended behavior
## Request Format
All POST requests should use JSON format with `Content-Type: application/json`:
```json theme={null}
{
"field1": "value1",
"field2": "value2"
}
```
## Response Format
All successful responses return JSON with appropriate HTTP status codes:
```json theme={null}
{
"result_field": "value",
"timestamp": "2024-01-01T12:00:00Z"
}
```
## Error Handling
Errors return appropriate HTTP status codes with detailed error messages:
```json theme={null}
{
"error": {
"message": "Detailed error description",
"type": "error_type",
"code": "ERROR_CODE"
}
}
```
### Common Status Codes
| Status | Description |
| ------ | ---------------------------------------- |
| 200 | Success |
| 400 | Bad Request - Invalid parameters |
| 401 | Unauthorized - Invalid API key |
| 403 | Forbidden - Insufficient permissions |
| 404 | Not Found - Invalid endpoint or resource |
| 429 | Too Many Requests - Rate limit exceeded |
| 500 | Internal Server Error |
## Rate Limits
API rate limits depend on your API key type:
| Key Type | Requests/Minute | Requests/Hour |
| ----------- | --------------- | ------------- |
| Development | 60 | 1,000 |
| Production | 600 | 10,000 |
| Enterprise | Custom | Custom |
## SDK Support
Arx is currently consumed via the REST API directly. Official Arx SDKs are coming soon.
Looking for the **Nyx** CLI? See the [Nyx API reference](/api-reference/nyx/introduction). The `@fabraix/nyx` npm package is generally available today.
## Pagination
For endpoints that return lists (future releases), pagination is handled via query parameters:
* `limit`: Number of items to return (max 100)
* `offset`: Number of items to skip
* `cursor`: Cursor for cursor-based pagination
Example:
```
GET /v1/events?limit=20&offset=40
```
## Versioning
The API version is included in the URL path. The current version is `v1`.
Breaking changes will result in a new API version. Non-breaking changes may be added to the current version.
## Webhooks (Coming Soon)
Future releases will support webhooks for real-time notifications:
* Action blocked events
* Anomaly detection alerts
* Session completion notifications
## Support
Need help with the API?
* 📧 Email: [founders@fabraix.com](mailto:founders@fabraix.com)
* 📖 GitHub: [Report API issues](https://github.com/fabraix/api/issues)
* 💬 Discord: [Join our developer community](https://discord.gg/n4scEY9NF6)
## Quick Start Examples
```bash theme={null}
curl -X POST https://api.fabraix.com/v1/register-agent-run \
-H "x-api-key: YOUR_KEY" \
-H "Content-Type: application/json" \
-d '{"agent_id": "123", "timestamp": "2024-01-01T00:00:00Z", "system_prompt": "Helper"}'
```
```bash theme={null}
curl -X POST https://api.fabraix.com/v1/event \
-H "x-api-key: YOUR_KEY" \
-H "Content-Type: application/json" \
-d '{"trace_id": "abc", "type": "user", "content": "{}", "schema": "{}"}'
```
```bash theme={null}
curl -X POST https://api.fabraix.com/v1/check \
-H "x-api-key: YOUR_KEY" \
-H "Content-Type: application/json" \
-d '{"trace_id": "abc", "content": "{}", "schema": "{}"}'
```
```python theme={null}
try:
response = client.check_action(...)
except RateLimitError as e:
time.sleep(e.retry_after)
response = client.check_action(...)
```
# Get Report
Source: https://docs.fabraix.com/api-reference/nyx/endpoint/get-report
GET /nyx/runs/{run_id}/report
Download the full markdown audit report for a completed Nyx run
## Overview
Returns the full audit report as markdown. The report includes findings, reproduction steps, and the attack chain Nyx used to surface each vulnerability.
Only available once a run reaches `status: "completed"`. Calling it earlier returns **425 Too Early**.
## Path Parameters
The `run_id` returned by `POST /nyx/runs`.
## Headers
Use `text/markdown` to receive the report body. Defaults to `text/markdown` if omitted.
## Response
A `text/markdown` body containing the audit report. The CLI writes it to `