Airtable MCP

Verified
MIT License
Overview InspectNew Schema Related Servers Reviews Score
airtable_mcp/src/server.py
#!/usr/bin/env python3
"""
Airtable MCP Server
-------------------
This is a Model Context Protocol (MCP) server that exposes Airtable operations as tools.
"""
import os
import sys
import json
import asyncio
import logging
import argparse
from contextlib import asynccontextmanager
from typing import Any, Dict, List, Optional, AsyncIterator
from dotenv import load_dotenv

# Import MCP-related modules - will be available when run with Python 3.10+
try:
    from mcp.server.fastmcp import FastMCP
    from mcp.server import stdio
    from mcp.server.base import MCPRequestContext, MCPToolDefinition
except ImportError:
    print("Error: MCP SDK requires Python 3.10+")
    print("Please install Python 3.10 or newer and try again.")
    sys.exit(1)

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("airtable-mcp")

# Parse command line arguments
def parse_args():
    parser = argparse.ArgumentParser(description="Airtable MCP Server")
    parser.add_argument("--token", dest="api_token", help="Airtable Personal Access Token")
    parser.add_argument("--base", dest="base_id", help="Airtable Base ID")
    parser.add_argument("--port", type=int, default=8080, help="MCP server port for dev mode")
    parser.add_argument("--host", default="127.0.0.1", help="MCP server host for dev mode")
    parser.add_argument("--dev", action="store_true", help="Run in development mode")
    return parser.parse_args()

# Load environment variables as fallback
load_dotenv()

# Create MCP server
mcp = FastMCP("Airtable Tools")

# Server state will be initialized in main()
server_state = {
    "base_id": "",
    "token": "",
}

# Authentication middleware
@mcp.middleware
async def auth_middleware(context: MCPRequestContext, next):
    # Skip auth check for tool listing
    if context.operation == "list_tools":
        return await next(context)
    
    # Allow all operations without a token check - actual API calls will be checked later
    return await next(context)

# Helper functions for Airtable API calls
async def api_call(endpoint, method="GET", data=None, params=None):
    """Make an Airtable API call"""
    import requests
    
    # Check if token is available before making API calls
    if not server_state["token"]:
        return {"error": "No Airtable API token provided. Please set via --token or AIRTABLE_PERSONAL_ACCESS_TOKEN"}
    
    headers = {
        "Authorization": f"Bearer {server_state['token']}",
        "Content-Type": "application/json"
    }
    
    url = f"https://api.airtable.com/v0/{endpoint}"
    
    try:
        if method == "GET":
            response = requests.get(url, headers=headers, params=params)
        elif method == "POST":
            response = requests.post(url, headers=headers, json=data)
        elif method == "PATCH":
            response = requests.patch(url, headers=headers, json=data)
        elif method == "DELETE":
            response = requests.delete(url, headers=headers, params=params)
        else:
            raise ValueError(f"Unsupported method: {method}")
        
        response.raise_for_status()
        return response.json()
    except Exception as e:
        logger.error(f"API call error: {str(e)}")
        return {"error": str(e)}


# Define MCP tool functions

@mcp.tool()
async def list_bases() -> str:
    """List all accessible Airtable bases"""
    if not server_state["token"]:
        return "Please provide an Airtable API token to list your bases."
    
    result = await api_call("meta/bases")
    
    if "error" in result:
        return f"Error: {result['error']}"
    
    bases = result.get("bases", [])
    if not bases:
        return "No bases found accessible with your token."
    
    base_list = [f"{i+1}. {base['name']} (ID: {base['id']})" for i, base in enumerate(bases)]
    return "Available bases:\n" + "\n".join(base_list)


@mcp.tool()
async def list_tables(base_id: Optional[str] = None) -> str:
    """List all tables in the specified base or the default base"""
    if not server_state["token"]:
        return "Please provide an Airtable API token to list tables."
    
    base = base_id or server_state["base_id"]
    
    if not base:
        return "Error: No base ID provided. Please specify a base_id or set AIRTABLE_BASE_ID in your .env file."
    
    result = await api_call(f"meta/bases/{base}/tables")
    
    if "error" in result:
        return f"Error: {result['error']}"
    
    tables = result.get("tables", [])
    if not tables:
        return "No tables found in this base."
    
    table_list = [f"{i+1}. {table['name']} (ID: {table['id']}, Fields: {len(table.get('fields', []))})" 
                 for i, table in enumerate(tables)]
    return "Tables in this base:\n" + "\n".join(table_list)


