create_records
Create multiple records in Airtable by specifying a base ID, table ID, and record data. Supports automatic data conversion for streamlined integration.
Instructions
Create multiple records
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| base_id | Yes | The Airtable base ID | |
| records | Yes | List of records to create | |
| table_id | Yes | The table ID or name | |
| typecast | No | Enable automatic data conversion |
Implementation Reference
- src/airtable_mcp/mcp/server.py:330-358 (handler)MCP tool 'create_records' handler function. Registers the tool via @mcp.tool decorator, defines input schema via Annotated[Field], authenticates client, calls AirtableClient.create_records, and formats response.@self.mcp.tool(description="Create multiple records") async def create_records( base_id: Annotated[str, Field(description="The Airtable base ID")], table_id: Annotated[str, Field(description="The table ID or name")], records: Annotated[ list[dict[str, Any]], Field(description="List of records to create") ], typecast: Annotated[ bool, Field(description="Enable automatic data conversion") ] = False, ) -> list[dict[str, Any]]: """Create multiple records in a table.""" client = await self._get_authenticated_client() created_records = await client.create_records( base_id, table_id, records, typecast, ) return [ { "id": record.id, "fields": record.fields, "createdTime": record.created_time, } for record in created_records ]
- Pydantic schema class for create_records tool input arguments, used in tests and documentation.class CreateRecordsArgs(BaseArgs): """Arguments for create_records tool.""" base_id: str = Field(description="The Airtable base ID") table_id: str = Field(description="The table ID or name") records: list[dict[str, Any]] = Field(description="List of records to create") typecast: bool | None = Field( default=False, description="Enable automatic data conversion" )
- Core AirtableClient helper method implementing the HTTP POST to Airtable API for creating records, with rate limiting, authentication, retries, and Pydantic request/response models.async def create_records( self, base_id: str, table_id: str, records: list[dict[str, Any]], typecast: bool = False, ) -> list[AirtableRecord]: """Create new records in a table. Args: base_id: The Airtable base ID table_id: The table ID or name records: List of record data (each should have 'fields' key) typecast: Whether to enable automatic data conversion Returns: List of created records """ logger.info(f"Creating {len(records)} records in {base_id}/{table_id}") request_data = CreateRecordsRequest( records=records, typecast=typecast, ) response = await self._make_request( "POST", f"/v0/{base_id}/{table_id}", data=request_data.model_dump(by_alias=True, exclude_none=True), response_model=CreateRecordsResponse, ) return response.records
- Helper method in the MCP server to obtain an authenticated AirtableClient instance using the OAuth access token from the request context.async def _get_authenticated_client(self) -> AirtableClient: """Get an authenticated Airtable client using access token from context. Returns: Authenticated AirtableClient instance Raises: AirtableAuthError: If authentication fails """ # Get access token from authentication context (set by middleware) access_token = AuthContext.require_access_token() try: # Use the OAuth provider provider = self._get_oauth_provider() if not provider: raise AirtableAuthError( "OAuth provider not available - ensure OAuth endpoints are enabled and credentials are configured" ) # Set the access token in the provider provider.access_token = access_token return AirtableClient(provider) except Exception as e: logger.error(f"Failed to create authenticated client: {e}") raise AirtableAuthError(f"Authentication failed: {e}") from e