Interceptors

Every Gavio feature is an interceptor. This page covers the built-ins and how to write your own.

v0.1.0 — PII Guard · Secret scanner · Reliability · Audit · custom scanner · custom interceptor

v0.2.0 (Production core) — Caching · Circuit breaker + load balancer · Governance · Guardrails · Prompt-injection · Tamper-evident audit

v0.3.0 (Observability depth) — Prompt lineage · Prometheus metrics · Risk scoring · Streaming

v0.4.0 — Redis cache backend

v0.5.0 — Cost-optimiser routing

v0.6.0 (Inspector) — see the dedicated Inspector page (F-DX-09/F-DX-10): not an interceptor itself, but it instruments every interceptor in the chain with live span events and a local web UI.

v0.7.0 (Inspector: agentic & production mode) — the Inspector page covers the additions (F-OBS-10/F-DX-11/F-DX-08/F-DX-12): agent call graphs, session views, trace replay, RED stats, hash-chain verification, test-case export, and the gavio inspect --store read-only dashboard.

v0.9.0 — Embedding call guard (F-SEC-10): gw.embed(texts) runs embedding inputs through the same pre-interceptor chain (PII guard included) before the provider's embedding API.

v0.11.0 — Cost Intelligence (F-COST-01/02/04): spend attribution, cost reports, retry overhead, cache savings, and scoped budget fallback.

v0.12.0 — Policy Pack architecture: core/FinTech manifests, detector action metadata, redaction strategies, and custom regex-rule packs.

v0.14.0 — Tool Runtime: schema, freshness, conflict, confidence, and provenance checks for tool calls/results.

---

PII Guard (F-SEC-01)

Scans every request message before the provider call and acts on detected entities. In redact mode the originals are restored in the response.

Built-in tier-1 scanners (all validated, not just regex):

Scanner	Entity	Validation
Email	EMAIL	RFC-5322-ish
IBAN	IBAN	ISO 13616 mod-97 checksum
BSN	BSN	Dutch 11-proef
CreditCard	CREDIT_CARD	Luhn
Phone	PHONE	E.164 + national
IpAddress	IP_ADDRESS	IPv4/IPv6 parse
SSN	SSN	US format

Modes: redact (typed placeholder [EMAIL_1], restored on response) · mask (****) · tag (<EMAIL>…</EMAIL>) · block (raise & refuse).

Sensitivity: strict (all matches) · balanced (confidence ≥ 0.6) · permissive (≥ 0.9).

Overlapping matches are resolved greedily (prefer the longer / higher-confidence match). Only entity types and counts are logged — never raw values.

from gavio.interceptors.pii import PiiGuard, Sensitivity, PiiMode
from gavio.interceptors.pii.scanners import EmailScanner, IbanScanner

PiiGuard(
    scanners=[EmailScanner(), IbanScanner()],   # omit to use the default set
    sensitivity=Sensitivity.STRICT,
    mode=PiiMode.REDACT,
    restore_on_response=True,
)

Domain policy packs (F-PACK-01/02/05)

Generic PII detection misses domain-specific identifiers. Policy packs wrap scanner sets in a manifest: pack id/name/version/domain, detectors, default action, redaction strategy, audit labels, and scanner composition. The scanners still plug into PiiGuard unchanged.

Built-ins:

• Core PII (gavio.core-pii) backs the default scanner set.
• FinTech (gavio.fintech) promotes the v0.10.0 scanner prototype into a first-class pack for SWIFT/BIC and US ABA routing numbers.

from gavio.interceptors.pii import (
    PiiGuard,
    RegexPolicyRule,
    core_policy_pack,
    custom_policy_pack,
    fintech_policy_pack,
    policy_pack_scanners,
)

pack = fintech_policy_pack()
PiiGuard(scanners=policy_pack_scanners(core_policy_pack(), pack))

custom = custom_policy_pack(
    id="acme.internal",
    name="Acme Internal IDs",
    rules=[
        RegexPolicyRule(
            name="employee_id",
            entity_type="EMPLOYEE_ID",
            pattern=r"\bEMP-[0-9]{6}\b",
            confidence=0.88,
            replacement_prefix="EMPLOYEE_ID",
            action="flag",
            redaction_strategy="hash",
            label="INTERNAL_IDENTIFIER",
        )
    ],
    default_action="flag",
    redaction_strategy="hash",
    audit_labels=["INTERNAL_IDENTIFIER"],
)

