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
| Name | Required | Description | Default |
|---|---|---|---|
| base_id | Yes | The Airtable base ID | |
| table_id | Yes | The table ID or name | |
| view | No | View name or ID | |
| filter_by_formula | No | Airtable formula for filtering | |
| sort | No | Sort configuration - array of {field: string, direction: 'asc'|'desc'} | |
| fields | No | Specific fields to include (field name or array of field names) |
Implementation Reference
- src/airtable_mcp/mcp/server.py:224-284 (handler)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
- src/airtable_mcp/api/models.py:64-69 (schema)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
- src/airtable_mcp/mcp/server.py:224-284 (registration)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 ]