Airlock

Airlock is an open-source egress security layer for agent-to-agent handoffs. It checks agent outputs and artifacts before they are shared, blocks secret leaks, and produces a receipt that downstream agents can verify and consume safely.

The core idea is simple:

agent output/artifact -> Airlock policy gate -> verified receipt -> recipient agent

Airlock is the publish gate between one agent's handoff payload and another agent, user, or external service.

Why

Agent-to-agent systems create a new security problem: handoffs can accidentally carry credentials, private context, hidden instructions, debug logs, or tool-output residue into another agent's context.

Airlock addresses this with:

• secret scanning for files and planned egress payloads
• policy checks for protected secrets and sensitive file paths
• artifact receipts with hashes, verification status, and TTL
• a small CLI and Python API that can be embedded into A2A servers

Quick Start

cd /Users/chenhong/project/airlock
python3 -m airlock.cli scan examples/leaky-debug.log
python3 -m airlock.cli publish examples/safe-report.md --from-agent researcher --to-agent coder

Example

from airlock import Airlock

guard = Airlock.default()

receipt = guard.publish_files(
    files=["outputs/report.md", "outputs/sources.json"],
    from_agent="researcher",
    to_agent="coder",
    task_id="task_123",
)

print(receipt.to_a2a_artifact())

Agent Loop Integration

Airlock can also be installed once at the agent-loop boundary. Each wrapped turn checks input before the agent runs, output after the agent returns, and the final reply/files before handoff.

For most managed agents and SDK agents, the integration is two lines:

from airlock import Airlock

airlock = Airlock.config(
    from_agent="openai-agent",
    to_agent="downstream-agent",
)
airlock(agent)

airlock(agent) detects OpenAI Agents SDK-style agents and installs native input/output guardrails. For plain managed agent objects, it wraps common entrypoints such as run, invoke, call, deliver, and handoff.

For Claude Agent SDK options, use the same configured object:

airlock = Airlock.config(
    from_agent="claude-agent-sdk",
    to_agent="downstream-agent",
)
options = airlock.options(allowed_tools=["Read", "Write", "Bash"])

Use the lower-level loop wrapper when you own the turn function directly:

from airlock import Airlock, BehaviorCheck


def llm_judge(check: BehaviorCheck) -> dict:
    # Call your model with check.to_prompt(), then return JSON-compatible data:
    # {"allowed": true, "findings": []}
    return {"allowed": True, "findings": []}


loop_guard = Airlock.default().agent_loop(
    from_agent="researcher",
    to_agent="coder",
    task_id="task_123",
    llm_judge=llm_judge,
)

@loop_guard.wrap
def agent_step(prompt: str) -> dict:
    report_path = "outputs/report.md"
    return {
        "message": "Report is ready.",
        "artifacts": [{"name": "report", "path": report_path}],
    }

result = agent_step("Summarize the research.")

The wrapper detects common artifact shapes such as artifacts, files, and file_paths, scans referenced files before they leave the loop, and enriches clean results with airlock_receipt metadata plus an A2A-compatible airlock-receipt artifact.

Use wrap_delivery for the final response or handoff package:

@loop_guard.wrap_delivery
def final_delivery() -> dict:
    return {
        "reply": "The final report is attached.",
        "files": ["outputs/report.md"],
    }

If input, output, or final delivery looks suspicious, the loop is canceled and Airlock returns:

{
  "status": "blocked",
  "airlock_suspicious": true,
  "stage": "input",
  "reason": "suspicious agent-loop behavior detected"
}

For exception-based integrations, pass on_violation="raise" and catch AirlockBlocked.

For managed agent objects, install Airlock from the outside:

agent = ClaudeManagedAgent(...)

airlock = Airlock.default().managed_agent(
    from_agent="claude-managed-agent",
    to_agent="downstream-agent",
    llm_judge=llm_judge,
)

airlock.install(agent)

result = agent.run("prepare handoff")
delivery = agent.deliver(result)

install() wraps common managed-agent entrypoints when they exist: run, arun, invoke, ainvoke, step, run_turn, call, deliver, finalize, final_delivery, and handoff.

For SDK-native integrations, use the dedicated adapters:

# OpenAI Agents SDK: install native input/output guardrails.
airlock = Airlock.default().openai_agents(
    from_agent="openai-agent",
    to_agent="downstream-agent",
    llm_judge=llm_judge,
)
airlock.install(agent)
result = await Runner.run(agent, "prepare handoff")

# Claude Agent SDK: inject hooks into ClaudeAgentOptions.
airlock = Airlock.default().claude_agent_sdk(
    from_agent="claude-agent-sdk",
    to_agent="downstream-agent",
    llm_judge=llm_judge,
)
options = airlock.options(allowed_tools=["Read", "Write", "Bash"])

Project Scope

Airlock focuses on four jobs:

1. Detect whether a file or payload contains secrets.
2. Decide whether a proposed handoff is allowed by policy.
3. Create an immutable receipt for allowed artifacts.
4. Provide metadata that can be embedded in A2A Task.artifacts.

It intentionally does not implement:

• agent discovery
• task orchestration
• model hosting
• long-term memory
• general-purpose cloud drive UX

Status

This repository is an initial MVP scaffold. The current implementation includes local scanning, policy decisions, receipt creation, and a CLI. Storage adapters, HTTP API, and A2A server plugins are documented next steps.