### 开源 AI 代理防火墙
**一个装饰器。零信任。完全控制。**
[](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)
[**30 秒快速入门**](#-30-second-quickstart) · [**为什么选择 Airlock**](#-the-problem-no-one-talks-about) · [**所有框架**](#-framework-compatibility) · [**文档**](#-documentation)
```
┌────────────────────────────────────────────────────────────────┐
│ 🤖 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 秒快速启动
```
pip install agent-airlock
```
```
from agent_airlock import Airlock
@Airlock()
def transfer_funds(account: str, amount: int) -> dict:
return {"status": "transferred", "amount": amount}
# LLM 发送 amount="500"（字符串）→ 因 fix_hint 被阻止
# LLM 发送 force=True（虚构参数）→ 被静默剥离
# LLM 发送 amount=500（正确）→ 安全执行
```
**就这样。** 你的函数现在具备幽灵参数剥离、严格的类型验证和自愈错误。
## 🧠 未被提及的问题
|
### 炒作
|
### 现实
**LLM 每天都在编造工具调用。**
- Claude 编造不存在的参数
- GPT-4 在需要 `100` 时发送 `"100"`
- 代理在删除生产数据前链式调用 47 次
|
**企业解决方案存在：** Prompt Security（每年 5 万美元）、Pangea（代理数据）、Cisco（“即将推出”）。
**我们提供了开源替代方案。** 一个装饰器。无供应商锁定。你的数据永不离开你的基础设施。
## ✨ 你得到什么
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
|
## 📋 目录
点击展开完整导航
- [30 秒快速入门](#-30-second-quickstart)
- [问题](#-the-problem-no-one-talks-about)
- [你得到什么](#-what-you-get)
- [核心功能](#-core-features)
- [E2B 沙箱](#-e2b-sandbox-execution)
- [安全策略](#-security-policies)
- [成本控制](#-cost-control)
- [PII 掩码](#-pii--secret-masking)
- [网络空气隔离](#-network-airgap-v030)
- [框架接种](#-framework-vaccination-v030)
- [断路器](#-circuit-breaker-v040)
- [OpenTelemetry 可观测性](#-opentelemetry-observability-v040)
- [框架兼容性](#-framework-compatibility)
- [FastMCP 集成](#-fastmcp-integration)
- [对比](#-why-not-enterprise-vendors)
- [安装](#-installation)
- [OWASP 合规性](#️-owasp-compliance)
- [性能](#-performance)
- [文档](#-documentation)
- [贡献](#-contributing)
- [支持](#-support)
## 🔥 核心功能
### 🔒 E2B 沙箱执行
```
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"
```
| 特性 | 值 |
|---------|-------|
| 启动时间 | ~125ms 冷启动，<200ms 暖启动 |
| 隔离 | Firecracker MicroVM |
| 回退 | `sandbox_required=True` 阻止本地执行 |
### 📜 安全策略
```
from agent_airlock import (
PERMISSIVE_POLICY, # Dev - no restrictions
STRICT_POLICY, # Prod - rate limited, agent ID required
READ_ONLY_POLICY, # Analytics - query only
BUSINESS_HOURS_POLICY, # Dangerous ops 9-5 only
)
# 或者自行构建：
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"},
)
```
### 💰 成本控制
失控的代理可能在通知你之前消耗 500 美元的 API 费用。
```
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:
return massive_log_query(query) # 10MB → 5KB
```
**投资回报率：** 10MB 日志 ≈ 2.5M token ≈ 25 美元/响应。截断 ≈ 1.25K token ≈ 0.01 美元。**节省 99.96%。**
### 🔐 PII 与密钥掩码
```
config = AirlockConfig(
mask_pii=True, # SSN, credit cards, phones, emails
mask_secrets=True, # API keys, passwords, JWTs
)
@Airlock(config=config)
def get_user(user_id: str) -> dict:
return db.users.find_one({"id": user_id})
# LLM 接收：{"name": "John", "ssn": "[REDACTED]", "api_key": "sk-...XXXX"}
```
**检测 12 种 PII 类型** · **4 种掩码策略** · **零数据泄漏**
### 🌐 网络空气隔离（V0.3.0）
在工具执行期间阻止数据外泄：
```
from agent_airlock import network_airgap, NO_NETWORK_POLICY
# 阻止所有网络访问
with network_airgap(NO_NETWORK_POLICY):
result = untrusted_tool() # Any socket call → NetworkBlockedError
# 或仅允许特定主机
from agent_airlock import NetworkPolicy
INTERNAL_ONLY = NetworkPolicy(
allow_egress=True,
allowed_hosts=["api.internal.com", "*.company.local"],
allowed_ports=[443],
)
```
### 💉 框架接种（V0.3.0）
在不更改一行代码的情况下保护现有代码：
```
from agent_airlock import vaccinate, STRICT_POLICY
# 之前：您的现有 LangChain 工具未受保护
vaccinate("langchain", policy=STRICT_POLICY)
# 之后：所有 @tool 装饰器均包含 Airlock 安全机制
# 无需代码更改！
```
**支持：** LangChain、OpenAI Agents SDK、PydanticAI、CrewAI
### ⚡ 断路器（V0.4.0）
通过容错防止级联故障：
```
from agent_airlock import CircuitBreaker, AGGRESSIVE_BREAKER
breaker = CircuitBreaker("external_api", config=AGGRESSIVE_BREAKER)
@breaker
def call_external_api(query: str) -> dict:
return external_service.query(query)
# 5 次失败后 → 断路器 打开 → 30 秒内快速失败
# 随后进入 HALF_OPEN → 允许 1 次测试请求 → 恢复或重新打开
```
### 📈 OpenTelemetry 可观测性（V0.4.0）
企业级监控：
```
from agent_airlock import configure_observability, observe
configure_observability(
service_name="my-agent",
otlp_endpoint="http://otel-collector:4317",
)
@observe(name="critical_operation")
def process_data(data: dict) -> dict:
# Automatic span creation, metrics, and audit logging
return transform(data)
```
## 🔌 框架兼容性
```
@framework_decorator # ← Framework sees 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}"
```
|
### OpenAI Agents SDK
```
from agents import function_tool
from agent_airlock import Airlock
@function_tool
@Airlock()
def get_weather(city: str) -> str:
"""Get weather for a city."""
return f"Weather in {city}: 22°C"
```
|
|
### PydanticAI
```
from pydantic_ai import Agent
from agent_airlock import Airlock
@Airlock()
def get_stock(symbol: str) -> str:
return f"Stock {symbol}: $150"
agent = Agent("openai:gpt-4o", tools=[get_stock])
```
|
### CrewAI
```
from crewai.tools import tool
from agent_airlock import Airlock
@tool
@Airlock()
def search_docs(query: str) -> str:
"""Search internal docs."""
return f"Found 5 docs for: {query}"
```
|
更多框架：LlamaIndex、AutoGen、smolagents、Anthropic
### LlamaIndex
```
from llama_index.core.tools import FunctionTool
from agent_airlock import Airlock
@Airlock()
def calculate(expression: str) -> int:
return eval(expression, {"__builtins__": {}})
calc_tool = FunctionTool.from_defaults(fn=calculate)
```
### AutoGen
```
from autogen import ConversableAgent
from agent_airlock import Airlock
@Airlock()
def analyze_data(dataset: str) -> str:
return f"Analysis of {dataset}: mean=42.5"
assistant = ConversableAgent(name="analyst", llm_config={"model": "gpt-4o"})
assistant.register_for_llm()(analyze_data)
```
### smolagents
```
from smolagents import tool
from agent_airlock import Airlock
@tool
@Airlock(sandbox=True)
def run_code(code: str) -> str:
"""Execute in E2B sandbox."""
exec(code)
return "Executed"
```
### Anthropic（直接 API）
```
from agent_airlock import Airlock
@Airlock()
def get_weather(city: str) -> str:
return f"Weather in {city}: 22°C"
# 在工具处理器中使用
def handle_tool_call(name, inputs):
if name == "get_weather":
return get_weather(**inputs) # Airlock validates
```
### 完整示例
| 框架 | 示例 | 关键特性 |
|-----------|---------|--------------|
| LangChain | [`langchain_integration.py`](./examples/langchain_integration.py) | @tool, AgentExecutor |
| LangGraph | [`langgraph_integration.py`](./examples/langgraph_integration.py) | StateGraph, ToolNode |
| OpenAI Agents | [`openai_agents_sdk_integration.py`](./examples/openai_agents_sdk_integration.py) | Handoffs, manager pattern |
| PydanticAI | [`pydanticai_integration.py`](./examples/pydanticai_integration.py) | 依赖项、结构化输出 |
| LlamaIndex | [`llamaindex_integration.py`](./examples/llamaindex_integration.py) | ReActAgent |
| CrewAI | [`crewai_integration.py`](./examples/crewai_integration.py) | Crews, roles |
| AutoGen | [`autogen_integration.py`](./examples/autogen_integration.py) | ConversableAgent |
| smolagents | [`smolagents_integration.py`](./examples/smolagents_integration.py) | CodeAgent, E2B |
| Anthropic | [`anthropic_integration.py`](https://img.shields.io/pypi/dm/agent-airlock?style=for-the-badge&logo=python&logoColor=white&color=success) | 直接 API |
| Claude Agent SDK | [`anthropic_integration.py`](./examples/anthropic_integration.py) (示例 7) | 代理 SDK 工具作为进程内 MCP 服务器 |
## ⚡ FastMCP 集成
```
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)
```
## 🏆 为何不选择企业供应商？
| | Prompt Security | Pangea | **Agent-Airlock** |
|---|:---:|:---:|:---:|
| **定价** | 每年 5 万美元+ | 企业版 | **永久免费** |
| **集成** | 代理网关 | 代理网关 | **一个装饰器** |
| **自愈** | ❌ | ❌ | **✅** |
| **E2B 沙箱** | ❌ | ❌ | **✅ 原生** |
| **你的数据** | 他们的服务器 | 他们的服务器 | **永不离开你** |
| **源代码** | 闭源 | 闭源 | **MIT 许可** |
## 📦 安装
```
# 核心（验证 + 策略 + 清洗）
pip install agent-airlock
# 支持 E2B 沙箱
pip install agent-airlock[sandbox]
# 与 FastMCP 集成
pip install agent-airlock[mcp]
# 全部
pip install agent-airlock[all]
```
```
# E2B 沙箱执行密钥
export E2B_API_KEY="your-key-here"
```
## 🛡️ OWASP 合规性
Agent-Airlock 缓解了 [OWASP 大型语言模型十大风险 (2025)](https://owasp.org/www-project-top-10-for-large-language-model-applications/)：
| OWASP 风险 | 缓解措施 |
|------------|------------|
| **LLM01：提示注入** | 严格的类型验证阻止注入的有效载荷 |
| **LLM02：敏感数据泄露** | 网络空气隔离防止数据外泄 |
| **LLM05：不正确的输出处理** | PII/密钥掩码清理输出 |
| **LLM06：过度代理** | 速率限制 + RBAC + 能力门控防止失控代理 |
| **LLM07：系统提示泄露** | 蜜罐返回假数据而非错误 |
| **LLM09：错误信息** | 幽灵参数拒绝阻止幻觉参数 |
## 🏢 使用方
Agent-Airlock 在生产环境中保护 AI 代理系统：
| 项目 | 用例 |
|---------|----------|
| [**Attri.ai**](https://attri.ai) | 多代理编排平台 — 治理与安全层 |
| [**FerrumDeck**](https://github.com/sattyamjjain/FerrumDeck) | AgentOps 控制平面 — 默认拒绝工具执行 |
| [**Mnemo**](https://github.com/sattyamjjain/Mnemo) | MCP 原生内存数据库 — 安全的工具调用验证 |
## 📊 性能
| 指标 | 值 |
|--------|-------|
| **测试** | 1,157 个通过 |
| **覆盖率** | 79%+（CI 强制执行） |
| **代码行数** | ~25,900 |
| **验证开销** | <50ms |
| **沙箱冷启动** | ~125ms |
| **沙箱热池** | <200ms |
| **框架集成** | 9 |
| **核心依赖** | 0（仅 Pydantic） |
## 📖 文档
| 资源 | 描述 |
|----------|-------------|
| [**示例**](./examples/) | 9 个框架集成，带复制粘贴代码 |
| [**安全指南**](./docs/SECURITY.md) | 生产部署检查清单 |
| [**API 参考**](./docs/API.md) | 每个函数、每个参数 |
## 👤 关于
由 [**Sattyam Jain**](https://github.com/sattyamjjain) 构建 — AI 基础设施工程师。
这最初是一个内部工具，源于观看一个代理在生产数据库中胡言乱语。现在它属于你。
## 🤝 贡献
我们会在 48 小时内审查每个 PR。
```
git clone https://github.com/sattyamjjain/agent-airlock
cd agent-airlock
pip install -e ".[dev]"
pytest tests/ -v
```
- **发现 Bug？** [提出问题](https://github.com/sattyamjjain/agent-airlock/issues)
- **功能建议？** [开始讨论](https://github.com/sattyamjjain/agent-airlock/discussions)
- **想贡献？** [查看开放问题](https://github.com/sattyamjjain/agent-airlock/issues?q=is%3Aissue+is%3Aopen+label%3A%22good+first+issue%22)
## 💖 支持
如果 Agent-Airlock 拯救了你的生产数据库：
- ⭐ **给仓库加星** — 帮助他人发现它
- 🐛 **报告 Bug** — [提出问题](https://github.com/sattyamjjain/agent-airlock/issues)
- 📣 **传播消息** — 推特、博客、分享
## ⭐ 星标历史
[](https://star-history.com/#sattyamjjain/agent-airlock&Date)
**用 🛡️ 构建的 [Sattyam Jain](https://github.com/sattyamjjain)**
让 AI 代理安全，一次一个装饰器。
[](https://github.com/sattyamjjain)
[](https://twitter.com/sattyamjjain)
**资料来源：** 本 README 参考了 [awesome-readme](https://github.com/matiassingers/awesome-readme)、[Best-README-Template](https://github.com/othneildrew/Best-README-Template) 和 [GitHub 博客](https://github.blog/open-source/maintainers/marketing-for-maintainers-how-to-promote-your-project-to-both-users-and-contributors/) 的最佳实践。