Existing factory APIs remain compatible: default_scanners() / fintech_scanners() in Python, defaultScanners() / fintechScanners() in JavaScript, and DefaultScanners.defaults() / DefaultScanners.fintech() in Java are now backed by the pack objects.

---

Tool Runtime (F-TOOL-01/02/03/04)

Tool Runtime validates structured tool context before tool results re-enter model context. It reads metadata.tools.calls[], validates declared input/output schemas, checks freshness with per-call or context TTLs, detects conflicts across configured result keys, computes confidence, and records provenance under runtime context plus Inspector decision state.

Tool Runtime v2 also accepts a lightweight registry under metadata.tools.definitions[], global grants under metadata.tools.permissions, approval records under metadata.tools.approvals[], and replay records under metadata.tools.records[]. Definitions can declare required permissions, risk, approval requirements, freshness TTLs, provenance requirements, and MCP metadata without coupling Gavio to a specific MCP implementation.

from gavio.interceptors.tool_runtime import ToolRuntimeInterceptor

gw = Gateway.builder().use(
    ToolRuntimeInterceptor(
        on_failure="error",
        max_age_seconds=120,
        conflict_keys=["delivery_date"],
    )
).build()

JavaScript imports toolRuntime from gavio/interceptors/tool-runtime; Java uses io.gavio.interceptors.toolruntime.ToolRuntimeInterceptor.

from gavio.interceptors.tool_runtime import analyze_tool_runtime, replay_tool_runtime

tools = {
    "permissions": ["read.billing"],
    "definitions": [
        {
            "name": "lookup_invoice",
            "permissions": ["read.billing"],
            "risk": "low",
            "provenance_required": True,
            "mcp": {"server": "billing-mcp", "tool": "lookup_invoice"},
        }
    ],
    "calls": [
        {
            "id": "invoice-1",
            "name": "lookup_invoice",
            "result": {"status": "paid"},
            "mcp": {"server": "billing-mcp", "tool": "lookup_invoice"},
        }
    ],
}

decision = analyze_tool_runtime(tools)
replayed = replay_tool_runtime({**tools, "records": tools["calls"]})

---

Secret scanner (F-SEC-04)

