Airtable API skill for Codex

Done well means: the code addresses bases, tables and fields by ID, writes in batches of 10 with upserts so reruns are safe, never exceeds 5 requests per second per base, and handles Airtable’s habit of omitting empty fields from responses.

Setup facts

• Auth is a personal access token (PAT) or OAuth; legacy API keys no longer work. A PAT has scopes (data.records:read, data.records:write, schema.bases:read, webhook:manage) and an explicit list of bases. A 403/404 on a base you can see in the UI usually means the token was not granted that base.
• IDs: base app…, table tbl…, field fld…, record rec…, view viw…. Names change when someone renames a column; IDs do not. Use returnFieldsByFieldId=true for syncs.
• Token from env (AIRTABLE_TOKEN); read AGENTS.md for secrets. All calls need network; in an offline Codex sandbox write the code and ask before running.
• Plan limits apply: records per base and monthly API calls on the Free plan are low; check before designing a high-volume sync.

Read the schema first

curl -s -H "Authorization: Bearer $AIRTABLE_TOKEN" \
  "https://api.airtable.com/v0/meta/bases/$BASE_ID/tables" \
  | jq '.tables[] | {id, name, fields: [.fields[] | {id, name, type}]}'

Map your source columns to field IDs and types before writing. Computed fields (formula, rollup, lookup, count, created time, autonumber) are read-only; writing them returns 422.

List records

import os
from pyairtable import Api

api = Api(os.environ["AIRTABLE_TOKEN"])
orders = api.table("appXXXXXXXXXXXXXX", "tblYYYYYYYYYYYYYY")

rows = orders.all(
    formula="AND({Status} = 'Open', IS_AFTER({Due}, DATEADD(TODAY(), -7, 'days')))",
    fields=["Order ID", "Status", "Due", "Customer"],
    page_size=100,
)

Plain HTTP pages with offset:

async function* listAll(base: string, table: string, params: Record<string, string>) {
  let offset: string | undefined;
  do {
    const qs = new URLSearchParams({ pageSize: "100", returnFieldsByFieldId: "true", ...params });
    if (offset) qs.set("offset", offset);
    const res = await at(`https://api.airtable.com/v0/${base}/${table}?${qs}`);
    yield* res.records;
    offset = res.offset;
  } while (offset);
}

• Long formulas or many fields[] overflow the URL limit; use POST /v0/{base}/{table}/listRecords with the same parameters in the JSON body.
• view=viw… applies that view’s filters and sort; handy, but a teammate editing the view changes your sync. Prefer explicit formulas in code.
• Formula strings: field names in braces, string literals in quotes; escape user values (value.replace("\\", "\\\\").replace("'", "\\'")) before interpolating.
• Empty fields are omitted from fields entirely (unchecked checkbox, empty text, empty link). Treat a missing key as empty, not as “unchanged”.

Write: batches of 10 and upserts

Operation	Endpoint	Limit
Create	POST /v0/{base}/{table} with records: [{fields}]	10 records per request
Update (merge)	PATCH with records: [{id, fields}]	10
Replace (clears unspecified fields)	PUT	10; avoid unless intended
Upsert	PATCH with performUpsert: {fieldsToMergeOn: ["External ID"]}	10; up to 3 merge fields
Delete	DELETE ?records[]=rec…&records[]=rec…	10; confirm with user first

records = [{"fields": {"External ID": r.id, "Name": r.name, "Amount": r.amount}} for r in source]
result = orders.batch_upsert(records, key_fields=["External ID"], typecast=True)
print(len(result["createdRecords"]), "created;", len(result["updatedRecords"]), "updated")

pyairtable chunks into requests of 10 and paces them. typecast: true lets Airtable convert strings into select options and linked records; it also silently creates new select options from typos, so validate choices against the schema first when data quality matters.

Field value shapes:

Field type	Write
Single select	"Open"
Multiple select	["a", "b"]
Linked record	["recAAA", "recBBB"] (IDs; names only with typecast)
Attachment	[{"url": "https://...", "filename": "x.pdf"}] (Airtable fetches the URL)
Checkbox	true / false
Date / DateTime	"2026-10-28" / ISO 8601 with offset
User (collaborator)	{"id": "usr…"} or {"email": "..."}

Rate limits

• 5 requests per second per base; 50 requests per second per token across all bases.
• A 429 means stop for 30 seconds before retrying; hammering extends the penalty.
• For bulk jobs, run one worker per base at about 4 requests per second, with backoff.

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
let last = 0;
async function at(url: string, init: RequestInit = {}, attempt = 0): Promise<any> {
  const wait = Math.max(0, last + 250 - Date.now());          // ~4 req/s per process
  await sleep(wait); last = Date.now();
  const res = await fetch(url, { ...init, headers: { Authorization: `Bearer ${process.env.AIRTABLE_TOKEN}`,
    "Content-Type": "application/json", ...(init.headers ?? {}) } });
  if (res.status === 429 && attempt < 3) { await sleep(30_000); return at(url, init, attempt + 1); }
  if (!res.ok) throw new Error(`${res.status} ${await res.text()}`);
  return res.json();
}

Webhooks

- [ ] POST /v0/bases/{base}/webhooks with notificationUrl and a specification (dataTypes, recordChangeScope)
- [ ] Store the returned id, macSecretBase64 and expirationTime securely
- [ ] On each ping: verify X-Airtable-Content-MAC, respond 200 quickly
- [ ] Fetch GET /v0/bases/{base}/webhooks/{id}/payloads?cursor=N until mightHaveMore is false
- [ ] Persist the cursor after processing
- [ ] Refresh with POST /v0/bases/{base}/webhooks/{id}/refresh before the 7-day expiry

The notification body carries no change data, only a signal; the payloads endpoint holds the changes in order. Verify the MAC as hmac-sha256= + hex HMAC-SHA256 of the raw body using the base64-decoded macSecretBase64. Pings can be missed: a scheduled payload fetch from the stored cursor is a cheap safety net.

In-Airtable alternatives

For logic that stays inside one base, an Automation with a “Run a script” action (JavaScript, input.config(), fetch available, short time limit) may be simpler than an external worker. Use the Web API when you need version control, retries, or more than one base.

Gotchas

• Attachment URLs returned by the API are temporary; download files promptly instead of storing the URLs.
• Linked record fields return IDs only; resolve names with a second lookup (cache by table).
• Record order without sort or a view is not guaranteed.
• Formula {Field} = '' matches both empty and missing; BLANK() comparisons behave similarly.
• Deleting records is recoverable from trash for a limited time in the UI, not via API. Get explicit confirmation of the exact IDs before any delete.

Verification

• Dry run prints creates vs updates computed from the merge field before writing.
• After a sync, list with filterByFormula on the external ID for a sample and compare fields.
• Count check: len(table.all(fields=["External ID"])) equals the source count.
• Confirm no 429s in the run log; if there were, lower the request rate.