Resource Guard
The Resource Guard prevents resource exhaustion, runaway costs, and infinite loops in agent execution. It enforces hard limits on execution time, memory usage, API costs, call depth, rate limits, and network domains.
Quick Start
from enforcecore import Enforcer
# Resource limits are configured in your policy
enforcer = Enforcer.from_file("policy.yaml")
# Limits are enforced automatically via @enforce or enforce_sync/enforce_async
# Manual access to the guard:
guard = enforcer.guardNote: When using the @enforce decorator, resource guards are applied automatically based on your policy. You only need direct access for custom pipelines.
Policy Configuration
Resource limits are defined in YAML policy files:
name: "production"
rules:
resource_limits:
max_call_duration_seconds: 30 # Timeout per call
max_memory_mb: 256 # Memory limit per call
max_cost_usd: 5.00 # Session cost budget
rate_limits:
enabled: true
per_tool:
web_search:
max_calls: 10
window_seconds: 60
llm_call:
max_calls: 50
window_seconds: 60
global:
max_calls: 500
window_seconds: 60
network:
enabled: true
allowed_domains:
- "api.openai.com"
- "api.anthropic.com"
- "*.wikipedia.org" # Wildcard support
denied_domains:
- "*.malicious.xyz"
deny_all_other: true # Block everything not in allowed listClasses
ResourceGuard
Monitors and enforces resource consumption limits during function execution.
The guard is created automatically from policy configuration. During enforcement, it handles:
- Execution time limits — kills calls exceeding
max_call_duration_seconds - Memory limits — monitors RSS memory, kills calls exceeding
max_memory_mb - Cost budget — tracks cumulative session cost, blocks when exceeded
CostTracker
Tracks cumulative token usage and estimated costs across the session.
# Accessed via Enforcer
enforcer = Enforcer.from_file("policy.yaml")
# Record a cost after an LLM call
total = enforcer.record_cost(0.02)
print(f"Session total: ${total:.4f}")
# Raises CostLimitError if budget exceededKillSwitch
Emergency termination mechanism for runaway agents. Activated automatically when resource limits are critically exceeded.
RateLimiter
Sliding-window rate limiter for per-tool and global call frequency limits.
from enforcecore.guard.ratelimit import RateLimiter, RateLimit
limiter = RateLimiter()
limiter.add_limit("web_search", RateLimit(max_calls=10, window_seconds=60))
# Check before calling
limiter.check("web_search") # Raises RateLimitError if exceededDomainChecker
Validates network requests against allowed/denied domain lists with wildcard support.
from enforcecore.guard.network import DomainChecker
checker = DomainChecker(
allowed_domains=["api.openai.com", "*.wikipedia.org"],
denied_domains=["*.malicious.xyz"],
deny_all_other=True,
)
checker.check("api.openai.com") # OK
checker.check("evil.malicious.xyz") # Raises DomainDeniedError
checker.check("random-site.com") # Raises DomainDeniedError (deny_all_other)Error Handling
All guard errors are specific to the type of limit exceeded:
EnforceCoreError
├── EnforcementViolation
│ ├── CostLimitError — cost budget exceeded
│ └── DomainDeniedError — network domain blocked
├── ResourceLimitError — time/memory limit exceeded
└── GuardError
└── RateLimitError — calls/minute exceededExample — Catching cost errors:
from enforcecore.core.types import CostLimitError
try:
enforcer.record_cost(50.00)
except CostLimitError as e:
print(f"Budget exceeded: {e}")Example — Catching rate limit errors:
from enforcecore.guard.ratelimit import RateLimitError
try:
limiter.check("web_search")
except RateLimitError as e:
print(f"Rate limited: {e}")Environment Variables
| Variable | Default | Description |
|---|---|---|
ENFORCECORE_COST_BUDGET_USD |
None |
Global cost budget (overrides policy). |
ENFORCECORE_FAIL_OPEN |
false |
If true, allow calls on guard errors. Never in production. |
See Also
- Enforcer API — How the guard integrates with enforcement.
- Policy Engine — Defining resource limits in YAML.
- Architecture — Where the guard sits in the enforcement pipeline.