MCP Server Airtable — table IDs, typecast, linked records, attachment URLs

Table IDs vs table names and rate limiting

Airtable's REST API accepts either the table name or the table ID in the URL path. Table names are human-readable but mutable — any collaborator with editor access can rename them. The tbl... table ID is a permanent 17-character identifier visible in the Airtable API documentation for your base. Find it by opening the Airtable API docs for your base: https://airtable.com/appXXXXX/api/docs — each table section shows its ID.

import Airtable from 'airtable';
import { z } from 'zod';

// Module-level setup — one Airtable client per base
const airtable = new Airtable({ apiKey: process.env.AIRTABLE_TOKEN! });
const base     = airtable.base(process.env.AIRTABLE_BASE_ID!);

// Rate limiter: 5 req/s per base (hard limit; 429 response if exceeded)
// Simple token bucket — sufficient for tool call volumes
let tokens = 5;
let lastRefill = Date.now();

function consumeToken(): Promise<void> {
  return new Promise(resolve => {
    const now = Date.now();
    const elapsed = (now - lastRefill) / 1000;
    tokens = Math.min(5, tokens + elapsed * 5);  // refill at 5/s
    lastRefill = now;

    if (tokens >= 1) {
      tokens -= 1;
      resolve();
    } else {
      const wait = Math.ceil((1 - tokens) / 5 * 1000);
      setTimeout(() => { tokens -= 1; resolve(); }, wait);
    }
  });
}

// Use table ID — not the human-readable name
const TASKS_TABLE_ID = 'tblABCDEF1234567';  // stable forever

server.tool(
  'airtable_list_records',
  {
    // Let the agent pass a formula filter using Airtable formula syntax
    filter_formula: z.string().optional()
      .describe('Airtable formula, e.g. AND({Status}="Open",{Priority}=3)'),
    max_records: z.number().int().min(1).max(100).default(20),
    sort_field:  z.string().optional(),
    sort_dir:    z.enum(['asc', 'desc']).default('asc'),
  },
  async ({ filter_formula, max_records, sort_field, sort_dir }) => {
    await consumeToken();

    const options: Airtable.SelectOptions = {
      maxRecords: max_records,
      view: 'Grid view',
    };

    if (filter_formula) options.filterByFormula = filter_formula;
    if (sort_field) options.sort = [{ field: sort_field, direction: sort_dir }];

    const records = await base(TASKS_TABLE_ID).select(options).all();

    return {
      content: [{
        type: 'text',
        text: JSON.stringify(
          records.map(r => ({ id: r.id, fields: r.fields }))
        ),
      }],
    };
  }
);

The all() method on an Airtable select query automatically iterates pages (100 records per page) until all results are fetched. For large tables use eachPage() with a page callback instead of all() to avoid loading thousands of records into memory at once.

filterByFormula field name syntax and typecast caution

Airtable's formula syntax requires field names that contain spaces to be wrapped in curly braces: {Field Name}. Without braces, Airtable interprets the space as a syntax boundary and returns a 422 error. String literals inside formulas use double quotes. The formula is URL-encoded when sent as a query parameter — the SDK handles this automatically.

// CORRECT: field names with spaces use {curly braces}
const openHighPriority = await base(TASKS_TABLE_ID).select({
  filterByFormula: 'AND({Task Status}="Open",{Priority Score}>=3)',
}).all();

// WRONG — spaces in field name without braces → 422 from Airtable
// filterByFormula: 'AND(Task Status="Open")' ← parse error

// Typecast: true creates new select options from unmatched strings
// SAFE for free-text fields:
await base(TASKS_TABLE_ID).create({
  'Description': 'Updated task notes',      // plain text — typecast doesn't matter
  'Due Date':    '2026-09-01',              // date string — typecast parses it
}, { typecast: true });

// DANGEROUS for Single Select fields — creates garbage options if value isn't in list
await base(TASKS_TABLE_ID).create({
  'Status': 'Compelted',  // typo → silently adds "Compelted" as a new option
}, { typecast: true });

// SAFE approach for select fields: validate against known options first
const VALID_STATUSES = ['Open', 'In Progress', 'Done', 'Blocked'];

server.tool(
  'airtable_create_task',
  {
    title:    z.string().min(1).max(512),
    status:   z.enum(['Open', 'In Progress', 'Done', 'Blocked']),
    priority: z.number().int().min(1).max(5).optional(),
    due_date: z.string().regex(/^\d{4}-\d{2}-\d{2}$/).optional(),
    notes:    z.string().max(5000).optional(),
  },
  async ({ title, status, priority, due_date, notes }) => {
    await consumeToken();

    const fields: Record<string, unknown> = {
      'Task Name': title,
      'Status':    status,
    };
    if (priority !== undefined) fields['Priority Score'] = priority;
    if (due_date) fields['Due Date'] = due_date;
    if (notes)    fields['Description'] = notes;

    // typecast: false (default) — don't let Airtable coerce values
    // Status is validated above, so we know it matches an existing option
    const record = await base(TASKS_TABLE_ID).create(fields, { typecast: false });

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({ id: record.id, fields: record.fields }),
      }],
    };
  }
);