Part of the default scanner set. Detects OpenAI (sk-…) / Anthropic (sk-ant-…) keys, AWS access keys (AKIA…), GitHub tokens, JWTs, PEM private keys, and DB connection strings (postgres://…, etc.) — all as entity type SECRET.

---

Reliability

All three implement ExecutorPolicy and wrap the provider call (first-registered = outermost — see architecture).

Policy	ID	Key options
RetryInterceptor	F-REL-01	max_attempts, base_delay_ms, max_delay_ms, jitter, retry_on
FallbackChain	F-REL-02	list of fallback adapters, tried in order on ProviderError
TimeoutPolicy	F-REL-07	timeout_seconds — raises TimeoutError on breach

Retry backs off exponentially with full jitter and only retries transient errors (RateLimitError, TimeoutError, ServerError, ProviderUnavailableError).

---

Audit (F-OBS-01)

Register it outermost so its after runs last and sees the final response. It builds an AuditRecord (metadata + SHA-256 hashes of the redacted prompt and the response — never raw text) and writes it to a sink. v0.1.0 ships StdoutSink (F-OBS-05); implement AuditSink.write(record) for your own (Elasticsearch, Kafka, …).

[gavio:audit] trace=019f1ac0-b801-7f48… mock/mock tokens=16 cost=$0.000000 \
              latency=0ms cache=miss pii=EMAIL,IBAN interceptors=[audit,pii_guard]

---

Writing a custom scanner

Implement the PiiScanner interface and add it to PiiGuard. Use ctx.next_index(type) for stable placeholder numbering.

import re
from gavio.interceptors.pii.scanner import PiiScanner
from gavio.interceptors.pii.match import PiiMatch

class IngAccountScanner(PiiScanner):
    entity_type = "ING_ACCOUNT"
    _PATTERN = re.compile(r"\bNL\d{2}INGB\d{10}\b")

    def scan(self, text, ctx):
        return [
            PiiMatch(
                entity_type=self.entity_type,
                start=m.start(), end=m.end(), value=m.group(),
                replacement=f"[ING_ACCOUNT_{ctx.next_index(self.entity_type)}]",
            )
            for m in self._PATTERN.finditer(text)
        ]

The JavaScript and Java equivalents use a factory object / class implementing the same interface — see the Python, JavaScript, or Java guide.

---

Writing a custom interceptor

Implement name plus any of before / after / on_error:

from gavio.interceptors.base import Interceptor

class HeaderTagger(Interceptor):
    name = "header-tagger"

    async def before(self, request, ctx):
        ctx.mark_fired(self.name)
        request.metadata["tagged_at_layer"] = "edge"
        return request

For something that must retry or wrap the provider call, subclass ExecutorPolicy and implement around(request, ctx, call_next) instead.

---

v0.2.0 — Production core interceptors

The following ship in v0.2.0 across all three SDKs. API names below use the Python style; JavaScript uses camelCase factory functions (semanticCache(), circuitBreaker(), …) and Java uses builders (SemanticCache.builder(), CircuitBreaker.builder(), …).

Caching (F-CACHE-01/02/03/04)

SemanticCache is an ExecutorPolicy — a hit returns the cached response and skips the provider. Register it outermost. Exact SHA-256 tier is always on; pass an embedder to enable the semantic (cosine-similarity) tier.

from gavio.interceptors.cache import SemanticCache, HashingEmbedder

SemanticCache(
    embedder=HashingEmbedder(),        # omit for exact-only
    similarity_threshold=0.95,
    exact_ttl_seconds=3600,
)

A cache hit sets cache_hit / cache_type on the response and composes with PiiGuard (PII is still restored). The HashingEmbedder is zero-dependency; plug in a real embedder implementing the Embedder protocol for production.

Redis backend (F-CACHE-04)

The default MemoryBackend/InMemoryVectorBackend are per-process; swap in a Redis-backed pair to share cache hits across processes/instances. Entries are namespaced under a Redis Set index, so clear() only removes keys that backend itself wrote, never the whole database.

from gavio.interceptors.cache import SemanticCache, HashingEmbedder
from gavio.interceptors.cache.backends import RedisBackend, RedisVectorBackend

SemanticCache(
    backend=RedisBackend(url="redis://localhost:6379"),
    embedder=HashingEmbedder(),
    vector_backend=RedisVectorBackend(url="redis://localhost:6379"),
)

Python requires the optional redis package (pip install gavio[redis]); JavaScript (redisCacheBackend()/redisVectorBackend() from gavio/interceptors/cache/backends) and Java (new RedisCacheBackend(url) / new RedisVectorBackend(url) in gavio-interceptor-cache) hand-roll a minimal RESP2 client over node:net/java.net.Socket — zero runtime dependencies, no extra install needed. Also wired into the Python and JavaScript config loaders (semantic_cache.backend: redis).

Reliability — circuit breaker + load balancer (F-REL-03/04)

from gavio.interceptors.reliability import CircuitBreaker, LoadBalancer

CircuitBreaker(failure_threshold=5, recovery_timeout_seconds=30)  # fast-fails while open
LoadBalancer([adapter_a, adapter_b], weights=[2, 1])              # weighted round-robin

Both are ExecutorPolicy. The breaker opens after N consecutive provider errors and fast-fails with CircuitOpenError; the balancer distributes across a pool of provider adapters.

Governance (F-GOV-02/03/04)

from gavio.interceptors.governance import CostControl, RateLimiter, ModelPolicy

CostControl(hard_cap_usd=50, soft_cap_usd=10, scope="agent", window="day")   # budget
RateLimiter(max_requests_per_minute=60, max_tokens_per_minute=100_000)       # rate limit
ModelPolicy(roles={"analyst": ["gpt-4o-mini"], "admin": ["*"]})              # RBAC

Budget blocks with BudgetExceededError, rate limiting with RateLimitExceededError, and RBAC with ModelNotAllowedError (role read from request.metadata["role"]).

In v0.11.0, CostControl can scope spend by global, agent, session, model, tenant, feature, or user. Tenant/feature/user scopes are read from request.metadata.costDimensions (or flat aliases like tenant_id and featureId). A hard cap still blocks by default, or can fall back to a cheaper model:

CostControl(
    hard_cap_usd=50,
    soft_cap_usd=40,
    scope="tenant",
    window="month",
    fallback_model="gpt-4o-mini",
)

Soft warnings and hard-cap fallback/block decisions surface as Inspector governance.event records with kind="budget".

Cost Governance v2 (v1.2.0) adds a canonical BudgetPolicy / BudgetDecision contract for production budget workflows. BudgetPolicyControl evaluates projected spend against a scoped policy and can allow, warn, block, fall back, downgrade, or run in dry-run mode while recording the decision as Inspector metadata.

from gavio.interceptors.governance import BudgetPolicy, BudgetPolicyControl

BudgetPolicyControl(
    BudgetPolicy(
        id="tenant-monthly",
        scope_type="tenant",
        scope_value="acme",
        window="monthly",
        limit_usd=500,
        hard_limit_action="fallback",
        fallback_model="gpt-4o-mini",
    ),
    estimated_request_cost_usd=0.02,
)

Cost-optimiser routing (F-GOV-06)

CostRouter reroutes a request to a cheaper simple_model when a pluggable ComplexityScorer scores its prompt below complexity_threshold. The default HeuristicComplexityScorer is zero-dependency — prompt length (via the same token estimator PricingProvider uses) plus reasoning-keyword density (why, compare, trade-off, explain, …).

from gavio.interceptors.governance import CostRouter

CostRouter(simple_model="gpt-4o-mini", complexity_threshold=0.35)

Register it early, before caching, so a rerouted request's cache key reflects the model it actually ran on. Records its decision (rerouted, original_model, complexity_score) in ctx.state["cost_router"] — the audit trail's model field already reflects the rerouted model once it runs, so no schema changes were needed.

Guardrails (F-QUA-01/02)

from gavio.interceptors.guardrails import GuardrailsInterceptor
from gavio.interceptors.guardrails.validators import (
    JsonSchemaValidator, RegexDenylistValidator, RegexAllowlistValidator,
)

GuardrailsInterceptor(
    validators=[JsonSchemaValidator({"type": "object", "required": ["answer"]}),
                RegexDenylistValidator([r"(?i)competitor"])],
    on_failure="error",   # error | retry | warn
)

An ExecutorPolicy that validates the response; on failure it raises GuardrailViolationError, retries the provider, or warns. Records guardrail_outcome for the audit trail.

Prompt-injection defense (F-SEC-05)

from gavio.interceptors.injection import PromptInjectionGuard

PromptInjectionGuard(action="block")   # or "flag"

Scans user/tool messages against a curated pattern corpus (Python/JS also support an optional semantic tier via an embedder); blocks with PromptInjectionError or flags by setting risk_score.

Tamper-evident audit + multi-agent trace (F-OBS-02/03)

from gavio.interceptors.audit import AuditInterceptor, verify_chain, build_call_graph

AuditInterceptor(hash_chain=True)   # links records via previous_hash
verify_chain(records)               # detects any edit/reorder/deletion
build_call_graph(records)           # reconstruct the multi-agent DAG

---

v0.3.0 — Observability depth

Ships in v0.3.0 across all three SDKs. Python style shown; JavaScript uses camelCase factories (metricsInterceptor(), riskScorer()) and Java uses classes / builders (new MetricsInterceptor(), new RiskScorer()).

Prompt lineage (F-OBS-04)

Attach the provenance of a prompt — the template, the variables bound into it, and the RAG chunk sources — to a request. The audit interceptor copies it onto the AuditRecord so any prompt can be reconstructed and debugged. RAG chunk text is never stored — only source references — so the record stays metadata-only.

from gavio import PromptLineage, RagChunk

await gw.complete(
    messages=[...],
    lineage=PromptLineage(
        template_id="support-reply",
        template_version="v3",
        variables={"customer": "Ada", "tier": "gold"},
        rag_chunks=[RagChunk(source="doc://kb/refunds", chunk_id="c1", score=0.92)],
    ),
)

Lineage participates in the hash-chain contentHash(). Also threaded through Gateway.stream(...).

Prometheus metrics (F-OBS-08)

MetricsInterceptor records per-request metrics into a PrometheusMetrics registry; scrape it via render() for the Prometheus text exposition format — no client library, zero dependencies.

from gavio.interceptors.metrics import MetricsInterceptor

metrics = MetricsInterceptor()
gw = Gateway.builder().dev_mode(True).use(metrics).build()
# ...
print(metrics.metrics.render())

Emits gavio_requests_total, gavio_tokens_total{kind}, gavio_cost_usd_total, gavio_request_latency_ms (histogram), and gavio_cache_hits_total — all labelled by provider and model. (Python gavio.interceptors.metrics; JS gavio/interceptors/metrics; Java module gavio-interceptor-metrics.)

Risk scoring (F-QUA-06)

RiskScorer folds the per-request signals other interceptors leave on the context — PII entities found, guardrail outcome (FAIL/HITL), and the prompt-injection risk — into a single composite score in [0, 1], written to ctx.risk_score and recorded on the AuditRecord.

from gavio.interceptors.quality import RiskScorer, RiskWeights

RiskScorer()                                   # default weights
RiskScorer(RiskWeights(pii=0.3, guardrail=0.4, injection=0.3, pii_saturation=4))

Register it inside the audit interceptor so audit sees the composite. Weights are configurable and the composite is clamped. (New quality family: Python gavio.interceptors.quality; JS gavio/interceptors/quality; Java module gavio-interceptor-quality.)

Streaming — StreamBuffer (F-REL-06)

Gateway.stream(...) drives the provider's streaming API but buffers the response in full (via StreamBuffer) before the post-interceptor pipeline runs — so guardrails, PII restore, and audit see, and can rewrite or block, the complete response before any chunk reaches the caller.

async for chunk in gw.stream(messages=[{"role": "user", "content": "hi"}]):
    print(chunk, end="")

Executor policies (retry, circuit breaker, cache) are not applied to the streaming path in this release.

v0.10.0 — Compliance & policy packs

License / copyright detection (F-QUA-10)

licenseDetector is a guardrails OutputValidator that flags known open-source license text (MIT, Apache-2.0, GPL-2.0/3.0, BSD-3-Clause, MPL-2.0) in a model response before it lands in user code. It matches a shipped corpus of hashed 8-word shingles — only hashes ship, never license text — and drops shingles shared by more than one license so a hit is discriminative (GPL-2.0 and GPL-3.0 don't cross-fire). Detections surface in the guardrail outcome and audit record.

from gavio.interceptors.guardrails import GuardrailsInterceptor, LicenseDetectorValidator

GuardrailsInterceptor([LicenseDetectorValidator()], on_failure="warn")

Right to erasure — GDPR Art. 17 (F-QUA-09)

Pass a subject_id in request metadata and it is persisted on every AuditRecord. Persistent sinks expose purge(subject_id), which removes every matching record and returns the count erased — a no-op for non-persistent sinks (stdout), a real implementation on the built-in JsonlSink (now shipped in all three SDKs).

await gateway.complete(messages=[...], metadata={"subject_id": "user-123"})
removed = await sink.purge("user-123")   # -> int records erased

Scope: erasure covers records in a persistent, purgeable sink; caches and downstream copies (log shippers, SIEMs, backups) are out of scope this release.

Drift detection (F-GOV-07)

DriftMonitor watches a provider's response distribution and alerts when a signal (latency, tokens, cost, risk) shifts away from its recent baseline. The default StatisticalDriftDetector keeps a rolling window per metric and flags a sample beyond a z-score threshold; supply your own DriftDetector to change the algorithm. Alerts are observe-only — each surfaces as a governance.event inspector event and is counted in driftAlerts on /api/stats.

from gavio.interceptors.governance import DriftMonitor

DriftMonitor(metrics=["latency_ms", "total_tokens"], window_size=50, threshold=3.0)

Image PII detection (F-SEC-09)

ModalityGuard extends the PII pipeline to image inputs passed on the side-channel images request field. Each ModalityScanner extracts text (OCR) and/or direct detections (e.g. faces); the extracted text runs through the standard text scanners, and detections land in the AuditRecord's pii_entity_types like text PII. Ships a reference OcrModalityScanner behind an optional dependency (Python [ocr] extra, JS tesseract.js, Java tess4j).

from gavio.interceptors.pii import ModalityGuard, OcrModalityScanner

ModalityGuard(scanners=[OcrModalityScanner()], on_detect="tag")  # or "block"
await gateway.complete(messages=[...], images=[png_bytes])

Domain policy packs — FinTech

Policy packs are scanner sets for an industry, composed with the default set. The FinTech pack adds financial identifiers beyond the core IBAN scanner: a context-gated SWIFT/BIC scanner (SWIFT_BIC) and a US ABA routing number scanner (ROUTING_NUMBER, mod-10 checksum).

from gavio.interceptors.pii import PiiGuard, default_scanners, fintech_scanners

PiiGuard(scanners=[*default_scanners(), *fintech_scanners()])

(JS: fintechScanners(); Java: DefaultScanners.fintech().)