Agent Airlock

A type-checker and contract layer for AI agent tool calls — deny-by-default, in-process, zero-dep

Strict validation, ghost-argument stripping, and self-healing retries — one decorator, any agent or MCP server.

[](https://pypi.org/project/agent-airlock/) [](https://pypistats.org/packages/agent-airlock) [](https://github.com/sattyamjjain/agent-airlock/actions/workflows/ci.yml) [](https://codecov.io/gh/sattyamjjain/agent-airlock)

[](https://www.python.org/downloads/) [](https://opensource.org/licenses/MIT) [](https://github.com/sattyamjjain/agent-airlock/stargazers) [](http://makeapullrequest.com)

Test suite: 3,280 tests · Coverage: 85.32% · v0.8.39

[Get Started in 30 Seconds](#-30-second-quickstart) · [Why Airlock?](#-the-problem-no-one-talks-about) · [All Frameworks](#-framework-compatibility) · [Benchmark](BENCHMARK.md) · [Cross-tool comparison](benchmarks/blockrate/RESULTS.md) · [Least-Privilege Benchmark](benchmarks/toolprivbench/RESULTS.md) · [Docs](#-documentation)

> Benchmarks — deterministic, in-process, zero-dep. Guard-suite block-rate + the cross-tool comparison vs Meta LlamaFirewall & Invariant Guardrails (block-rate + p50/p95 latency contrasted with model-in-the-loop approaches; incumbents cited, not re-run — no fabricated competitor number): [BENCHMARK.md](BENCHMARK.md), full results + latency in [benchmarks/blockrate/RESULTS.md](benchmarks/blockrate/RESULTS.md) (reproduce: python -m benchmarks.blockrate). Least-privilege / over-privileged-tool-selection block-rate vs ToolPrivBench (arXiv:2606.20023), OWASP-Agentic-mapped: [benchmarks/toolprivbench/RESULTS.md](benchmarks/toolprivbench/RESULTS.md).

---

┌────────────────────────────────────────────────────────────────┐
│  🤖 AI Agent: "Let me help clean up disk space..."            │
│                           ↓                                    │
│               rm -rf / --no-preserve-root                      │
│                           ↓                                    │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  🛡️ AIRLOCK: BLOCKED                                     │  │
│  │                                                          │  │
│  │  Reason: Matches denied pattern 'rm_*'                   │  │
│  │  Policy: STRICT_POLICY                                   │  │
│  │  Fix: Use approved cleanup tools only                    │  │
│  └──────────────────────────────────────────────────────────┘  │
└────────────────────────────────────────────────────────────────┘

---

🎯 30-Second Quickstart

pip install agent-airlock

from agent_airlock import Airlock

@Airlock()
def transfer_funds(account: str, amount: int) -> dict:
    return {"status": "transferred", "amount": amount}

# LLM sends amount="500" (string) → BLOCKED with fix_hint
# LLM sends force=True (invented arg) → STRIPPED silently
# LLM sends amount=500 (correct) → EXECUTED safely

That's it. Your function now has ghost argument stripping, strict type validation, and self-healing errors.

---

🧠 The Problem No One Talks About

The Hype

> "MCP has 16,000+ servers on GitHub!" > "OpenAI adopted it!" > "Linux Foundation hosts it!"

The Reality

LLMs hallucinate tool calls. Every. Single. Day.

• Claude invents arguments that don't exist
• GPT-4 sends "100" when you need 100
• Agents chain 47 calls before one deletes prod data

Enterprise solutions exist: Prompt Security ($50K/year), Pangea (proxy your data), Cisco ("coming soon").

We built the open-source alternative. One decorator. No vendor lock-in. Your data never leaves your infrastructure.

---

✨ What You Get

Ghost Args Strip LLM-invented params

Strict Types No silent coercion

Self-Healing LLM-friendly errors

E2B Sandbox Isolated execution

RBAC Role-based access

PII Mask Auto-redact secrets

Network Guard Block data exfiltration

Path Validation CVE-resistant traversal

Circuit Breaker Fault tolerance

OpenTelemetry Enterprise observability

Cost Tracking Budget limits

Vaccination Auto-secure frameworks

---

📋 Table of Contents

Click to expand full navigation

• [30-Second Quickstart](#-30-second-quickstart)
• [The Problem](#-the-problem-no-one-talks-about)
• [What You Get](#-what-you-get)
• [Core Features](#-core-features)
• [E2B Sandbox](#-e2b-sandbox-execution)
• [Security Policies](#-security-policies)
• [Cost Control](#-cost-control)
• [PII Masking](#-pii--secret-masking)
• [Network Airgap](#-network-airgap-v030)
• [Framework Vaccination](#-framework-vaccination-v030)
• [Circuit Breaker](#-circuit-breaker-v040)
• [OpenTelemetry](#-opentelemetry-observability-v040)
• [Framework Compatibility](#-framework-compatibility)
• [FastMCP Integration](#-fastmcp-integration)
• [Comparison](#-why-not-enterprise-vendors)
• [Installation](#-installation)
• [OWASP Compliance](#️-owasp-compliance)
• [Performance](#-performance)
• [Documentation](#-documentation)
• [Contributing](#-contributing)
• [Support](#-support)

---

🔥 Core Features

🔒 E2B Sandbox Execution

from agent_airlock import Airlock, STRICT_POLICY

@Airlock(sandbox=True, sandbox_required=True, policy=STRICT_POLICY)
def execute_code(code: str) -> str:
    """Runs in an E2B Firecracker MicroVM. Not on your machine."""
    exec(code)
    return "executed"

| Feature | Value | |---------|-------| | Boot time | ~125ms cold, str: exec(code) return "executed"

**Isolation model — read before you reach for `cap_drop`.** Modal
sandboxes run under **gVisor** (kernel-syscall filtering), not under
Docker-style capability dropping. The Modal Python SDK does not
expose `cap_drop` / `cap_add` / `seccomp` / `no-new-privileges` —
there is no equivalent knob to map. If your threat model needs
Linux-capability dropping at the container layer, keep using
`DockerBackend`. The network posture *is* configurable: `ModalBackend`
defaults to `block_network=True` (deny-by-default), and a supplied
`NetworkPolicy` maps to Modal's `block_network` flag (`allow_egress=False`
→ blocked, `True` → allowed). Hostname allowlists in `NetworkPolicy.allowed_hosts`
do **not** forward to Modal (their API is CIDR-only); the backend logs
a structlog warning and the operator is expected to re-state hostname
constraints at the `Airlock` policy layer.

`ModalBackend` is **opt-in only** — it is NOT added to the
`get_default_backend()` priority chain (E2B → Docker → Local stays the
default flow). Existing callers see no behavior change.

---

### 📜 Security Policies

| Preset | Use case | Key posture |
|---|---|---|
| `PERMISSIVE_POLICY` | Dev / sandbox | No restrictions |
| `STRICT_POLICY` | Prod | Rate-limited, requires agent identity, denies dangerous capabilities |
| `READ_ONLY_POLICY` | Analytics / RAG | `read_*` / `get_*` / `list_*` / `search_*` only |
| `BUSINESS_HOURS_POLICY` | Compliance windows | `delete_*` / `drop_*` / `*_production` only 09:00–17:00 |
| `CAMOUFLAGE_RESISTANT_POLICY` *(v0.8.6)* | Detector-independent defense vs. domain-camouflaged injection | Deny-by-default allowlist, ghost-arg BLOCK, output cap, per-call reauthorization |

```python
from agent_airlock import (
    PERMISSIVE_POLICY,
    STRICT_POLICY,
    READ_ONLY_POLICY,
    BUSINESS_HOURS_POLICY,
    CAMOUFLAGE_RESISTANT_POLICY,  # v0.8.6
)

# Or build your own:
from agent_airlock import SecurityPolicy

MY_POLICY = SecurityPolicy(
    allowed_tools=["read_*", "query_*"],
    denied_tools=["delete_*", "drop_*", "rm_*"],
    rate_limits={"*": "1000/hour", "write_*": "100/hour"},
    time_restrictions={"deploy_*": "09:00-17:00"},
)

CAMOUFLAGE_RESISTANT — detector-independent injection defense (v0.8.6)

arXiv:2605.22001 ("Blind Spots in the Guard", Pai, May 2026) shows that production injection detectors — Llama Guard 3 included — drop to IDR = 0.000 on payloads that mimic the target document's domain vocabulary and authority structure. Per the paper, detection rates collapse from 93.8% to 9.7% on Llama 3.1 8B and from 100% to 55.6% on Gemini 2.0 Flash.

CAMOUFLAGE_RESISTANT_POLICY does not rely on payload-content signatures at all. It blocks at four structural seams an attacker has to ride regardless of phrasing:

1. Deny-by-default tool allowlist. Empty allowed_tools means

nothing is callable; deployments opt every tool in by name. A camouflaged directive targeting an unlisted tool is blocked on allowlist grounds without ever invoking a detector.

1. Ghost-argument BLOCK. A camouflaged directive cannot smuggle

undeclared parameters past validation.

1. Hard output cap + sanitization. Tool output that re-enters the

model context is truncated and PII/secret-masked so a camouflaged directive embedded in tool output can't carry into a downstream agent at full length.

1. Per-call reauthorization (debate-amplification guard). Once a

tool's output has flowed back into the model, any reinvocation requires an explicit context.authorize_once(tool) grant from the harness — breaking the multi-agent fan-out path the paper identifies.

from agent_airlock import Airlock, apply_camouflage_resistant

bundle = apply_camouflage_resistant(allowed_tools=["read_file", "search"])

@Airlock(config=bundle.config, policy=bundle.policy)
def read_file(path: str) -> str:
    ...

apply_camouflage_resistant() composes the matching AirlockConfig (unknown-args BLOCK, sanitization on, output cap 4000 chars) with a SecurityPolicy carrying your explicit allowlist. The preset is deliberately incomplete on its own — the config-level knobs and the policy-level knobs span two seams, so the factory returns both as a CamouflageResistantBundle.

> Running an MCP server with STDIO transport? Also wire the > [Ox MCP STDIO sanitizer](#️-owasp-compliance) via > stdio_guard_ox_defaults() — it blocks the entire > CVE-2026-30616 class (shell metacharacter injection, > non-allowlisted binaries, Trojan-Source RTL overrides, and > inline-code flags) before subprocess.Popen.

---

🪪 MCP server attestation (v0.8.10)

arXiv:2605.24248 ("Attested Tool-Server Admission", Metere, May 2026) calls out a gap MCP itself does not close: the protocol standardises message exchange between LLM agents and tool servers but says nothing about trust. Anybody who can answer on the wire can declare themselves a tool server.

mcp_attested_admission_defaults() is a deny-by-default opt-in preset that closes the gap host-side, mirroring the paper's three additive mechanisms:

1. Offline-signed clearance assertion. Before any tool from an MCP

server is dispatched, the host fetches a JWS-compact clearance from {server_url}/.well-known/mcp-clearance (path is configurable) and verifies its signature against an operator-pinned trust root. The trust root is supplied to AttestedAdmissionConfig at process startup — never network-fetched on the hot path.

1. Deny-by-default per-server tool allowlist. Admitting a server

is not the same as trusting its every tool. The verified clearance carries an explicit list of tool names the host will permit; everything else is denied. The sub claim is matched against the server identity the host is about to dispatch to (so a stolen clearance from server A can't admit a tool call to server B).

1. Flavor-gated enforcement. ENFORCE (default) hard-denies on

missing / invalid / expired clearance; WARN logs and admits — the staged turn-up an operator wants when introducing the gate against real traffic.

Every admission decision emits a [ReceiptVerdict](./docs/attest/receipt.md) on the guard="mcp_attested_admission" channel, so the existing airlock attest DSSE pipeline picks decisions up unchanged — this preset does not invent a new log.

from agent_airlock.mcp_proxy_guard import MCPProxyConfig, MCPProxyGuard
from agent_airlock.mcp_spec.attested_admission import TrustRoot
from agent_airlock.policy_presets import mcp_attested_admission_defaults

# Operator pins the trust root at startup. Never fetched at runtime.
with open("/etc/airlock/mcp-clearance-root.pem", "rb") as fh:
    pinned_pem = fh.read()

cfg = mcp_attested_admission_defaults(
    trust_root=TrustRoot(key_id="ops-2026Q2", ed25519_pem=pinned_pem),
    enforcement_mode="ENFORCE",       # deny-by-default
    max_clearance_age_days=30,
)
guard = MCPProxyGuard(MCPProxyConfig(attested_admission=cfg))

decision = guard.audit_tool_admission(
    server_url="https://mcp.example.com",
    server_id="srv-alpha",            # expected `sub` claim
    tool_name="read",
)
if not decision.admitted:
    raise RuntimeError(decision.reason)

Signature verification needs the [attested] extra (pulls in cryptography for offline Ed25519 / RSA-PSS / JWKS verification); the base install stays zero-runtime-dep.

> Install with pip install "agent-airlock[attested]". Opt-in only — > existing callers that don't set attested_admission get exactly > v0.8.9 behavior.

---

🧭 Behavioral sequence guard (v0.8.12)

Watches the ordered stream of tool calls in a session and flags divergence from a declared expected order — not the model's stated reasoning trace.

arXiv:2605.27901 ("The Fragility of Chain-of-Thought Monitoring", Onyame, Zhou, Thopalli, Kailkhura & Agarwal, May 2026) reports an average 95.9% CoT unfaithfulness across 8B–120B-parameter models — including answer-switching, post-hoc rationalisation, and procedural exploitation of hints. Trusting the model's stated reasoning to detect misbehavior is therefore not viable. Trusting its behavior — the sequence of tools it actually invokes — is.

SequenceGuard is an opt-in field on SecurityPolicy that runs in the @Airlock seam right after the standard policy check, in two modes:

DECLARED mode — operator supplies a permitted-transition DAG. Any transition not in the DAG is a SequenceViolation. Deny-by-default.

from agent_airlock import Airlock, SecurityPolicy
from agent_airlock.sequence_guard import SequenceGuard, ENTRY_SENTINEL

policy = SecurityPolicy(
    sequence_guard=SequenceGuard(
        mode="declared",
        action="block",                       # or "warn"
        dag={
            ENTRY_SENTINEL: {"read"},         # only `read` may start a session
            "read": {"read", "summarize"},    # after read, either re-read or summarize
            "summarize": {"send"},            # after summarize, only send
            "send": set(),                    # send is terminal
        },
    ),
)

BASELINE mode — guard maintains a per-session-key Markov transition profile in a local JSON file (no cloud, no PII — only tool names and SHA-256 shape hashes of (arg types, kwarg names+types), never argument values) and flags transitions with observed P(curr | prev) Strictly opt-in. The new SecurityPolicy.sequence_guard field > defaults to None`; callers that don't set it get exactly v0.8.11 > behavior. Zero new runtime deps — Pydantic-only core stays intact.

---

🛑 Action-time contradiction gate (v0.8.15)

arXiv:2605.27157 ("Detecting Is Not Resolving: The Monitoring Control Gap in Retrieval Augmented LLMs", Yu et al., 2026) shows that LLMs readily acknowledge contradictory evidence in their reasoning trace yet "this awareness fails to constrain their final recommendations". The deficit is at action selection — single-turn diagnostics overestimate RAG safety, and detection alone is not a control.

ActionContradictionGate is an opt-in policy hook that wraps three pluggable detectors (any one trips) and a privileged-sink glob set. When a detector trips AND the dispatched tool m

…

Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: sattyamjjain
• Source: sattyamjjain/agent-airlock
• License: MIT

Install and usage instructions live in the source repository linked above.