airtable

name	airtable
description	Build integrations with the Airtable Web API — bases, tables, records, fields, views, webhooks, and OAuth. Use when tasks involve reading or writing Airtable data, syncing external sources with Airtable bases, building automations triggered by record changes, or migrating data to/from Airtable.
license	Apache-2.0
compatibility	No special requirements
metadata	{"author":"terminal-skills","version":"1.0.0","category":"productivity","tags":["airtable","api","productivity","databases","automation"]}

Airtable API Integration

Automate and integrate with Airtable bases using the REST API.

Authentication

Personal Access Token (simplest)

Generate at https://airtable.com/create/tokens. Scope to specific bases and permissions.

export AIRTABLE_TOKEN="pat..."

OAuth 2.0 (multi-user apps)

Register at https://airtable.com/create/oauth. Supports PKCE for public clients.

"""airtable_oauth.py — OAuth 2.0 with PKCE for Airtable."""
import hashlib, secrets, base64, requests

def start_oauth(client_id: str, redirect_uri: str) -> tuple[str, str]:
    """Generate authorization URL with PKCE challenge.

    Args:
        client_id: From Airtable OAuth integration settings.
        redirect_uri: Your callback URL.

    Returns:
        Tuple of (authorization_url, code_verifier) — store verifier for token exchange.
    """
    verifier = secrets.token_urlsafe(64)
    challenge = base64.urlsafe_b64encode(
        hashlib.sha256(verifier.encode()).digest()
    ).rstrip(b"=").decode()

    url = (
        f"https://airtable.com/oauth2/v1/authorize?"
        f"client_id={client_id}&redirect_uri={redirect_uri}"
        f"&response_type=code&scope=data.records:read data.records:write schema.bases:read"
        f"&code_challenge={challenge}&code_challenge_method=S256"
    )
    return url, verifier

def exchange_token(code: str, verifier: str, client_id: str, redirect_uri: str) -> dict:
    """Exchange authorization code for access token.

    Args:
        code: From Airtable's redirect.
        verifier: The PKCE code_verifier from start_oauth.
        client_id: OAuth client ID.
        redirect_uri: Must match the one used in authorization.
    """
    resp = requests.post("https://airtable.com/oauth2/v1/token", data={
        "grant_type": "authorization_code",
        "code": code,
        "redirect_uri": redirect_uri,
        "client_id": client_id,
        "code_verifier": verifier,
    })
    return resp.json()  # access_token, refresh_token, expires_in

Core API Patterns

Record Operations

"""airtable_records.py — CRUD operations on Airtable records."""
import requests, time

API = "https://api.airtable.com/v0"

def headers(token: str) -> dict:
    return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

def list_records(token: str, base_id: str, table_name: str,
                 view: str = None, formula: str = None,
                 fields: list = None, sort: list = None) -> list:
    """List all records from a table with optional filtering.

    Args:
        token: Personal access token or OAuth token.
        base_id: Base ID (starts with 'app').
        table_name: Table name or ID.
        view: Optional view name to filter/sort by.
        formula: Airtable formula for filtering (e.g., "AND({Status}='Active', {Score}>80)").
        fields: List of field names to return (reduces payload).
        sort: List of dicts with 'field' and 'direction' keys.

    Returns:
        List of all matching record objects.
    """
    params = {}
    if view:
        params["view"] = view
    if formula:
        params["filterByFormula"] = formula
    if fields:
        for i, f in enumerate(fields):
            params[f"fields[{i}]"] = f
    if sort:
         i, s  (sort):
            params[] = s[]
            params[] = s.get(, )

    records = []
    offset = 
     :
         offset:
            params[] = offset
        resp = requests.get(,
                            params=params, headers=headers(token))
        resp.raise_for_status()
        data = resp.json()
        records.extend(data[])
        offset = data.get()
          offset:
            
        time.sleep()  

     records

 () -> :
    
    
    created = []
     i  (, (records), ):
        batch = [{: r}  r  records[i:i + ]]
        resp = requests.post(
            ,
            json={: batch, : typecast},
            headers=headers(token),
        )
        resp.raise_for_status()
        created.extend(resp.json()[])
         i +  < (records):
            time.sleep()
     created

 () -> :
    
    updated = []
     i  (, (updates), ):
        batch = updates[i:i + ]
        resp = requests.patch(
            ,
            json={: batch},
            headers=headers(token),
        )
        resp.raise_for_status()
        updated.extend(resp.json()[])
         i +  < (updates):
            time.sleep()
     updated

 () -> :
    
    deleted = []
     i  (, (record_ids), ):
        batch = record_ids[i:i + ]
        params = .join(  rid  batch)
        resp = requests.delete(
            ,
            headers=headers(token),
        )
        resp.raise_for_status()
        deleted.extend(resp.json()[])
         i +  < (record_ids):
            time.sleep()
     deleted

Formula Filtering

Airtable formulas are powerful for server-side filtering:

