list_records
Retrieve and filter records from an Airtable table using view, formula, sorting, and field selection. Integrates with the Airtable OAuth MCP Server for secure data access.
Instructions
List records from a table with optional filtering
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| base_id | Yes | The Airtable base ID | |
| fields | No | Specific fields to include (field name or array of field names) | |
| filter_by_formula | No | Airtable formula for filtering | |
| sort | No | Sort configuration - array of {field: string, direction: 'asc'|'desc'} | |
| table_id | Yes | The table ID or name | |
| view | No | View name or ID |
Implementation Reference
- src/airtable_mcp/mcp/server.py:225-285 (handler)The MCP tool handler and registration for 'list_records'. Uses FastMCP @tool decorator to register and defines the execution logic: authenticates client, normalizes parameters into ListRecordsOptions, calls AirtableClient.list_records(), and formats the response.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 input schema model ListRecordsArgs defining the parameters for the list_records tool, matching the handler's Annotated fields.class ListRecordsArgs(BaseArgs): """Arguments for list_records tool.""" base_id: str = Field(description="The Airtable base ID") table_id: str = Field(description="The table ID or name") view: str | None = Field(default=None, description="View name or ID") max_records: int | None = Field( default=None, description="Maximum number of records to return" ) filter_by_formula: str | None = Field( default=None, description="Airtable formula for filtering" ) sort: list[dict[str, str]] | None = Field( default=None, description="Sort configuration" ) fields: list[str] | None = Field( default=None, description="Specific fields to include" )
- AirtableClient.list_records helper method that handles pagination, rate limiting, and API calls to fetch records from Airtable, used by the MCP handler.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 API list records response, used internally by the client.class ListRecordsResponse(BaseModel): """Response from listing records.""" records: list[AirtableRecord] offset: str | None = None
- Pydantic model for list records options, constructed in MCP handler and 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)