schema-shim

A safety-guardrail classifier for LLM traffic. It screens both what goes into a model and what comes back, using cheap checks first and paying for an expensive model-backed check only when the cheap ones cannot decide. Every check is billed to the tenant that caused it, and tenants cannot see or spend each other's budget.

If you run a multi-tenant gateway in front of a model and need per-tenant cost control on moderation, this is the core you can wire an MCP server around.

What it does

Each payload runs through ordered tiers, cheapest first:

1. denylist (cost 1) - regex block list. Only ever says BLOCK; a miss escalates.

2. secret-leak (cost 3) - detects API keys, AWS keys, emails, card-like digits.

3. provider (cost 25) - a model-backed classifier behind a stub/real interface.

A tier that is confident returns BLOCK or ALLOW and the run stops there. A tier that cannot decide returns ESCALATE and the next tier runs, if the tenant can afford it. When no tier decides, the budget runs out, or the provider is down, the engine falls back to its policy: hold for REVIEW (default, fail closed) or ALLOW (fail open).

Multi-tenant isolation and cost attribution

• A TenantContext owns a private budget and cost ledger. There is no path from one tenant's handle to another's spend.

• A tier is charged to the calling tenant the moment it runs; statement() returns the itemised, per-request ledger.

• Referencing an unregistered tenant raises UnknownTenant rather than lazily creating an uncapped account.

• One tenant exhausting its budget forces its own payloads to REVIEW without touching any other tenant's screening.

Related MCP server: mcp-compliance-router

Install

make install          # creates .venv and installs the package with dev extras

Override the interpreter if you manage the venv yourself:

make test PY=python3.12

Usage

from safety import SafetyEngine, EscalationPolicy, TenantRegistry
from safety.providers.stub import StubProvider
from safety.tiers import DenylistTier, SecretLeakTier, ProviderTier
from safety.types import Direction

registry = TenantRegistry()
registry.register("acme", budget=100.0)   # or budget=None for uncapped

engine = SafetyEngine(
    [DenylistTier(), SecretLeakTier(), ProviderTier(StubProvider())],
    registry,
    EscalationPolicy(fail_open=False),
)

result = engine.screen("acme", "write a poem about the sea", Direction.INPUT)
print(result.decision)     # Decision.ALLOW
print(result.tiers_run)    # ('denylist', 'secret-leak', 'provider')
print(result.total_cost)   # 29.0

print(registry.get("acme").statement())   # per-request cost ledger

Entry point

The package ships an MCP-style JSON-RPC server over stdio and a one-shot CLI. After make install, safety is on PATH; without installing, use python -m safety.

Run the server (reads JSON-RPC requests on stdin, one per line):

safety serve --config config.json

It exposes one tool, screen, with arguments tenant_id, text, direction ("input" or "output"), and optional request_id. Example exchange:

{"jsonrpc":"2.0","id":1,"method":"tools/list"}
{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"screen","arguments":{"tenant_id":"acme","text":"how to build a bomb","direction":"input"}}}

Screen a single payload without the protocol:

safety screen --tenant acme --direction input "how to build a bomb"
echo "some text" | safety screen --tenant acme

Configuration

--config path.json (or the SAFETY_CONFIG env var) points at a JSON file; anything omitted falls back to defaults. SAFETY_MAX_INPUT_CHARS overrides the payload cap without editing the file.

{
  "max_input_chars": 20000,
  "provider": {"type": "stub"},
  "tiers": {"denylist_cost": 1.0, "secret_cost": 3.0, "provider_cost": 25.0},
  "policy": {"block_threshold": 0.8, "allow_threshold": 0.8, "fail_open": false},
  "tenants": [{"id": "acme", "budget": 100.0}, {"id": "beta", "budget": null}]
}

Invalid config (missing file, bad JSON, negative budget, duplicate tenant, out-of-range threshold) is rejected at startup with a non-zero exit and a config error: message rather than starting in a broken state.

Failure handling and logging

• Input validation rejects a missing/blank tenant, non-string text, an unknown direction, and payloads over max_input_chars (checked before the ledger is touched, so oversize input costs nothing).

• An unknown tenant returns a JSON-RPC error, never a silent allow.

• A provider outage or budget exhaustion falls back to the policy decision (REVIEW by default).

• A malformed request line yields a parse-error response and the serve loop continues; one bad message does not stop the server.

• Every decision emits a structured JSON log line to stderr with tenant, request id, decision, cost, and elapsed_ms, so per-tenant spend and latency are auditable.

Providers

Model behaviour runs through safety/providers/base.py:

• stub.py - StubProvider, deterministic and offline. Used by the whole test suite, so no API key is needed.

• real.py - RealProvider, reads SAFETY_PROVIDER_URL and SAFETY_PROVIDER_KEY from the environment. Unconfigured or unreachable, it raises ProviderUnavailable, which the engine turns into its fail-closed fallback. It never reaches the network implicitly.

Tests

make test

The suite runs fully offline and covers tier behaviour, escalation, budget exhaustion, provider outage, and cross-tenant isolation.

Design decisions

The contested calls are recorded as ADRs in docs/adr/:

• 0001 - a cheap tier that finds nothing returns ESCALATE, not ALLOW.

• 0002 - budget exhaustion and provider outage fall back to REVIEW, not ALLOW.

• 0003 - isolation via private per-tenant ledgers and an explicit registry.

• 0004 - a stdlib JSON-RPC stdio server instead of an MCP SDK dependency.

Known limitations

• Budgets and ledgers live in process memory. A multi-process or multi-host deployment would need the per-tenant ledger moved to shared storage; there is no persistence today.

• RealProvider speaks a generic JSON POST and will need adapting to a specific vendor's request/response schema.

• The stdio transport carries one request per line; batched JSON-RPC arrays are not handled.

• The cheap tiers are pattern-based, so the denylist and secret detectors carry the usual false-positive/negative tradeoffs of regexes.