The @guard Decorator
Complete reference for agentguard's core API — the @guard decorator and GuardConfig.
Basic Usage
The @guard decorator is the primary API surface. Wrap any tool function to add safety guardrails.
python
from agentguard import guard
# Zero config — just wraps the function
@guard
def my_tool(x: str) -> str:
return x
# With options
@guard(
validate_input=True, # Validate args against type hints
validate_output=True, # Validate return value
detect_hallucination=True, # Multi-signal hallucination detection
max_retries=3, # Retry on failure
timeout=30.0, # Kill after 30 seconds
record=True, # Write traces to the configured store
trace_backend="sqlite", # Production default backend
trace_dir="./traces", # SQLite parent directory or JSONL directory
session_id="my-session", # Group traces by session
)
def my_tool(x: str) -> dict:
...GuardConfig Reference
For reusable configuration across multiple tools, use GuardConfig:
python
from agentguard import guard, GuardConfig
from agentguard.config import CircuitBreakerConfig, RateLimitConfig, BudgetConfig
config = GuardConfig(
validate_input=True,
max_retries=2,
timeout=10.0,
circuit_breaker=CircuitBreakerConfig(failure_threshold=5),
rate_limit=RateLimitConfig(calls_per_minute=60),
budget=BudgetConfig(max_cost_per_session=5.00),
)
@guard(config=config)
def tool_a(x: str) -> str: ...
@guard(config=config)
def tool_b(y: int) -> int: ...Configuration Fields
| Field | Type | Default | Description |
|---|---|---|---|
validate_input | bool | False | Validate arguments against type hints |
validate_output | bool | False | Validate return value against return type |
detect_hallucination | bool | False | Enable multi-signal hallucination detection |
max_retries | int | 0 | Maximum retry attempts on failure |
timeout | float | None | Timeout in seconds |
record | bool | False | Record traces to the configured store |
trace_dir | str | "./traces" | SQLite parent directory or JSONL directory |
trace_backend | str | "sqlite" | Trace backend: sqlite or jsonl |
trace_db_path | str | None | Optional explicit SQLite database path |
session_id | str | auto | Session identifier for trace grouping |
circuit_breaker | CircuitBreakerConfig | None | Circuit breaker configuration |
rate_limit | RateLimitConfig | None | Rate limiter configuration |
budget | BudgetConfig | None | Budget enforcement configuration |
retry | RetryConfig | None | Detailed retry configuration |
before_call | Callable | None | Hook called before tool execution |
after_call | Callable | None | Hook called after tool execution |
custom_validators | list | None | Custom validator instances |
Retry & Timeout
Configure detailed retry behavior with RetryConfig:
python
from agentguard.config import RetryConfig
@guard(
retry=RetryConfig(
max_retries=5,
initial_delay=0.5, # First retry after 500ms
backoff_factor=2.0, # Double delay each retry
max_delay=30.0, # Cap at 30s between retries
jitter=True, # Add randomness to prevent thundering herd
retry_on=(TimeoutError, ConnectionError), # Only retry these
),
timeout=60.0, # Hard timeout per attempt
)
def unreliable_api(query: str) -> dict:
return requests.get(f"https://api.example.com?q={{query}}").json()💡 Timeout applies per attempt
The timeout setting applies to each individual attempt, not the total time across all retries. A function with max_retries=3 and timeout=10.0 could take up to 30+ seconds total.
Custom Validators
Extend validation logic with your own validator classes:
python
from agentguard import guard, Validator, ValidationResult
class ContentSafetyValidator(Validator):
name = "content_safety"
def validate(self, result, context):
# Check the tool output for unsafe content
text = str(result)
if any(word in text.lower() for word in self.blocked_words):
return ValidationResult(
valid=False,
message="Response contains blocked content"
)
return ValidationResult(valid=True)
@guard(custom_validators=[ContentSafetyValidator(blocked_words=["password", "secret"])])
def query_database(sql: str) -> list:
return db.execute(sql)Lifecycle Hooks
Run code before or after every guarded tool call:
python
def log_call(func_name, args, kwargs):
print(f"Calling {{func_name}} with {{args}}")
def log_result(func_name, result, elapsed_ms):
print(f"{{func_name}} returned in {{elapsed_ms:.1f}}ms")
@guard(before_call=log_call, after_call=log_result)
def my_tool(x: str) -> str:
return x.upper()