With AI agents going rogue at major companies, there's never been a better time to add safety guardrails to your AI-powered applications. In this tutorial, you'll learn how to build a practical guardrail system that monitors, filters, and controls AI agent actions in real time.
We'll build this using Python and the OpenAI-compatible API format — the same format used by Qubax AI and most modern AI providers.
Prerequisites
- Python 3.10+
- An API key from Qubax AI or any OpenAI-compatible provider
- Basic familiarity with Python async programming
Step 1: Set Up Your Environment
First, install the required packages:
pip install openai pydantic structlogCreate a new project directory and set up your environment variables:
export AI_API_KEY="your-api-key-here"
export AI_BASE_URL="https://api.qubax.ai/v1"Step 2: Define Your Safety Policy
Before writing any code, you need a clear safety policy. This defines what actions are allowed, what requires human approval, and what's blocked outright.
from enum import Enum
from pydantic import BaseModel
from typing import Optional
class RiskLevel(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
BLOCKED = "blocked"
class SafetyRule(BaseModel):
name: str
description: str
risk_level: RiskLevel
pattern: str # regex or keyword pattern to match
action: str # "allow", "require_approval", "block"
# Define your safety rules
SAFETY_RULES = [
SafetyRule(
name="no_file_deletion",
description="Block attempts to delete files",
risk_level=RiskLevel.BLOCKED,
pattern=r"(rm\s+-rf|del\s+/[fqs]|os\.remove|shutil\.rmtree)",
action="block"
),
SafetyRule(
name="no_network_exfiltration",
description="Block data exfiltration attempts",
risk_level=RiskLevel.BLOCKED,
pattern=r"(curl\s+.*\|\s*bash|wget.*--post-data|requests\.post.*api\.telegram)",
action="block"
),
SafetyRule(
name="production_database",
description="Require approval for production DB access",
risk_level=RiskLevel.HIGH,
pattern=r"(prod|production|live).*\.(db|sql|database)",
action="require_approval"
),
SafetyRule(
name="payment_actions",
description="Require approval for payment-related actions",
risk_level=RiskLevel.HIGH,
pattern=r"(charge|payment|refund|transfer|stripe|paypal)",
action="require_approval"
),
SafetyRule(
name="email_sending",
description="Monitor email sending actions",
risk_level=RiskLevel.MEDIUM,
pattern=r"(send_email|smtp|mail\.send|resend\.emails)",
action="log"
),
]Step 3: Build the Guardrail Engine
The guardrail engine inspects every action the AI agent wants to take and decides whether to allow, block, or require approval.
import re
import logging
from dataclasses import dataclass
from datetime import datetime, timezone
logger = logging.getLogger(__name__)
@dataclass
class GuardrailResult:
allowed: bool
risk_level: RiskLevel
matched_rules: list
requires_approval: bool
reason: str
timestamp: str
class GuardrailEngine:
def __init__(self, rules: list):
self.rules = rules
self.action_log = []
self.pending_approvals = {}
def evaluate(self, action: str, context: dict = None) -> GuardrailResult:
'''Evaluate an agent action against safety rules.'''
context = context or {}
matched = []
highest_risk = RiskLevel.LOW
should_block = False
should_require_approval = False
for rule in self.rules:
if re.search(rule.pattern, action, re.IGNORECASE):
matched.append(rule.name)
if rule.action == "block":
should_block = True
if rule.risk_level == RiskLevel.BLOCKED:
highest_risk = RiskLevel.BLOCKED
elif rule.action == "require_approval":
should_require_approval = True
if highest_risk != RiskLevel.BLOCKED:
highest_risk = RiskLevel.HIGH
result = GuardrailResult(
allowed=not should_block,
risk_level=highest_risk,
matched_rules=matched,
requires_approval=should_require_approval,
reason=self._build_reason(matched, should_block, should_require_approval),
timestamp=datetime.now(timezone.utc).isoformat()
)
# Log every evaluation
self.action_log.append({
"action": action[:200], # truncate for logging
"result": result,
"context": context
})
if should_block:
logger.warning(f"BLOCKED action: {matched} | {action[:100]}")
elif should_require_approval:
logger.info(f"Approval required: {matched} | {action[:100]}")
return result
def _build_reason(self, matched, blocked, approval):
if blocked:
return f"Action blocked by rules: {', '.join(matched)}"
if approval:
return f"Action requires approval: {', '.join(matched)}"
if matched:
return f"Action logged by rules: {', '.join(matched)}"
return "No safety rules triggered"
def get_action_log(self):
'''Return the full action log for auditing.'''
return self.action_logStep 4: Wrap Your AI Agent with Guardrails
Now let's create a safe agent wrapper that intercepts every tool call:
from openai import AsyncOpenAI
import asyncio
class SafeAgent:
def __init__(self, api_key: str, base_url: str, model: str = "gpt-5"):
self.client = AsyncOpenAI(api_key=api_key, base_url=base_url)
self.model = model
self.guardrail = GuardrailEngine(SAFETY_RULES)
self.max_tool_calls = 25 # Hard limit per session
self.tool_call_count = 0
async def execute_tool(self, tool_name: str, arguments: dict) -> dict:
'''Execute a tool call with guardrail protection.'''
self.tool_call_count += 1
# Rate limit check
if self.tool_call_count > self.max_tool_calls:
return {"error": "Tool call limit exceeded. Session terminated."}
# Serialize the action for guardrail inspection
action_str = f"{tool_name} {str(arguments)}"
# Evaluate against guardrails
result = self.guardrail.evaluate(action_str, {
"tool": tool_name,
"call_number": self.tool_call_count
})
if not result.allowed:
return {
"error": f"Action blocked by safety guardrail: {result.reason}",
"risk_level": result.risk_level.value
}
if result.requires_approval:
# In production, this would integrate with your approval workflow
approved = await self._request_human_approval(tool_name, arguments, result)
if not approved:
return {"error": "Action not approved by human operator"}
# Execute the actual tool
return await self._run_tool(tool_name, arguments)
async def _request_human_approval(self, tool_name, arguments, result):
'''Request human approval for high-risk actions.'''
print(f"\n{'='*60}")
print(f"APPROVAL REQUIRED for: {tool_name}")
print(f"Arguments: {arguments}")
print(f"Risk level: {result.risk_level.value}")
print(f"Reason: {result.reason}")
print(f"{'='*60}")
response = input("Approve this action? (yes/no): ")
return response.lower().startswith("y")
async def _run_tool(self, tool_name, arguments):
'''The actual tool implementation goes here.'''
# Replace with your real tool implementations
return {"status": "executed", "tool": tool_name, "args": arguments}
def get_stats(self):
'''Get session statistics.'''
return {
"total_tool_calls": self.tool_call_count,
"action_log_entries": len(self.guardrail.get_action_log()),
"remaining_calls": self.max_tool_calls - self.tool_call_count
}Step 5: Add Behavioral Monitoring
Beyond rule-based guardrails, you should monitor the agent's behavior patterns for anomalies:
class BehaviorMonitor:
'''Detects anomalous agent behavior patterns.'''
def __init__(self):
self.action_history = []
self.alerts = []
def record_action(self, tool_name: str, success: bool):
self.action_history.append({
"tool": tool_name,
"success": success,
"time": datetime.now(timezone.utc)
})
self._check_patterns()
def _check_patterns(self):
# Check for rapid-fire tool calls (potential loop)
if len(self.action_history) >= 5:
recent = self.action_history[-5:]
time_span = (recent[-1]["time"] - recent[0]["time"]).total_seconds()
if time_span < 2:
self.alerts.append({
"type": "rapid_fire",
"message": "Agent made 5+ tool calls in under 2 seconds"
})
# Check for repeated failures (agent may be stuck)
recent_results = [a["success"] for a in self.action_history[-10:]]
if len(recent_results) >= 5 and not any(recent_results):
self.alerts.append({
"type": "repeated_failures",
"message": "Agent has failed 5+ consecutive actions"
})
# Check for tool diversity (agent using too many different tools)
recent_tools = set(a["tool"] for a in self.action_history[-20:])
if len(recent_tools) > 10:
self.alerts.append({
"type": "tool_explosion",
"message": "Agent is using an unusually large number of different tools"
})Step 6: Put It All Together
async def main():
agent = SafeAgent(
api_key="your-api-key",
base_url="https://api.qubax.ai/v1",
model="gpt-5"
)
# Example: Agent tries to delete a file (BLOCKED)
result = await agent.execute_tool("terminal", {"command": "rm -rf /"})
print(f"Result: {result}")
# Output: Action blocked by safety guardrail
# Example: Agent tries to send email (LOGGED)
result = await agent.execute_tool("send_email", {
"to": "[email protected]",
"subject": "Hello",
"body": "Test message"
})
print(f"Result: {result}")
# Output: Action executed (logged for monitoring)
# Print session stats
print(f"\nSession stats: {agent.get_stats()}")
asyncio.run(main())Best Practices for Production
Here are the key principles to follow when deploying guardrails in production:
- Default to deny. If the guardrail can't determine whether an action is safe, block it. It's easier to whitelist safe actions than to recover from an unsafe one.
- Log everything. Every action — allowed, blocked, or pending — should be logged with full context. You need this for debugging, compliance, and post-incident analysis.
- Make guardrails tamper-proof. The AI agent should not be able to modify the guardrail rules or bypass the evaluation step. Run the guardrail in a separate process or service if possible.
- Test adversarially. Before deploying, try to break your own guardrails. Have a red team attempt to craft inputs that bypass the safety rules. See our guide to AI red teaming for techniques.
- Keep rules updated. New attack vectors emerge constantly. Review and update your safety rules regularly based on incident reports and new research.
- Combine rule-based and AI-based detection. Rules are fast and predictable but can't catch everything. Consider adding an AI-based "second opinion" layer that uses a small model to flag suspicious actions that don't match any rule.
Build safer AI agents with Qubax AI. Our API platform supports OpenAI-compatible tool calling, streaming, and integrates seamlessly with guardrail systems. Read our docs for more examples.
FAQ
What are AI agent guardrails?
Guardrails are safety systems that monitor and control what an AI agent can do. They intercept the agent's actions before execution and decide whether to allow, block, or require human approval based on predefined safety rules.
How do guardrails prevent rogue AI agents?
Guardrails inspect every action the agent attempts. If an action matches a dangerous pattern (like deleting files, accessing production systems, or exfiltrating data), the guardrail blocks it before it can execute. This prevents the most common forms of rogue behavior.
Should I use rules or AI-based filtering for guardrails?
Both. Rule-based filtering is fast and predictable — use it for known dangerous patterns. AI-based filtering (using a small model to evaluate actions) catches novel threats that rules miss. Combining both gives you the best protection.
How much latency do guardrails add?
Minimal. Rule-based guardrails use simple regex matching, which takes microseconds. Even AI-based second-opinion systems typically add less than 500ms. The safety benefit far outweighs the latency cost.
Can an AI agent bypass guardrails?
If implemented correctly, no. The guardrail must run in a separate trust zone that the agent cannot modify or bypass. The key principle is that the agent should never have direct access to tools — all tool calls must go through the guardrail layer.
What tools do I need to implement guardrails?
You need a programming language with regex support (Python, JavaScript, Go, etc.), a logging system, and optionally a small AI model for advanced detection. The guardrail system itself is straightforward to build — the hard part is defining the right safety rules for your use case.
