agent-airlock 0.1.3

The Pydantic-based Firewall for MCP Servers. Stops hallucinated tool calls, validates schemas, and sandboxes dangerous operations.

🛡️ Agent-Airlock

Your AI Agent Just Tried to rm -rf /. We Stopped It.

The open-source firewall for AI agents. One decorator. Zero trust. Full control.

📦 Install • 🚀 Quick Start • 🔌 Frameworks • 📖 Docs • 🤝 Contribute

---

📋 Table of Contents

Click to expand

• 🚨 The Problem
• 🚀 Quick Start
• 🔒 E2B Sandbox
• 📜 Security Policies
• 💰 Cost Control
• 🔐 PII Masking
• ⚡ FastMCP Integration
• 🔌 Framework Compatibility — LangChain, OpenAI, PydanticAI, LlamaIndex, CrewAI, AutoGen, smolagents
• 🏆 Comparison
• 📦 Install
• 🛡️ OWASP Compliance
• 📊 The Numbers
• 📖 Documentation
• 🤝 Contributing
• ⭐ Star History

---

Agent: "I'll help you clean up disk space..."
       ↓
       rm -rf / --no-preserve-root
       ↓
┌─────────────────────────────────────────┐
│  🛡️ AIRLOCK_BLOCK: Operation Denied     │
│                                         │
│  Reason: Matches denied pattern 'rm_*'  │
│  Policy: PRODUCTION_POLICY              │
│  Fix: Use approved cleanup tools only   │
└─────────────────────────────────────────┘

Agent-Airlock is the open-source firewall for MCP servers. One decorator. Zero trust. Full control.

pip install agent-airlock

🚫 Ghost Args Strip hallucinated params

✅ Strict Types No silent coercion

🔄 Self-Healing LLM-friendly errors

🔒 E2B Sandbox Isolated execution

📜 RBAC Role-based access

🔐 PII Mask Auto-redact secrets

---

🚨 The Reality No One Talks About

In January 2026, MCP has 16,000+ servers on GitHub. OpenAI adopted it. The Linux Foundation hosts it.

But here's what the hype cycle ignores:

LLMs hallucinate tool calls. Every. Single. Day.

• Claude invents arguments that don't exist in your function signature
• GPT-4 sends "100" when your code expects 100
• Agents chain 47 tool calls before you notice one deleted production data

The enterprise vendors saw this coming. Prompt Security charges $50K/year. Pangea wants your data flowing through their proxy. Cisco is "coming soon."

We built the alternative.

---

🚀 What This Actually Does

from agent_airlock import Airlock

@Airlock()
def transfer_funds(from_account: str, to_account: str, amount: int) -> dict:
    # Your banking logic here
    return {"status": "transferred", "amount": amount}

That's it. One line. Now your function has:

Protection	What It Stops
Ghost Argument Stripping	LLM sends force=True that doesn't exist → stripped silently
Strict Type Validation	LLM sends amount="500" → blocked, not silently coerced to 500
Self-Healing Errors	Instead of crashing, returns {"fix_hint": "amount must be int"}

The LLM gets a structured error. It retries correctly. Your system stays alive.

---

🔒 When You Need the Big Guns

from agent_airlock import Airlock, STRICT_POLICY

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

sandbox=True means:

• Code executes in an isolated VM (125ms boot time)
• No access to your filesystem, network, or secrets
• Warm pool keeps latency under 200ms after first call

policy=STRICT_POLICY means:

• Rate limited to 100 calls/hour
• Requires agent identity tracking
• Every call logged for audit

---

📜 The Policies You'll Actually Use

from agent_airlock import (
    PERMISSIVE_POLICY,      # Development - no restrictions
    STRICT_POLICY,          # Production - rate limited, requires agent ID
    READ_ONLY_POLICY,       # Analytics agents - can query, can't mutate
    BUSINESS_HOURS_POLICY,  # Dangerous ops only during 9-5
)

Or build your own:

from agent_airlock import SecurityPolicy

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

---

💰 The Cost Problem (And How We Solve It)

A single runaway agent can burn $500 in API costs before you notice.

from agent_airlock import Airlock, AirlockConfig

config = AirlockConfig(
    max_output_chars=5000,    # Truncate before token explosion
    max_output_tokens=2000,   # Hard limit on response size
)

@Airlock(config=config)
def query_logs(query: str) -> str:
    # Even if this returns 10MB of logs,
    # Airlock truncates to 5000 chars before the LLM sees it
    return massive_log_query(query)

Result: Token costs drop dramatically when you:

• Truncate 10MB logs to 5KB before tokenization
• Prevent infinite retry loops from validation errors
• Block runaway agent chains with rate limiting

