$npx -y skills add TerminalSkills/skills --skill airtableBuild 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
| 1 | # Airtable API Integration |
| 2 | |
| 3 | Automate and integrate with Airtable bases using the REST API. |
| 4 | |
| 5 | ## Authentication |
| 6 | |
| 7 | ### Personal Access Token (simplest) |
| 8 | |
| 9 | Generate at https://airtable.com/create/tokens. Scope to specific bases and permissions. |
| 10 | |
| 11 | ```bash |
| 12 | export AIRTABLE_TOKEN="pat..." |
| 13 | ``` |
| 14 | |
| 15 | ### OAuth 2.0 (multi-user apps) |
| 16 | |
| 17 | Register at https://airtable.com/create/oauth. Supports PKCE for public clients. |
| 18 | |
| 19 | ```python |
| 20 | """airtable_oauth.py — OAuth 2.0 with PKCE for Airtable.""" |
| 21 | import hashlib, secrets, base64, requests |
| 22 | |
| 23 | def start_oauth(client_id: str, redirect_uri: str) -> tuple[str, str]: |
| 24 | """Generate authorization URL with PKCE challenge. |
| 25 | |
| 26 | Args: |
| 27 | client_id: From Airtable OAuth integration settings. |
| 28 | redirect_uri: Your callback URL. |
| 29 | |
| 30 | Returns: |
| 31 | Tuple of (authorization_url, code_verifier) — store verifier for token exchange. |
| 32 | """ |
| 33 | verifier = secrets.token_urlsafe(64) |
| 34 | challenge = base64.urlsafe_b64encode( |
| 35 | hashlib.sha256(verifier.encode()).digest() |
| 36 | ).rstrip(b"=").decode() |
| 37 | |
| 38 | url = ( |
| 39 | f"https://airtable.com/oauth2/v1/authorize?" |
| 40 | f"client_id={client_id}&redirect_uri={redirect_uri}" |
| 41 | f"&response_type=code&scope=data.records:read data.records:write schema.bases:read" |
| 42 | f"&code_challenge={challenge}&code_challenge_method=S256" |
| 43 | ) |
| 44 | return url, verifier |
| 45 | |
| 46 | def exchange_token(code: str, verifier: str, client_id: str, redirect_uri: str) -> dict: |
| 47 | """Exchange authorization code for access token. |
| 48 | |
| 49 | Args: |
| 50 | code: From Airtable's redirect. |
| 51 | verifier: The PKCE code_verifier from start_oauth. |
| 52 | client_id: OAuth client ID. |
| 53 | redirect_uri: Must match the one used in authorization. |
| 54 | """ |
| 55 | resp = requests.post("https://airtable.com/oauth2/v1/token", data={ |
| 56 | "grant_type": "authorization_code", |
| 57 | "code": code, |
| 58 | "redirect_uri": redirect_uri, |
| 59 | "client_id": client_id, |
| 60 | "code_verifier": verifier, |
| 61 | }) |
| 62 | return resp.json() # access_token, refresh_token, expires_in |
| 63 | ``` |
| 64 | |
| 65 | ## Core API Patterns |
| 66 | |
| 67 | ### Record Operations |
| 68 | |
| 69 | ```python |
| 70 | """airtable_records.py — CRUD operations on Airtable records.""" |
| 71 | import requests, time |
| 72 | |
| 73 | API = "https://api.airtable.com/v0" |
| 74 | |
| 75 | def headers(token: str) -> dict: |
| 76 | return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} |
| 77 | |
| 78 | def list_records(token: str, base_id: str, table_name: str, |
| 79 | view: str = None, formula: str = None, |
| 80 | fields: list = None, sort: list = None) -> list: |
| 81 | """List all records from a table with optional filtering. |
| 82 | |
| 83 | Args: |
| 84 | token: Personal access token or OAuth token. |
| 85 | base_id: Base ID (starts with 'app'). |
| 86 | table_name: Table name or ID. |
| 87 | view: Optional view name to filter/sort by. |
| 88 | formula: Airtable formula for filtering (e.g., "AND({Status}='Active', {Score}>80)"). |
| 89 | fields: List of field names to return (reduces payload). |
| 90 | sort: List of dicts with 'field' and 'direction' keys. |
| 91 | |
| 92 | Returns: |
| 93 | List of all matching record objects. |
| 94 | """ |
| 95 | params = {} |
| 96 | if view: |
| 97 | params["view"] = view |
| 98 | if formula: |
| 99 | params["filterByFormula"] = formula |
| 100 | if fields: |
| 101 | for i, f in enumerate(fields): |
| 102 | params[f"fields[{i}]"] = f |
| 103 | if sort: |
| 104 | for i, s in enumerate(sort): |
| 105 | params[f"sort[{i}][field]"] = s["field"] |
| 106 | params[f"sort[{i}][direction]"] = s.get("direction", "asc") |
| 107 | |
| 108 | records = [] |
| 109 | offset = None |
| 110 | while True: |
| 111 | if offset: |
| 112 | params["offset"] = offset |
| 113 | resp = requests.get(f"{API}/{base_id}/{table_name}", |
| 114 | params=params, headers=headers(token)) |
| 115 | resp.raise_for_status() |
| 116 | data = resp.json() |
| 117 | records.extend(data["records"]) |
| 118 | offset = data.get("offset") |
| 119 | if not offset: |
| 120 | break |
| 121 | time.sleep(0.2) # Stay under 5 req/s rate limit |
| 122 | |
| 123 | return records |
| 124 | |
| 125 | def create_records(token: str, base_id: str, table_name: str, |
| 126 | records: list[dict], typecast: bool = False) -> list: |
| 127 | """Create up to 10 records at a time. |
| 128 | |
| 129 | Args: |
| 130 | token: Auth token. |
| 131 | base_id: Base ID. |
| 132 | table_name: Target table. |
| 133 | records: List of dicts with field values (max 10 per call). |
| 134 | typecast: If True, Airtable auto-converts string values to proper types. |
| 135 | |
| 136 | Returns: |
| 137 | List of created record objects with IDs. |
| 138 | """ |
| 139 | # Airtable limi |