Guardrails and Safety in Agent Systems (2026)
Agents that can execute arbitrary tools pose safety risks: they might delete the wrong database record, trigger expensive API calls, or expose sensitive information. Guardrails are defensive mechanisms that constrain agent behavior while preserving autonomy. Guardrails include input validation (rejecting malformed requests), output filtering (blocking unsafe responses), tool access controls (limiting which agents can call which tools), budget enforcement (preventing runaway costs), and alignment checks (ensuring actions match user intent). This article teaches you to build secure agents that operate safely in production while maintaining trust and transparency.
Safety in agents is not optional. A human assistant's mistakes are caught by oversight. An autonomous agent's mistakes can cause real harm before anyone notices. Guardrails make agents auditable and trustworthy.
Input Validation and Sanitization
Before an agent processes a user request, validate and sanitize it:
import re
from typing import Optional
class InputValidator:
"""Validates and sanitizes user inputs to agents."""
def __init__(self):
# Patterns that indicate potentially dangerous requests
self.dangerous_patterns = [
r"delete.*where", # SQL injection
r"drop\s+(table|database)",
r"exec\s*\(",
r"eval\s*\(",
r"rm\s+-rf", # Dangerous shell commands
r"shutdown",
]
# Maximum request length
self.max_length = 10000
# Blocked keywords
self.blocked_keywords = ["admin_password", "private_key", "secret_key"]
def validate(self, user_input: str) -> tuple[bool, Optional[str]]:
"""Validate user input. Return (is_valid, error_message)."""
# Length check
if len(user_input) > self.max_length:
return False, f"Input exceeds {self.max_length} characters"
# Empty check
if not user_input.strip():
return False, "Empty input"
# Pattern matching for dangerous content
for pattern in self.dangerous_patterns:
if re.search(pattern, user_input, re.IGNORECASE):
return False, f"Request contains potentially dangerous pattern: {pattern}"
# Check for blocked keywords
lower_input = user_input.lower()
for keyword in self.blocked_keywords:
if keyword in lower_input:
return False, f"Request mentions blocked keyword: {keyword}"
return True, None
def sanitize(self, user_input: str) -> str:
"""Remove or escape potentially problematic content."""
# Remove null bytes
sanitized = user_input.replace('\x00', '')
# Remove excessive whitespace
sanitized = ' '.join(sanitized.split())
# Escape special characters if needed
# (depends on downstream usage)
return sanitized
# Usage
validator = InputValidator()
user_input = "Show me the weather in NYC"
is_valid, error = validator.validate(user_input)
if not is_valid:
print(f"Invalid input: {error}")
else:
sanitized = validator.sanitize(user_input)
# Pass to agent
Tool Access Control
Restrict which tools agents can access based on context, user role, or sensitivity:
from enum import Enum
from dataclasses import dataclass
class AccessLevel(Enum):
PUBLIC = "public"
INTERNAL = "internal"
RESTRICTED = "restricted"
@dataclass
class ToolAccessPolicy:
"""Defines who can access a tool."""
tool_name: str
access_level: AccessLevel
allowed_roles: list[str] = None
max_calls_per_session: int = None
max_cost_per_call: float = None
class ToolAccessController:
"""Controls which tools agents can call."""
def __init__(self):
self.policies = {
"delete_user": ToolAccessPolicy(
"delete_user",
AccessLevel.RESTRICTED,
allowed_roles=["admin"],
max_calls_per_session=1
),
"send_email": ToolAccessPolicy(
"send_email",
AccessLevel.INTERNAL,
allowed_roles=["staff", "admin"],
max_cost_per_call=0.10
),
"search_database": ToolAccessPolicy(
"search_database",
AccessLevel.PUBLIC,
max_calls_per_session=100
),
}
# Track usage per session
self.session_usage = {}
def check_access(
self,
tool_name: str,
user_role: str,
session_id: str
) -> tuple[bool, Optional[str]]:
"""Check if user can call this tool. Return (allowed, reason_if_denied)."""
if tool_name not in self.policies:
return False, f"Unknown tool: {tool_name}"
policy = self.policies[tool_name]
# Role-based access
if policy.allowed_roles and user_role not in policy.allowed_roles:
return False, f"User role '{user_role}' cannot access tool '{tool_name}'"
# Rate limiting
if policy.max_calls_per_session:
key = f"{session_id}:{tool_name}"
current_calls = self.session_usage.get(key, 0)
if current_calls >= policy.max_calls_per_session:
return False, f"Tool limit reached for '{tool_name}' (max {policy.max_calls_per_session})"
return True, None
def record_call(self, tool_name: str, session_id: str):
"""Record a tool call for rate limiting."""
key = f"{session_id}:{tool_name}"
self.session_usage[key] = self.session_usage.get(key, 0) + 1
def filter_tools(self, all_tools: list, user_role: str, session_id: str) -> list:
"""Filter tools to only those the user can access."""
accessible_tools = []
for tool in all_tools:
allowed, _ = self.check_access(tool["name"], user_role, session_id)
if allowed:
accessible_tools.append(tool)
return accessible_tools
# Usage
controller = ToolAccessController()
# Check access before calling
allowed, reason = controller.check_access("delete_user", "staff", "session_123")
if not allowed:
print(f"Access denied: {reason}")
else:
controller.record_call("delete_user", "session_123")
# Filter tools for a specific user
all_tools = [
{"name": "delete_user", "description": "..."},
{"name": "search_database", "description": "..."},
{"name": "send_email", "description": "..."},
]
user_tools = controller.filter_tools(all_tools, user_role="staff", session_id="session_123")
print(f"Available tools: {[t['name'] for t in user_tools]}")
Output Filtering and Content Moderation
Filter agent responses to prevent leaks and ensure appropriateness:
import json
class OutputFilter:
"""Filters agent outputs for safety."""
def __init__(self):
# Sensitive data patterns
self.sensitive_patterns = {
"email": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
"api_key": r"(sk-|api_key['\"]?[:=])",
"credit_card": r"\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b",
"ssn": r"\b\d{3}-\d{2}-\d{4}\b",
}
def redact_sensitive_data(self, text: str) -> str:
"""Remove or redact sensitive information."""
redacted = text
for data_type, pattern in self.sensitive_patterns.items():
redacted = re.sub(pattern, f"[REDACTED_{data_type}]", redacted)
return redacted
def filter_response(self, response: str) -> tuple[str, list[str]]:
"""Filter response and return (filtered_text, warnings)."""
warnings = []
# Check for sensitive data
redacted = self.redact_sensitive_data(response)
if redacted != response:
warnings.append("Response contained sensitive data (redacted)")
# Check for extremely long responses (potential information dump)
if len(redacted) > 50000:
redacted = redacted[:50000] + "\n[Response truncated]"
warnings.append("Response was too long (truncated)")
# Check for suspicious content
if re.search(r"password|secret|private", redacted, re.IGNORECASE):
warnings.append("Response mentions sensitive credentials (verify appropriateness)")
return redacted, warnings
# Usage
filter_output = OutputFilter()
response = "User email is [email protected] and API key is sk-12345..."
filtered, warnings = filter_output.filter_response(response)
print(f"Filtered: {filtered}")
print(f"Warnings: {warnings}")
Budget and Cost Controls
Prevent agents from incurring unexpected costs:
from datetime import datetime, timedelta
class BudgetController:
"""Tracks and enforces API costs for agents."""
def __init__(self, monthly_budget: float):
self.monthly_budget = monthly_budget
self.usage_log = [] # List of (timestamp, cost) tuples
def log_cost(self, cost: float, tool_name: str):
"""Log an API call cost."""
self.usage_log.append({
"timestamp": datetime.now(),
"cost": cost,
"tool_name": tool_name
})
def get_month_to_date_cost(self) -> float:
"""Calculate total cost this month."""
now = datetime.now()
month_start = now.replace(day=1)
month_cost = sum(
entry["cost"]
for entry in self.usage_log
if entry["timestamp"] >= month_start
)
return month_cost
def can_afford_call(self, estimated_cost: float) -> tuple[bool, str]:
"""Check if the agent can afford the next API call."""
current_month_cost = self.get_month_to_date_cost()
remaining_budget = self.monthly_budget - current_month_cost
if estimated_cost > remaining_budget:
return False, (
f"Cost ${estimated_cost} exceeds remaining budget ${remaining_budget:.2f}. "
f"Monthly budget: ${self.monthly_budget:.2f}, Used: ${current_month_cost:.2f}"
)
# Also check percentage of budget (warn if over 80%)
if (current_month_cost + estimated_cost) / self.monthly_budget > 0.8:
return True, (
f"Warning: This call will exceed 80% of monthly budget. "
f"Remaining: ${remaining_budget - estimated_cost:.2f}"
)
return True, None
def report(self) -> str:
"""Generate a usage report."""
month_cost = self.get_month_to_date_cost()
remaining = self.monthly_budget - month_cost
percentage = (month_cost / self.monthly_budget) * 100
return (
f"Budget Report:\n"
f"Monthly Budget: ${self.monthly_budget:.2f}\n"
f"Used: ${month_cost:.2f} ({percentage:.1f}%)\n"
f"Remaining: ${remaining:.2f}"
)
# Usage
budget = BudgetController(monthly_budget=100.0)
# Before making an expensive API call
estimated_cost = 5.0
can_proceed, warning = budget.can_afford_call(estimated_cost)
if not can_proceed:
print(f"Cannot proceed: {warning}")
else:
# Make the call
budget.log_cost(estimated_cost, "search_web")
print(warning if warning else "Proceeding with call")
print(budget.report())
Transparency and Auditing
Log all agent actions for auditing and debugging:
import logging
import json
from datetime import datetime
class AgentAuditLog:
"""Logs all agent activities for auditing."""
def __init__(self, log_file: str):
self.log_file = log_file
self.logger = logging.getLogger("agent_audit")
handler = logging.FileHandler(log_file)
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
self.logger.setLevel(logging.INFO)
def log_user_request(self, session_id: str, user_input: str, user_role: str):
"""Log incoming user request."""
self.logger.info(
json.dumps({
"event": "user_request",
"session_id": session_id,
"user_role": user_role,
"input_length": len(user_input),
"timestamp": datetime.now().isoformat()
})
)
def log_tool_call(self, session_id: str, tool_name: str, params: dict, result: str):
"""Log agent tool call."""
self.logger.info(
json.dumps({
"event": "tool_call",
"session_id": session_id,
"tool_name": tool_name,
"params": params,
"result_length": len(result),
"timestamp": datetime.now().isoformat()
})
)
def log_guardrail_violation(self, session_id: str, violation_type: str, details: str):
"""Log a guardrail violation."""
self.logger.warning(
json.dumps({
"event": "guardrail_violation",
"session_id": session_id,
"violation_type": violation_type,
"details": details,
"timestamp": datetime.now().isoformat()
})
)
# Usage
audit_log = AgentAuditLog("/var/log/agent_audit.log")
audit_log.log_user_request("session_123", "What's the weather?", "user")
audit_log.log_tool_call("session_123", "get_weather", {"city": "NYC"}, "65F and clear")
audit_log.log_guardrail_violation("session_123", "rate_limit", "Tool call limit exceeded")
Key Takeaways
- Input validation rejects malformed or dangerous requests before processing
- Tool access controls restrict which agents/users can call sensitive tools
- Output filtering redacts sensitive data and prevents information leaks
- Budget controls prevent runaway API costs
- Audit logging provides transparency and enables debugging of agent behavior
Frequently Asked Questions
Should guardrails block or warn?
Depends on severity. Block dangerous requests (delete without confirmation). Warn on suspicious but acceptable ones (large responses). Log everything for later review.
How do I validate agent reasoning without blocking it?
Use a review agent: have a conservative agent validate the primary agent's decisions before execution. This adds latency but improves safety.
Can guardrails prevent all failures?
No. They reduce risk but don't eliminate it. Combine guardrails with human oversight for high-stakes decisions.
How do I handle a legitimate request that looks suspicious?
Allow users to override guardrails with explicit confirmation. Log the override for audit trails.