Airtable OAuth MCP Server

Instructions

Implementation Reference

• src/airtable_mcp/mcp/server.py:360-388 (handler)

  MCP tool handler function for update_records. Decorated with @self.mcp.tool() which registers it as an MCP tool. Delegates to AirtableClient.update_records() and formats the response.

  @self.mcp.tool(description="Update multiple records")
  async def update_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 record updates")
      ],
      typecast: Annotated[
          bool, Field(description="Enable automatic data conversion")
      ] = False,
  ) -> list[dict[str, Any]]:
      """Update multiple records in a table."""
      client = await self._get_authenticated_client()
  
      updated_records = await client.update_records(
          base_id,
          table_id,
          records,
          typecast,
      )
  
      return [
          {
              "id": record.id,
              "fields": record.fields,
              "createdTime": record.created_time,
          }
          for record in updated_records
      ]

• src/airtable_mcp/api/client.py:407-439 (helper)

  AirtableClient helper method that makes the actual PATCH request to Airtable API /v0/{base}/{table} endpoint to update records. Uses UpdateRecordsRequest model and handles authentication, rate limiting, and errors.

  async def update_records(
      self,
      base_id: str,
      table_id: str,
      records: list[dict[str, Any]],
      typecast: bool = False,
  ) -> list[AirtableRecord]:
      """Update existing records in a table.
  
      Args:
          base_id: The Airtable base ID
          table_id: The table ID or name
          records: List of record updates (each should have 'id' and 'fields' keys)
          typecast: Whether to enable automatic data conversion
  
      Returns:
          List of updated records
      """
      logger.info(f"Updating {len(records)} records in {base_id}/{table_id}")
  
      request_data = UpdateRecordsRequest(
          records=records,
          typecast=typecast,
      )
  
      response = await self._make_request(
          "PATCH",
          f"/v0/{base_id}/{table_id}",
          data=request_data.model_dump(by_alias=True, exclude_none=True),
          response_model=UpdateRecordsResponse,
      )
  
      return response.records

• src/airtable_mcp/api/models.py:84-95 (schema)

  Pydantic models defining the request and response structure for the update_records API operation.

  class UpdateRecordsRequest(BaseModel):
      """Request for updating records."""
  
      records: list[dict[str, Any]]
      typecast: bool | None = False
  
  
  class UpdateRecordsResponse(BaseModel):
      """Response from updating records."""
  
      records: list[AirtableRecord]

• src/airtable_mcp/mcp/schemas.py:84-93 (schema)

  Pydantic schema defining arguments for the MCP update_records tool (though not directly used in server.py).

  class UpdateRecordsArgs(BaseArgs):
      """Arguments for update_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 record updates")
      typecast: bool | None = Field(
          default=False, description="Enable automatic data conversion"
      )