Skip to main content
Glama
onimsha

Airtable OAuth MCP Server

by onimsha

list_records

Retrieve records from an Airtable table with options to filter, sort, and select specific fields for data analysis and management.

Instructions

List records from a table with optional filtering

Input Schema

TableJSON Schema
NameRequiredDescriptionDefault
base_idYesThe Airtable base ID
table_idYesThe table ID or name
viewNoView name or ID
filter_by_formulaNoAirtable formula for filtering
sortNoSort configuration - array of {field: string, direction: 'asc'|'desc'}
fieldsNoSpecific fields to include (field name or array of field names)

Implementation Reference

  • MCP tool handler for 'list_records': processes parameters including special handling for 'fields' (JSON string or list), creates ListRecordsOptions, calls AirtableClient.list_records, and returns formatted records.
    @self.mcp.tool(description="List records from a table with optional filtering")
    async def list_records(
        base_id: Annotated[str, Field(description="The Airtable base ID")],
        table_id: Annotated[str, Field(description="The table ID or name")],
        view: Annotated[str | None, Field(description="View name or ID")] = None,
        filter_by_formula: Annotated[
            str | None, Field(description="Airtable formula for filtering")
        ] = None,
        sort: Annotated[
            list[dict[str, str]] | None,
            Field(
                description="Sort configuration - array of {field: string, direction: 'asc'|'desc'}"
            ),
        ] = None,
        fields: Annotated[
            str | list[str] | None,
            Field(
                description="Specific fields to include (field name or array of field names)"
            ),
        ] = None,
    ) -> list[dict[str, Any]]:
        """List records from a table with optional filtering and pagination."""
        client = await self._get_authenticated_client()
    
        # Normalize fields parameter
        normalized_fields = None
        if fields is not None:
            if isinstance(fields, str):
                # Check if it's a JSON-encoded array string
                if fields.startswith("[") and fields.endswith("]"):
                    try:
                        import json
    
                        normalized_fields = json.loads(fields)
                    except (json.JSONDecodeError, ValueError):
                        # If JSON parsing fails, treat as single field name
                        normalized_fields = [fields]
                else:
                    # Single field name
                    normalized_fields = [fields]
            else:
                normalized_fields = fields
    
        options = ListRecordsOptions(
            view=view,
            sort=sort,
            fields=normalized_fields,
        )
        if filter_by_formula:
            options.filter_by_formula = filter_by_formula
    
        records = await client.list_records(base_id, table_id, options)
    
        return [
            {
                "id": record.id,
                "fields": record.fields,
                "createdTime": record.created_time,
            }
            for record in records
        ]
  • Pydantic model ListRecordsOptions used to configure listing records (aliases for Airtable API params like filterByFormula, maxRecords). Passed to client.list_records.
    class ListRecordsOptions(BaseModel):
        """Options for listing records."""
    
        max_records: int | None = Field(alias="maxRecords", default=None)
        filter_by_formula: str | None = Field(alias="filterByFormula", default=None)
        view: str | None = None
        sort: list[dict[str, str]] | None = None
        fields: list[str] | None = None
        cell_format: str | None = Field(alias="cellFormat", default=None)
        time_zone: str | None = Field(alias="timeZone", default=None)
        user_locale: str | None = Field(alias="userLocale", default=None)
  • Helper method in AirtableClient that implements paginated record listing from Airtable API, handles query param formatting for arrays (fields[], sort[0][field]), rate limiting, and response pagination.
    async def list_records(
        self,
        base_id: str,
        table_id: str,
        options: ListRecordsOptions | None = None,
    ) -> list[AirtableRecord]:
        """List records from a table with pagination support.
    
        Args:
            base_id: The Airtable base ID
            table_id: The table ID or name
            options: Options for filtering and pagination
    
        Returns:
            List of records
        """
        logger.info(f"Listing records from {base_id}/{table_id}")
    
        all_records = []
        offset = None
    
        while True:
            # Build query parameters
            params = {}
            if options:
                # Convert Pydantic model to dict and filter None values
                option_dict = options.model_dump(by_alias=True, exclude_none=True)
    
                # Handle array parameters specially for Airtable API
                for key, value in option_dict.items():
                    if key == "fields" and isinstance(value, list):
                        # Airtable expects fields as multiple parameters: fields[]=field1&fields[]=field2
                        params[key] = value
                    elif key == "sort" and isinstance(value, list):
                        # Airtable expects sort as indexed parameters: sort[0][field]=Name&sort[0][direction]=asc
                        for i, sort_item in enumerate(value):
                            if isinstance(sort_item, dict):
                                for sort_key, sort_value in sort_item.items():
                                    params[f"sort[{i}][{sort_key}]"] = sort_value
                    else:
                        params[key] = value
    
            if offset:
                params["offset"] = offset
    
            # Debug: Log the parameters being sent to Airtable
            logger.debug(f"Sending request to Airtable with params: {params}")
    
            response = await self._make_request(
                "GET",
                f"/v0/{base_id}/{table_id}",
                params=params,
                response_model=ListRecordsResponse,
            )
    
            all_records.extend(response.records)
    
            # Check if there are more records
            if not response.offset:
                break
    
            offset = response.offset
    
            # Respect maxRecords if specified
            if (
                options
                and options.max_records
                and len(all_records) >= options.max_records
            ):
                all_records = all_records[: options.max_records]
                break
    
        return all_records
  • Pydantic model for Airtable list records API response, containing records list and pagination offset.
    class ListRecordsResponse(BaseModel):
        """Response from listing records."""
    
        records: list[AirtableRecord]
        offset: str | None = None
  • The @self.mcp.tool decorator registers the list_records function as an MCP tool.
    @self.mcp.tool(description="List records from a table with optional filtering")
    async def list_records(
        base_id: Annotated[str, Field(description="The Airtable base ID")],
        table_id: Annotated[str, Field(description="The table ID or name")],
        view: Annotated[str | None, Field(description="View name or ID")] = None,
        filter_by_formula: Annotated[
            str | None, Field(description="Airtable formula for filtering")
        ] = None,
        sort: Annotated[
            list[dict[str, str]] | None,
            Field(
                description="Sort configuration - array of {field: string, direction: 'asc'|'desc'}"
            ),
        ] = None,
        fields: Annotated[
            str | list[str] | None,
            Field(
                description="Specific fields to include (field name or array of field names)"
            ),
        ] = None,
    ) -> list[dict[str, Any]]:
        """List records from a table with optional filtering and pagination."""
        client = await self._get_authenticated_client()
    
        # Normalize fields parameter
        normalized_fields = None
        if fields is not None:
            if isinstance(fields, str):
                # Check if it's a JSON-encoded array string
                if fields.startswith("[") and fields.endswith("]"):
                    try:
                        import json
    
                        normalized_fields = json.loads(fields)
                    except (json.JSONDecodeError, ValueError):
                        # If JSON parsing fails, treat as single field name
                        normalized_fields = [fields]
                else:
                    # Single field name
                    normalized_fields = [fields]
            else:
                normalized_fields = fields
    
        options = ListRecordsOptions(
            view=view,
            sort=sort,
            fields=normalized_fields,
        )
        if filter_by_formula:
            options.filter_by_formula = filter_by_formula
    
        records = await client.list_records(base_id, table_id, options)
    
        return [
            {
                "id": record.id,
                "fields": record.fields,
                "createdTime": record.created_time,
            }
            for record in records
        ]

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/onimsha/airtable-mcp-server-oauth'

If you have feedback or need assistance with the MCP directory API, please join our Discord server