Expanding linked records

A Linked Record field in Airtable stores an array of record IDs from another table. The REST API returns these IDs as strings in the field value — it does not auto-expand the linked records. You need a separate API call to the linked table to fetch the field values of those records. Design your MCP tools to handle this two-step pattern explicitly.

const PROJECTS_TABLE_ID = 'tblPROJECTS12345';  // table linked to tasks

server.tool(
  'airtable_get_task_with_project',
  { record_id: z.string().startsWith('rec') },
  async ({ record_id }) => {
    await consumeToken();

    // Step 1: fetch the task record
    const task = await base(TASKS_TABLE_ID).find(record_id);

    // Step 2: expand linked "Project" records
    // task.fields['Project'] is an array of record IDs like ["recXXX", "recYYY"]
    const linkedProjectIds = (task.fields['Project'] as string[] | undefined) ?? [];

    let projects: Array<{ id: string; name: string; status: string }> = [];

    if (linkedProjectIds.length > 0) {
      await consumeToken();

      // Fetch all linked records in a single query using filterByFormula with FIND
      const ids    = linkedProjectIds.map(id => `RECORD_ID()="${id}"`).join(',');
      const filter = linkedProjectIds.length === 1
        ? `RECORD_ID()="${linkedProjectIds[0]}"`
        : `OR(${ids})`;

      const projectRecords = await base(PROJECTS_TABLE_ID).select({
        filterByFormula: filter,
        fields: ['Project Name', 'Status'],
      }).all();

      projects = projectRecords.map(p => ({
        id:     p.id,
        name:   p.fields['Project Name'] as string,
        status: p.fields['Status'] as string,
      }));
    }

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({
          task:     { id: task.id, fields: task.fields },
          projects,
        }),
      }],
    };
  }
);

// Attachment fields — URLs expire after ~2 hours
// Re-fetch the record when you need a fresh download URL
server.tool(
  'airtable_get_attachment_url',
  { record_id: z.string(), field_name: z.string() },
  async ({ record_id, field_name }) => {
    await consumeToken();

    // Always fetch fresh — never use a cached attachment URL
    const record = await base(TASKS_TABLE_ID).find(record_id);
    const attachments = record.fields[field_name] as Array<{
      id: string; url: string; filename: string; size: number;
    }> | undefined;

    if (!attachments || attachments.length === 0) {
      return { content: [{ type: 'text', text: JSON.stringify([]) }] };
    }

    // Return URLs immediately — caller must use them within ~2 hours
    return {
      content: [{
        type: 'text',
        text: JSON.stringify(attachments.map(a => ({
          id:       a.id,
          filename: a.filename,
          url:      a.url,     // temporary — expires ~2h
          size:     a.size,
          expires_note: 'URL expires in ~2 hours; re-call this tool to get a fresh URL',
        }))),
      }],
    };
  }
);

When fetching linked records in bulk, the OR(RECORD_ID()="recA",RECORD_ID()="recB") formula pattern retrieves multiple specific records in one request instead of one request per record. This keeps you under the 5 req/s rate limit even when a task links to many project records.

Batch create and update limits

Airtable's REST API accepts a maximum of 10 records per create or update request. To insert 50 records, you must make 5 separate requests. Batch them into chunks of 10 and respect the 5 req/s rate limit between chunks — otherwise you'll receive 429 responses and need to retry with exponential backoff.

function chunk<T>(arr: T[], size: number): T[][] {
  const out: T[][] = [];
  for (let i = 0; i < arr.length; i += size) out.push(arr.slice(i, i + size));
  return out;
}

server.tool(
  'airtable_bulk_create',
  {
    records: z.array(z.object({
      title:  z.string(),
      status: z.enum(['Open', 'In Progress', 'Done', 'Blocked']),
    })).min(1).max(100),
  },
  async ({ records }) => {
    const batches = chunk(records, 10);  // max 10 per Airtable request
    const created: string[] = [];

    for (const batch of batches) {
      await consumeToken();

      const result = await base(TASKS_TABLE_ID).create(
        batch.map(r => ({
          fields: {
            'Task Name': r.title,
            'Status':    r.status,
          },
        })),
        { typecast: false }
      );

      created.push(...result.map(r => r.id));

      // Small delay between batches to stay well under the 5 req/s limit
      if (batches.length > 1) await new Promise(r => setTimeout(r, 250));
    }

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({ created_count: created.length, record_ids: created }),
      }],
    };
  }
);