Guard Agent¶
Guard Agent is a framework-agnostic telemetry and monitoring solution for the Guard security ecosystem. Paired with an adapter (fastapi-guard, flaskapi-guard, djangoapi-guard, or tornadoapi-guard), it collects security events, performance metrics, and operational telemetry in real time — feeding centralized security operations, compliance reporting, and dynamic threat response through an enterprise-grade management platform.
Renamed from fastapi-guard-agent in 2.0.0
This package was previously published as fastapi-guard-agent. The Python import path (from guard_agent import ...) is unchanged. Existing pip install fastapi-guard-agent commands continue to work via a meta-package that transitively pulls guard-agent.
Website · Playground · Dashboard · Discord
Guard Agent bridges application-level security enforcement and enterprise security operations. It collects security events, performance metrics, and operational telemetry from any supported Guard adapter, providing real-time visibility into threats and anomalies across your Python web stack regardless of framework.
The collected telemetry flows to the Guard platform:
- app.guard-core.com — the dashboard where your project's security events, metrics, and dynamic rules are managed in real time.
- playground.guard-core.com — a sandbox environment for experimenting with Guard configurations before wiring them into production.
- guard-core.com — the umbrella product site with ecosystem overview and adapter matrix.
Quick Start¶
Guard Agent is embedded by each framework's adapter — enable it via the adapter's SecurityConfig. The following example uses the FastAPI adapter; Flask, Django, and Tornado adapters expose analogous interfaces (see Installation for per-adapter setup).
from fastapi import FastAPI
from guard import SecurityConfig, SecurityMiddleware
app = FastAPI()
# Configure FastAPI Guard with built-in agent support
config = SecurityConfig(
# Basic security settings
auto_ban_threshold=5,
auto_ban_duration=300,
# Enable agent for telemetry
enable_agent=True,
agent_api_key="YOUR_API_KEY",
agent_project_id="YOUR_PROJECT_ID",
agent_endpoint="https://api.guard-core.com",
# Agent configuration
agent_buffer_size=100,
agent_flush_interval=30,
agent_enable_events=True,
agent_enable_metrics=True,
# Enable dynamic rules from SaaS
enable_dynamic_rules=True,
dynamic_rule_interval=300,
)
# Add security middleware - events are sent automatically
middleware = SecurityMiddleware(app, config=config)
@app.get("/")
async def root():
return {"message": "Hello World"}
Core Capabilities¶
Security Intelligence¶
- Automated Event Collection: Captures comprehensive security events including authentication failures, authorization violations, rate limit breaches, and suspicious request patterns without manual instrumentation
- Real-Time Threat Detection: Provides immediate visibility into security incidents with sub-second event propagation to the management platform
- Behavioral Analytics: Tracks request patterns and user behavior to identify anomalies and potential security threats
Enterprise Architecture¶
- Zero-Impact Performance: Leverages asynchronous I/O and intelligent buffering to ensure telemetry collection adds negligible overhead to application performance
- Fault-Tolerant Design: Implements circuit breakers, exponential backoff with jitter, and intelligent retry mechanisms to maintain operation during network disruptions
- Multi-Tier Buffering: Combines in-memory and persistent Redis buffering to guarantee zero data loss during outages or maintenance windows
Operational Excellence¶
- Dynamic Policy Management: Supports real-time security policy updates without application restart, enabling immediate threat response
- Protocol-Based Extensibility: Provides clean abstractions for custom transport implementations, storage backends, and data processors
- Comprehensive Observability: Captures granular metrics including response times, error rates, and resource utilization for complete operational visibility
Data Governance¶
- Privacy-First Design: Implements configurable data redaction for sensitive headers and payload truncation to meet compliance requirements
- Intelligent Rate Limiting: Prevents API exhaustion through client-side rate limiting with adaptive backpressure
- Health Monitoring: Continuous self-diagnostics with automatic health reporting and degradation detection
Installation¶
Alternatives:
The legacy fastapi-guard-agent name is still published as a meta-package that installs guard-agent transitively — existing installs keep working, but new projects should use guard-agent directly.
System Requirements¶
- Python 3.10 or higher (3.11+ recommended for optimal performance)
- A Guard adapter matching your web framework:
fastapi-guard,flaskapi-guard,djangoapi-guard, ortornadoapi-guard - Optional Redis 6.0+ for persistent buffering
Implementation Patterns¶
Primary Integration Pattern¶
The recommended approach leverages FastAPI Guard's built-in agent support for automatic telemetry collection:
from fastapi import FastAPI
from guard import SecurityConfig, SecurityMiddleware
app = FastAPI()
config = SecurityConfig(
# Enable agent
enable_agent=True,
agent_api_key="your-api-key",
agent_project_id="your-project-id",
# Other security settings
enable_rate_limiting=True,
rate_limit=100,
rate_limit_window=60,
)
middleware = SecurityMiddleware(app, config=config)
# Agent automatically initialized with comprehensive event collection
Advanced Usage Pattern¶
For specialized use cases requiring direct agent control, such as custom event handling or integration with non-FastAPI Guard systems:
from fastapi import FastAPI
from guard_agent import guard_agent, AgentConfig, SecurityEvent
from guard_agent.utils import get_current_timestamp
app = FastAPI()
# Configure agent directly
config = AgentConfig(
api_key="your-api-key",
project_id="your-project-id",
buffer_size=100,
flush_interval=30,
)
agent = guard_agent(config)
@app.on_event("startup")
async def startup_event():
await agent.start()
@app.on_event("shutdown")
async def shutdown_event():
await agent.stop()
@app.post("/report-event")
async def report_event():
# Manually send an event
event = SecurityEvent(
timestamp=get_current_timestamp(),
event_type="custom_rule_triggered",
ip_address="192.168.1.100",
action_taken="logged",
reason="Manual event",
endpoint="/report-event",
method="POST",
)
await agent.send_event(event)
return {"status": "event sent"}
Advanced Configuration¶
Redis Integration¶
For production deployments, enable Redis for persistent buffering:
from redis.asyncio import Redis
from guard_agent.redis_handler import RedisHandler
# Configure Redis
redis_client = Redis.from_url("redis://localhost:6379")
redis_handler = RedisHandler(redis_client)
# Initialize agent with Redis
agent = guard_agent(config)
await agent.initialize_redis(redis_handler)
Custom Transport¶
Implement custom transport for specialized backends:
from guard_agent.protocols import TransportProtocol
from guard_agent.models import SecurityEvent, SecurityMetric, AgentStatus, DynamicRules
class CustomTransport(TransportProtocol):
async def send_events(self, events: list[SecurityEvent]) -> bool:
# Your custom implementation
return True
async def send_metrics(self, metrics: list[SecurityMetric]) -> bool:
# Your custom implementation
return True
async def fetch_dynamic_rules(self) -> DynamicRules | None:
# Your custom implementation
return None
async def send_status(self, status: AgentStatus) -> bool:
# Your custom implementation
return True
# Use custom transport
agent = guard_agent(config)
agent.transport = CustomTransport()
Data Models¶
SecurityEvent¶
Represents security-related events in your application:
from guard_agent.models import SecurityEvent
from guard_agent.utils import get_current_timestamp
event = SecurityEvent(
timestamp=get_current_timestamp(),
event_type="ip_banned",
ip_address="192.168.1.100",
country="US",
user_agent="Mozilla/5.0...",
action_taken="block",
reason="Rate limit exceeded",
endpoint="/api/v1/login",
method="POST",
status_code=429,
response_time=0.125,
metadata={"attempts": 5, "window": 60}
)
await agent.send_event(event)
SecurityMetric¶
Represents performance and usage metrics:
from guard_agent.models import SecurityMetric
metric = SecurityMetric(
timestamp=get_current_timestamp(),
metric_type="request_count",
value=100.0,
endpoint="/api/v1/users",
tags={"method": "GET", "status": "200"}
)
await agent.send_metric(metric)
Dynamic Rules¶
Fetch and apply dynamic security rules from your SaaS backend:
# Fetch current rules
rules = await agent.get_dynamic_rules()
if rules:
# Apply IP blacklist
if client_ip in rules.ip_blacklist:
# Block request
pass
# Apply rate limits
endpoint_limit = rules.endpoint_rate_limits.get("/api/sensitive")
if endpoint_limit:
requests, window = endpoint_limit
# Apply endpoint-specific rate limit
pass
# Check emergency mode
if rules.emergency_mode:
# Apply stricter security measures
pass
Monitoring and Health¶
Agent Status¶
Monitor agent health and performance:
status = await agent.get_status()
print(f"Status: {status.status}") # healthy, degraded, error
print(f"Uptime: {status.uptime}s")
print(f"Events sent: {status.events_sent}")
print(f"Buffer size: {status.buffer_size}")
print(f"Errors: {status.errors}")
Statistics¶
Get detailed agent statistics:
stats = agent.get_stats()
print(f"Running: {stats['running']}")
print(f"Events sent: {stats['events_sent']}")
print(f"Transport stats: {stats['transport_stats']}")
print(f"Buffer stats: {stats['buffer_stats']}")
Error Handling¶
Circuit Breaker¶
The agent includes a circuit breaker to handle backend failures gracefully:
# Check circuit breaker state
transport_stats = agent.transport.get_stats()
if transport_stats["circuit_breaker_state"] == "OPEN":
print("Backend is unavailable, circuit breaker is open")
Retry Logic¶
Failed requests are automatically retried with exponential backoff:
config = AgentConfig(
api_key="your-api-key",
retry_attempts=3, # Number of retries
backoff_factor=1.5, # Exponential backoff factor
timeout=30, # Request timeout
)