# Common formula patterns
formulas = {
    # Exact match
    "status_active": "{Status} = 'Active'",

    # Multiple conditions
    "high_priority_open": "AND({Priority} = 'High', {Status} != 'Done')",

    # Date filtering — tasks due this week
    "due_this_week": "IS_BEFORE({Due Date}, DATEADD(TODAY(), 7, 'days'))",

    # Text search (case-insensitive)
    "name_contains": "FIND('search term', LOWER({Name}))",

    # Linked records — has at least one linked project
    "has_project": "{Project} != ''",

    # Number range
    "score_range": "AND({Score} >= 80, {Score} <= 100)",

    # Empty/non-empty checks
    "missing_email": "{Email} = ''",
    "has_attachment": "{Attachments} != ''",
}

Schema Operations

Read and modify base structure:

def get_base_schema(token: str, base_id: str) -> dict:
    """Get all tables, fields, and views in a base.

    Args:
        token: Auth token with schema.bases:read scope.
        base_id: Base ID.
    """
    resp = requests.get(f"https://api.airtable.com/v0/meta/bases/{base_id}/tables",
                        headers=headers(token))
    return resp.json()

def create_field(token: str, base_id: str, table_id: str,
                 name: str, field_type: str, options: dict = None) -> dict:
    """Add a new field (column) to a table.

    Args:
        token: Auth token with schema.bases:write scope.
        base_id: Base ID.
        table_id: Table ID (not name).
        name: Field name.
        field_type: One of: singleLineText, multilineText, number, percent,
                    currency, singleSelect, multipleSelects, date, checkbox, etc.
        options: Type-specific options (e.g., select choices, currency symbol).
    """
    body = {"name": name, "type": field_type}
    if options:
        body["options"] = options
    resp = requests.post(
        f"https://api.airtable.com/v0/meta/bases/{base_id}/tables/{table_id}/fields",
        json=body, headers=headers(token),
    )
    return resp.json()

Webhooks

Listen for changes in real time:

def create_webhook(token: str, base_id: str, notification_url: str,
                   table_id: str = None) -> dict:
    """Register a webhook for record changes.

    Args:
        token: Auth token with webhook:manage scope.
        base_id: Base ID to watch.
        notification_url: Your HTTPS endpoint for receiving payloads.
        table_id: Optional — watch a specific table instead of entire base.
    """
    spec = {"options": {"filters": {"sourceType": "client", "recordChangeScope": base_id}}}
    if table_id:
        spec["options"]["filters"]["watchDataInFieldIds"] = "all"
    resp = requests.post(
        f"https://api.airtable.com/v0/bases/{base_id}/webhooks",
        json={"notificationUrl": notification_url, "specification": spec},
        headers=headers(token),
    )
    return resp.json()

def list_webhook_payloads(token: str, base_id: str, webhook_id: str,
                          cursor: int = None) -> dict:
    """Fetch webhook payloads (poll-based — payloads expire after 7 days).

    Args:
        token: Auth token.
        base_id: Base ID.
        webhook_id: Webhook ID from create_webhook.
        cursor: Cursor from previous response for pagination.
    """
    params = {}
    if cursor:
        params["cursor"] = cursor
    resp = requests.get(
        f"https://api.airtable.com/v0/bases/{base_id}/webhooks//payloads",
        params=params, headers=headers(token),
    )
     resp.json()

Rate Limits

• 5 requests per second per base
• Record operations: max 10 records per create/update/delete call
• List: max 100 records per page (use offset for pagination)
• Implement backoff on 429 responses

Field Type Reference

Type	API name	Notes
Single line text	singleLineText	Plain string
Long text	multilineText	Supports markdown
Number	number	Integer or decimal
Single select	singleSelect	One from predefined list
Multiple select	multipleSelects	Array of selections
Date	date	ISO 8601 string
Checkbox	checkbox	Boolean
URL	url	Validated URL string
Email	email	Validated email string
Phone	phoneNumber	String
Currency	currency	Number with currency formatting
Percent	percent	Number displayed as percentage
Attachment	multipleAttachments	Array of file objects with URL
Linked record	multipleRecordLinks	Array of record IDs
Lookup	multipleLookupValues	Read-only, from linked records
Rollup	rollup	Read-only, aggregation of linked records
Formula	formula	Read-only, computed value

Guidelines

• Always paginate list requests -- never assume all records fit in one response
• Use filterByFormula to reduce payload instead of fetching all and filtering client-side
• Batch operations in groups of 10 (API limit) with 200ms delays
• Use typecast: true for imports where field values might need conversion
• Attachment fields require URLs -- upload to a hosting service first, then pass the URL
• Linked record fields accept record IDs, not display values
• Formula and rollup fields are read-only -- you cannot write to them via API
• Webhook payloads don't contain full record data -- they signal changes, then you fetch details
• Rate limits are per-base, not per-token -- multiple integrations on the same base share the limit