Enforcer

The core enforcement engine that intercepts function calls and applies policies at runtime. This is the primary entry point for integrating EnforceCore into your application.

Quick Start

from enforcecore import enforce, Enforcer, Policy

# Decorator approach (recommended)
@enforce(policy="policy.yaml")
async def run_agent(user_input: str) -> str:
    ...

# Or use bare decorator with default policy
@enforce
async def run_agent(user_input: str) -> str:
    ...

# Programmatic approach
enforcer = Enforcer.from_file("policy.yaml")
result = enforcer.enforce_sync(my_function, *args, tool_name="search")

The `@enforce` Decorator

The main public API for wrapping any sync or async callable with policy enforcement.

Signatures

# Bare decorator (uses ENFORCECORE_DEFAULT_POLICY env var)
@enforce
async def my_tool(query: str) -> str: ...

# With arguments
@enforce(policy="policy.yaml", tool_name="search")
async def my_tool(query: str) -> str: ...

Parameters

Parameter	Type	Default	Description
`policy`	`str \| Path \| Policy \| None`	`None`	Path to YAML, a `Policy` object, or `None` (uses default policy from env).
`tool_name`	`str \| None`	`None`	Override the tool name for policy matching. Defaults to the function name.

Behavior

Automatically detects sync vs async via inspect.iscoroutinefunction()
Caches policies by file path with mtime-based invalidation (max 64 entries, FIFO eviction)
Thread-safe: no mutable per-call state
Raises EnforcementViolation if the action is blocked

Enforcement Pipeline

When a decorated function is called, EnforceCore executes this pipeline:

Input validation — reject oversized payloads (default: 10MB limit)
Tool name validation — prevent injection via tool names
Pre-call policy evaluation — check allow/deny lists, content rules, network policy
PII redaction on inputs — redact sensitive data from arguments
Cost budget check — verify cost budget not exceeded
Rate limit check — sliding-window per-tool and global limits
Execute function — with resource guards (time, memory limits)
PII redaction on output — redact sensitive data from return value
Post-call content rules — validate output against block patterns
Audit trail — record the decision in the Merkle chain
Lifecycle hooks — fire post-call and violation hooks

Examples

Strict enforcement (default):

from enforcecore import enforce

@enforce(policy="strict-search.yaml")
async def search_web(query: str) -> str:
    return await search_api.query(query)

# If policy denies "search_web", raises EnforcementViolation
# BEFORE search_api.query() is called.

Log-only mode (shadow deployment):

# policy.yaml
name: "shadow-mode"
on_violation: "log"     # log violations but don't block
rules:
  denied_tools: ["execute_shell"]

@enforce(policy="policy.yaml")
async def run_shell(cmd: str) -> str:
    # Runs even if policy would deny — violation logged to audit trail
    return subprocess.check_output(cmd, shell=True).decode()

Sync functions:

@enforce(policy="policy.yaml")
def classify_document(text: str) -> str:
    return classifier.predict(text)

`Enforcer` Class

The programmatic interface for enforcement. Use this when the decorator pattern doesn't fit (e.g., middleware, custom frameworks, dynamic tool dispatch).

Constructor

from enforcecore import Enforcer, Policy

# From YAML file
enforcer = Enforcer.from_file("policy.yaml")

# From Policy object
policy = Policy.from_file("policy.yaml")
enforcer = Enforcer(policy)

Methods

`enforce_sync(func, *args, tool_name, **kwargs) -> T`

Enforces a synchronous function call through the full pipeline.

result = enforcer.enforce_sync(
    my_search_function,
    "market analysis",
    tool_name="search"
)

Parameter	Type	Description
`func`	`Callable`	The function to execute.
`*args`	`Any`	Positional arguments to pass.
`tool_name`	`str \| None`	Override tool name for policy matching.
`**kwargs`	`Any`	Keyword arguments to pass.

Returns: The function's return value (after output redaction). Raises: EnforcementViolation if blocked by policy.

`enforce_async(func, *args, tool_name, **kwargs) -> Any`

Enforces an asynchronous function call. Same semantics as enforce_sync().

result = await enforcer.enforce_async(
    my_async_search,
    "market analysis",
    tool_name="search"
)

`record_cost(cost_usd) -> float`

Records a cost against the session budget. Returns the new cumulative total.

total = enforcer.record_cost(0.02)
print(f"Session total: ${total:.4f}")

Raises: CostLimitError if the cumulative cost exceeds the policy budget.

Properties

Property	Type	Description
`policy`	`Policy`	The loaded policy object.
`policy_name`	`str`	The policy name.
`guard`	`ResourceGuard`	Access the resource guard for manual checks.

Error Handling

Exception Hierarchy

EnforceCoreError (base)
├── PolicyError
│   ├── PolicyLoadError            — file not found, invalid YAML
│   └── PolicyValidationError      — schema validation failed
├── EnforcementViolation           — action blocked by policy
│   ├── ToolDeniedError            — tool in deny list
│   ├── DomainDeniedError          — network domain blocked
│   ├── ContentViolationError      — content rule triggered
│   └── CostLimitError             — cost budget exceeded
├── ResourceLimitError             — time/memory limit exceeded
├── GuardError                     — rate limit, resource guard errors
├── RedactionError                 — PII redaction failure
└── AuditError                     — audit trail write failure

Catching Violations

from enforcecore.core.types import EnforcementViolation, ToolDeniedError

try:
    result = await run_agent("delete all files")
except ToolDeniedError as e:
    print(f"Tool denied: {e}")
except EnforcementViolation as e:
    print(f"Blocked: {e}")
except EnforceCoreError as e:
    print(f"EnforceCore error: {e}")

Configuration

EnforceCore uses environment variables (prefix: ENFORCECORE_) via pydantic-settings:

Variable	Default	Description
`ENFORCECORE_DEFAULT_POLICY`	`None`	Path to default policy file (used by bare `@enforce`).
`ENFORCECORE_AUDIT_ENABLED`	`true`	Enable/disable audit trail.
`ENFORCECORE_AUDIT_PATH`	`./audit_logs`	Audit trail directory.
`ENFORCECORE_REDACTION_ENABLED`	`true`	Enable/disable PII redaction.
`ENFORCECORE_LOG_LEVEL`	`INFO`	Log verbosity.
`ENFORCECORE_FAIL_OPEN`	`false`	If `true`, allow calls on enforcement errors. Never use in production.
`ENFORCECORE_COST_BUDGET_USD`	`None`	Global cost budget for the session.

Utility Functions

`clear_policy_cache() -> int`

Clears the policy file cache. Returns the number of evicted entries.

from enforcecore import clear_policy_cache
evicted = clear_policy_cache()