Math: 10MB = ~2.5M tokens at $0.01/1K = $25 per response. Truncated to 5KB = ~1.25K tokens = $0.01. That's 99.96% reduction per bloated response.

---

🔐 The Security You Forgot You Needed

Your agent just queried a user's profile. The LLM is about to see their SSN.

config = AirlockConfig(
    mask_pii=True,      # SSN, credit cards, phone numbers
    mask_secrets=True,  # API keys, passwords, connection strings
)

@Airlock(config=config)
def get_user(user_id: str) -> dict:
    return db.users.find_one({"id": user_id})

# What the LLM sees:
# {"name": "John", "ssn": "[REDACTED]", "api_key": "sk-...XXXX"}

The data exists in your database. The LLM never sees it. The audit log has the masked version.

---

⚡ FastMCP Integration (The Clean Way)

from fastmcp import FastMCP
from agent_airlock.mcp import secure_tool, STRICT_POLICY

mcp = FastMCP("production-server")

@secure_tool(mcp, policy=STRICT_POLICY)
def delete_user(user_id: str) -> dict:
    """One decorator. MCP registration + Airlock protection."""
    return db.users.delete(user_id)

No ceremony. No boilerplate. The @secure_tool decorator handles:

1. MCP tool registration
2. Ghost argument stripping
3. Type validation
4. Policy enforcement
5. Output sanitization

---

🔌 Framework Compatibility

The Golden Rule: @Airlock must be closest to the function definition.

@framework_decorator    ← Framework sees the secured function
@Airlock()             ← Security layer (innermost)
def my_function():     ← Your code

LangChain / LangGraph

from langchain_core.tools import tool
from agent_airlock import Airlock

@tool
@Airlock()
def search(query: str) -> str:
    """Search for information."""
    return f"Results for: {query}"

# Use with LangGraph ToolNode
from langgraph.prebuilt import ToolNode
tool_node = ToolNode([search])

OpenAI Agents SDK

from agents import Agent, function_tool
from agent_airlock import Airlock, STRICT_POLICY

@function_tool
@Airlock(policy=STRICT_POLICY)
def get_weather(city: str) -> str:
    """Get weather for a city."""
    return f"Weather in {city}: 22°C, Sunny"

agent = Agent(
    name="weather_agent",
    tools=[get_weather],
    model="gpt-4o-mini",
)

PydanticAI

from pydantic_ai import Agent
from agent_airlock import Airlock

# Option 1: Pre-secure, then pass to Agent
@Airlock()
def get_stock_price(symbol: str) -> str:
    return f"Stock {symbol}: $150.25"

agent = Agent("openai:gpt-4o", tools=[get_stock_price])

# Option 2: With @agent.tool_plain
@agent.tool_plain
@Airlock()
def get_forecast(city: str) -> str:
    return f"Forecast for {city}: Sunny"

LlamaIndex

from llama_index.core.tools import FunctionTool
from llama_index.core.agent import ReActAgent
from agent_airlock import Airlock

@Airlock()
def calculate(expression: str) -> int:
    return eval(expression, {"__builtins__": {}})

# Wrap with FunctionTool
calc_tool = FunctionTool.from_defaults(fn=calculate)
agent = ReActAgent.from_tools([calc_tool], llm=llm)

CrewAI

from crewai import Agent, Task, Crew
from crewai.tools import tool
from agent_airlock import Airlock, READ_ONLY_POLICY

@tool
@Airlock(policy=READ_ONLY_POLICY)
def search_documents(query: str) -> str:
    """Search internal documents."""
    return f"Found 5 docs for: {query}"

researcher = Agent(
    role="Researcher",
    tools=[search_documents],
    llm="gpt-4o",
)

AutoGen

from autogen import ConversableAgent
from agent_airlock import Airlock

@Airlock()
def analyze_data(dataset: str, metric: str = "mean") -> str:
    return f"Analysis of {dataset}: {metric}=42.5"

assistant = ConversableAgent(
    name="analyst",
    llm_config={"model": "gpt-4o"},
)
assistant.register_for_llm()(analyze_data)

Hugging Face smolagents

from smolagents import CodeAgent, tool
from agent_airlock import Airlock

@tool
@Airlock(sandbox=True, sandbox_required=True)
def execute_code(code: str) -> str:
    """Execute Python in E2B sandbox."""
    exec(code)
    return "Executed"

agent = CodeAgent(tools=[execute_code], model=model)

Anthropic Claude (Direct API)

import anthropic
from agent_airlock import Airlock

client = anthropic.Anthropic()

@Airlock()
def get_weather(city: str) -> str:
    return f"Weather in {city}: 22°C"

# Register as tool definition
tools = [{
    "name": "get_weather",
    "description": get_weather.__doc__,
    "input_schema": {...}
}]

