Py Python — `airlock-gateway`

An async Python client SDK for the Airlock Integrations Gateway API.

Installation

pip install airlock-gateway

pip install airlock-gateway

Quick Start

With Bearer Token

import asyncio
from airlock_gateway import (
    AirlockGatewayClient,
    ArtifactSubmitRequest,
    EncryptedPayload,
)

async def main():
    async with AirlockGatewayClient(
        "https://igw.airlocks.io", token="your-token"
    ) as client:
        echo = await client.echo()

asyncio.run(main())

import asyncio
from airlock_gateway import (
    AirlockGatewayClient,
    ArtifactSubmitRequest,
    EncryptedPayload,
)

async def main():
    async with AirlockGatewayClient(
        "https://igw.airlocks.io", token="your-token"
    ) as client:
        echo = await client.echo()

asyncio.run(main())

With Enforcer App Credentials

async with AirlockGatewayClient(
    "https://igw.airlocks.io",
    client_id="your-client-id",
    client_secret="your-client-secret",
) as client:
    echo = await client.echo()

async with AirlockGatewayClient(
    "https://igw.airlocks.io",
    client_id="your-client-id",
    client_secret="your-client-secret",
) as client:
    echo = await client.echo()

Authentication

Personal Access Token (PAT)

PAT is the recommended authentication for user-scoped operations. Sent via the X-PAT header.

# After obtaining a PAT from the mobile app (Settings → Access Tokens)
client.set_pat("airlock_pat_...")

# Clear PAT when no longer needed
client.set_pat(None)

# After obtaining a PAT from the mobile app (Settings → Access Tokens)
client.set_pat("airlock_pat_...")

# Clear PAT when no longer needed
client.set_pat(None)

Dual Auth (set_bearer_token)

# After user login (Device Auth Grant or Auth Code + PKCE)
client.set_bearer_token(access_token)

# After user login (Device Auth Grant or Auth Code + PKCE)
client.set_bearer_token(access_token)

Authentication by Enforcer App Kind

Kind	OAuth2 Flow	SDK Methods
Agent / Desktop / VsCode	Device Auth Grant (RFC 8628)	`login(on_user_code)`
Web	Auth Code + PKCE (RFC 7636)	`login_with_auth_code` or `get_authorization_url` + `exchange_code`
Mobile	Auth Code + PKCE (RFC 7636)	`get_authorization_url` + `exchange_code`

Gateway Client API

Method	Description
`echo()`	Gateway discovery / health
`set_pat(pat)`	Set Personal Access Token (X-PAT header)
`set_bearer_token(token)`	Set Bearer token for user-scoped operations
`check_consent()`	Check user consent for this enforcer app
`submit_artifact(request)`	Submit artifact for approval
`get_exchange_status(request_id)`	Get exchange status
`wait_for_decision(request_id, timeout)`	Long-poll for decision
`submit_ack(msg_id, request_id)`	Acknowledge decision delivery (`POST /v1/acks`, fire-and-forget)
`withdraw_exchange(request_id)`	Withdraw pending exchange
`initiate_pairing(request)`	Start pairing session
`claim_pairing(request)`	Claim a pre-generated pairing code
`get_pairing_status(nonce)`	Poll pairing status
`revoke_pairing(routing_token)`	Revoke a pairing
`send_heartbeat(request)`	Presence heartbeat
`get_effective_dnd_policies(enforcer_id, workspace_id, session_id)`	Fetch effective DND policies

Auth Client (`AirlockAuthClient`)

Method	Purpose
`discover`	OIDC discovery
`login`	Device code login
`login_with_auth_code`	Auth code + PKCE (local callback)
`get_authorization_url`, `exchange_code`	Auth code + PKCE (manual redirect)
`refresh_token`, `get_access_token`	Token refresh / access
`logout`	Sign out (revoke)

Pairing

Standard Pairing (Enforcer-Initiated)

from airlock_gateway import PairingInitiateRequest

# 1. Initiate a pairing session
resp = await client.initiate_pairing(PairingInitiateRequest(
    enforcer_id="my-enforcer",
    workspace_name="my-project",
    x25519_public_key=my_public_key,
))

# 2. Display pairing code to user
print(f"Pairing code: {resp.pairing_code}")