@mcp.tool()
async def list_records(table_name: str, max_records: Optional[int] = 100, filter_formula: Optional[str] = None) -> str:
    """List records from a table with optional filtering"""
    if not server_state["token"]:
        return "Please provide an Airtable API token to list records."
    
    base = server_state["base_id"]
    
    if not base:
        return "Error: No base ID set. Please set a base ID."
    
    params = {"maxRecords": max_records}
    
    if filter_formula:
        params["filterByFormula"] = filter_formula
    
    result = await api_call(f"{base}/{table_name}", params=params)
    
    if "error" in result:
        return f"Error: {result['error']}"
    
    records = result.get("records", [])
    if not records:
        return "No records found in this table."
    
    # Format the records for display
    formatted_records = []
    for i, record in enumerate(records):
        record_id = record.get("id", "unknown")
        fields = record.get("fields", {})
        field_text = ", ".join([f"{k}: {v}" for k, v in fields.items()])
        formatted_records.append(f"{i+1}. ID: {record_id} - {field_text}")
    
    return "Records:\n" + "\n".join(formatted_records)


@mcp.tool()
async def get_record(table_name: str, record_id: str) -> str:
    """Get a specific record from a table"""
    if not server_state["token"]:
        return "Please provide an Airtable API token to get records."
    
    base = server_state["base_id"]
    
    if not base:
        return "Error: No base ID set. Please set a base ID."
    
    result = await api_call(f"{base}/{table_name}/{record_id}")
    
    if "error" in result:
        return f"Error: {result['error']}"
    
    fields = result.get("fields", {})
    if not fields:
        return f"Record {record_id} found but contains no fields."
    
    # Format the fields for display
    formatted_fields = []
    for key, value in fields.items():
        formatted_fields.append(f"{key}: {value}")
    
    return f"Record ID: {record_id}\n" + "\n".join(formatted_fields)


@mcp.tool()
async def create_records(table_name: str, records_json: str) -> str:
    """Create records in a table from JSON string"""
    if not server_state["token"]:
        return "Please provide an Airtable API token to create records."
    
    base = server_state["base_id"]
    
    if not base:
        return "Error: No base ID set. Please set a base ID."
    
    try:
        records_data = json.loads(records_json)
        
        # Format the records for Airtable API
        if not isinstance(records_data, list):
            records_data = [records_data]
        
        records = [{"fields": record} for record in records_data]
        
        data = {"records": records}
        result = await api_call(f"{base}/{table_name}", method="POST", data=data)
        
        if "error" in result:
            return f"Error: {result['error']}"
        
        created_records = result.get("records", [])
        return f"Successfully created {len(created_records)} records."
        
    except json.JSONDecodeError:
        return "Error: Invalid JSON format in records_json parameter."
    except Exception as e:
        return f"Error creating records: {str(e)}"


@mcp.tool()
async def update_records(table_name: str, records_json: str) -> str:
    """Update records in a table from JSON string"""
    if not server_state["token"]:
        return "Please provide an Airtable API token to update records."
    
    base = server_state["base_id"]
    
    if not base:
        return "Error: No base ID set. Please set a base ID."
    
    try:
        records_data = json.loads(records_json)
        
        # Format the records for Airtable API
        if not isinstance(records_data, list):
            records_data = [records_data]
        
        records = []
        for record in records_data:
            if "id" not in record:
                return "Error: Each record must have an 'id' field."
            
            rec_id = record.pop("id")
            fields = record.get("fields", record)  # Support both {id, fields} format and direct fields
            records.append({"id": rec_id, "fields": fields})
        
        data = {"records": records}
        result = await api_call(f"{base}/{table_name}", method="PATCH", data=data)
        
        if "error" in result:
            return f"Error: {result['error']}"
        
        updated_records = result.get("records", [])
        return f"Successfully updated {len(updated_records)} records."
        
    except json.JSONDecodeError:
        return "Error: Invalid JSON format in records_json parameter."
    except Exception as e:
        return f"Error updating records: {str(e)}"


@mcp.tool()
async def set_base_id(base_id: str) -> str:
    """Set the current Airtable base ID"""
    server_state["base_id"] = base_id
    return f"Base ID set to: {base_id}"


def main():
    """Run the MCP server"""
    try:
        # Parse command line arguments
        args = parse_args()
        
        # Set server state from command line args or fallback to env vars
        server_state["token"] = args.api_token or os.getenv("AIRTABLE_PERSONAL_ACCESS_TOKEN", "")
        server_state["base_id"] = args.base_id or os.getenv("AIRTABLE_BASE_ID", "")
        
        if not server_state["token"]:
            logger.warning("No Airtable API token provided. Please set via --token or AIRTABLE_PERSONAL_ACCESS_TOKEN")
            logger.info("Tool listing will work but API calls will require a token")
        
        # Setup asyncio event loop
        if sys.platform == 'win32':
            asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

        # Run the server
        if args.dev:
            # Development mode
            mcp.run(host=args.host, port=args.port)
        else:
            # Production mode - stdio interface for MCP
            mcp.run()
            
    except Exception as e:
        logger.error(f"Server error: {str(e)}")
        sys.exit(1)


if __name__ == "__main__":
    main()