# Execute with Airlock protection
def handle_tool_call(name, inputs):
    if name == "get_weather":
        return get_weather(**inputs)  # Airlock validates

Full Examples

See the examples/ directory for complete, runnable integrations:

Framework	Example File	Features Demonstrated
LangChain	langchain_integration.py	@tool, AgentExecutor, chains
LangGraph	langgraph_integration.py	StateGraph, ToolNode, multi-agent
OpenAI Agents	openai_agents_sdk_integration.py	Handoffs, manager pattern
PydanticAI	pydanticai_integration.py	Dependencies, structured output
LlamaIndex	llamaindex_integration.py	ReActAgent, QueryEngineTool
CrewAI	crewai_integration.py	Crews, tasks, role-based
AutoGen	autogen_integration.py	ConversableAgent, group chat
smolagents	smolagents_integration.py	CodeAgent, E2B sandbox
Anthropic	anthropic_integration.py	Direct API, streaming

---

🏆 Why Not Just Use [Insert Enterprise Vendor]?

	Prompt Security	Pangea	Agent-Airlock
Pricing	$50K+/year	Enterprise	Free forever
Integration	Proxy gateway	Proxy gateway	One decorator
Self-Healing	No	No	Yes
E2B Sandboxing	No	No	Native
Your Data	Through their servers	Through their servers	Never leaves your infra
Source Code	Closed	Closed	MIT Licensed

We're not anti-enterprise. We're anti-gatekeeping.

Security for AI agents shouldn't require a procurement process.

---

📦 Install

# Core (validation + policies + sanitization)
pip install agent-airlock

# With E2B sandbox support
pip install agent-airlock[sandbox]

# With FastMCP integration
pip install agent-airlock[mcp]

# Everything
pip install agent-airlock[all]

Set your E2B key (if using sandbox):

export E2B_API_KEY="your-key-here"

---

🛡️ OWASP LLM Top 10 Compliance (2025)

Agent-Airlock directly mitigates the top security risks identified by OWASP:

OWASP Risk	Mitigation
LLM01: Prompt Injection	Strict type validation prevents injected payloads from exploiting type coercion
LLM05: Improper Output Handling	PII/secret masking sanitizes outputs before they reach the LLM
LLM06: Excessive Agency	Rate limiting + time restrictions + RBAC prevent runaway agent actions
LLM09: Misinformation	Ghost argument rejection prevents hallucinated parameters from executing

Reference: OWASP Top 10 for LLMs v2025

---

📊 The Numbers

• 187 tests passing
• 84% coverage
• 9 framework integrations — LangChain, LangGraph, OpenAI, PydanticAI, LlamaIndex, CrewAI, AutoGen, smolagents, Anthropic
• <50ms validation overhead
• ~125ms sandbox cold start (E2B Firecracker)
• <200ms sandbox execution (warm pool)
• 0 external dependencies for core functionality

---

📖 Documentation

• Examples — 9 framework integrations with copy-paste code
• Compatibility Guide — Detailed patterns for all major frameworks
• Security Guide — Production deployment checklist
• API Reference — Every function, every parameter

Quick Links

I want to...	Go to
Integrate with LangChain	langchain_integration.py
Use with LangGraph	langgraph_integration.py
Secure OpenAI Agents SDK	openai_agents_sdk_integration.py
Add to PydanticAI	pydanticai_integration.py
Protect LlamaIndex tools	llamaindex_integration.py
Secure CrewAI agents	crewai_integration.py
Use with AutoGen	autogen_integration.py
Integrate smolagents	smolagents_integration.py
Direct Anthropic API	anthropic_integration.py
Build MCP servers	fastmcp_integration.py

---

👤 Who Built This

Sattyam Jain — Building AI infrastructure at scale.

This started as an internal tool after watching an agent hallucinate its way through a production database. Now it's yours.

---

🤝 Contributing

We review every PR within 48 hours.

git clone https://github.com/sattyamjjain/agent-airlock
cd agent-airlock
pip install -e ".[dev]"
pytest tests/ -v

Found a bug? Open an issue. Have a feature idea? Start a discussion.

---

📄 License

MIT. Use it. Fork it. Ship it. No strings.

---

⭐ Star History

---

💖 Support

If Agent-Airlock saved your production database from an LLM hallucination:

• ⭐ Star this repo — It helps others discover the project
• 🐛 Report bugs — Open an issue
• 💡 Request features — Start a discussion
• 🔀 Contribute — PRs reviewed within 48 hours
• 📣 Spread the word — Tweet about it, write a blog post

---

Built with 🛡️ by Sattyam Jain

_{Making AI agents safe, one decorator at a time.}