# 3. Poll for approval from the mobile app
status = await client.get_pairing_status(resp.nonce)
# status.state == "Completed" → save status.routing_token

from airlock_gateway import PairingInitiateRequest

# 1. Initiate a pairing session
resp = await client.initiate_pairing(PairingInitiateRequest(
    enforcer_id="my-enforcer",
    workspace_name="my-project",
    x25519_public_key=my_public_key,
))

# 2. Display pairing code to user
print(f"Pairing code: {resp.pairing_code}")

# 3. Poll for approval from the mobile app
status = await client.get_pairing_status(resp.nonce)
# status.state == "Completed" → save status.routing_token

Pre-Generated Code Pairing (Approver-Initiated)

from airlock_gateway import PairingClaimRequest

claim = await client.claim_pairing(PairingClaimRequest(
    code="ABCD-1234",
    enforcer_id="my-enforcer",
    workspace_name="my-project",
    x25519_public_key=my_public_key,
))
# claim.routing_token is ready to use

from airlock_gateway import PairingClaimRequest

claim = await client.claim_pairing(PairingClaimRequest(
    code="ABCD-1234",
    enforcer_id="my-enforcer",
    workspace_name="my-project",
    x25519_public_key=my_public_key,
))
# claim.routing_token is ready to use

Consent Check

from airlock_gateway import AirlockGatewayError

try:
    status = await client.check_consent()
    # status == "approved" — proceed normally
except AirlockGatewayError as e:
    if e.error_code == "app_consent_required":
        print("User hasn't granted consent")
    elif e.error_code == "app_consent_pending":
        print("Consent request sent, waiting for approval")

from airlock_gateway import AirlockGatewayError

try:
    status = await client.check_consent()
    # status == "approved" — proceed normally
except AirlockGatewayError as e:
    if e.error_code == "app_consent_required":
        print("User hasn't granted consent")
    elif e.error_code == "app_consent_pending":
        print("Consent request sent, waiting for approval")

Error Handling

All errors raise AirlockGatewayError with helper properties:

from airlock_gateway import AirlockGatewayError

try:
    await client.submit_artifact(request)
except AirlockGatewayError as e:
    if e.is_quota_exceeded:
        print("Quota exceeded")
    elif e.is_pairing_revoked:
        print("Pairing revoked")
    elif e.is_conflict:
        print("Idempotency conflict")
    else:
        print(f"Error {e.status_code}: {e}")

from airlock_gateway import AirlockGatewayError

try:
    await client.submit_artifact(request)
except AirlockGatewayError as e:
    if e.is_quota_exceeded:
        print("Quota exceeded")
    elif e.is_pairing_revoked:
        print("Pairing revoked")
    elif e.is_conflict:
        print("Idempotency conflict")
    else:
        print(f"Error {e.status_code}: {e}")

Encryption

The SDK includes crypto_helpers for X25519 ECDH key exchange and AES-256-GCM using cryptography:

generate_x25519_keypair() — raw 32-byte X25519 keypair (base64url)
derive_shared_key(my_private, peer_public) — ECDH + HKDF-SHA256 (info: HARP-E2E-AES256GCM)
aes_gcm_encrypt / aes_gcm_decrypt — AES-256-GCM with detached nonce and tag

Test Enforcer CLI

A fully interactive TUI that demonstrates the complete enforcer lifecycle.

# From the repo root
cd src/python

# Create and activate virtual environment (required once)
python -m venv .venv
.venv\Scripts\Activate.ps1   # Windows PowerShell
# source .venv/bin/activate  # macOS / Linux

# Install dependencies (required once)
pip install -r requirements.txt

# Run the test enforcer
python test_enforcer.py

# From the repo root
cd src/python

# Create and activate virtual environment (required once)
python -m venv .venv
.venv\Scripts\Activate.ps1   # Windows PowerShell
# source .venv/bin/activate  # macOS / Linux

# Install dependencies (required once)
pip install -r requirements.txt

# Run the test enforcer
python test_enforcer.py

Prerequisites: Python 3.9+. Configuration saved to ~/.airlock/test-enforcer-python.json.

← All SDKs Developer Guide

Py Python — airlock-gateway