SpotifyLikedSongs

Instructions

Implementation Reference

• src/spotify_mcp/server.py:240-277 (handler)

  Handler for SpotifyLikedSongs tool - processes 'get' and 'get_with_genres' actions, calls get_liked_songs API method and optionally enriches tracks with genre information

  case "LikedSongs":
      action = arguments.get("action")
      limit = arguments.get("limit", 0)
  
      match action:
          case "get":
              tracks = await asyncio.to_thread(spotify_client.get_liked_songs, limit=limit)
              return [types.TextContent(
                  type="text",
                  text=json.dumps({"total": len(tracks), "tracks": tracks}, indent=2)
              )]
  
          case "get_with_genres":
              tracks = await asyncio.to_thread(spotify_client.get_liked_songs, limit=limit)
  
              all_artist_ids = set()
              for track in tracks:
                  for aid in track.get('artist_ids', []):
                      all_artist_ids.add(aid)
  
              genres_map = await asyncio.to_thread(
                  spotify_client.get_artists_genres, 
                  list(all_artist_ids)
              )
  
              for track in tracks:
                  track_genres = set()
                  for aid in track.get('artist_ids', []):
                      for genre in genres_map.get(aid, []):
                          track_genres.add(genre)
                  track['genres'] = list(track_genres)
                  if 'artist_ids' in track:
                      del track['artist_ids']
  
              return [types.TextContent(
                  type="text",
                  text=json.dumps({"total": len(tracks), "tracks": tracks}, indent=2)
              )]

• src/spotify_mcp/spotify_api.py:203-231 (helper)

  Core implementation of get_liked_songs method - fetches user's saved tracks from Spotify API with pagination support, includes added_at timestamp and artist_ids

  def get_liked_songs(self, limit: int = 0) -> List[Dict]:
      """Fetch user's liked/saved songs with pagination.
  
      Args:
          limit: Max songs to return. 0 means all songs.
      """
      if not self.auth_ok(): self.auth_refresh()
      all_tracks = []
      offset = 0
      batch_size = 50
      while True:
          results = self.sp.current_user_saved_tracks(limit=batch_size, offset=offset)
          if not results or not results.get('items'):
              break
          for item in results['items']:
              track = item.get('track')
              if not track:
                  continue
              track_info = utils.parse_track(track)
              track_info['added_at'] = item.get('added_at')
              artist_ids = [a['id'] for a in track.get('artists', []) if a.get('id')]
              track_info['artist_ids'] = artist_ids
              all_tracks.append(track_info)
              if 0 < limit <= len(all_tracks):
                  return all_tracks[:limit]
          offset += batch_size
          if not results.get('next'):
              break
      return all_tracks

• src/spotify_mcp/server.py:81-84 (schema)

  LikedSongs schema definition - defines action field ('get' or 'get_with_genres') and optional limit parameter, inherits from ToolModel which prefixes name with 'Spotify'

  class LikedSongs(ToolModel):
      """Get user's liked (saved) songs from Spotify library."""
      action: str = Field(description="Action to perform: 'get' or 'get_with_genres'.")
      limit: Optional[int] = Field(default=0, description="Max number of songs to return. 0 for all songs.")

• src/spotify_mcp/server.py:103-103 (registration)

  Registration of LikedSongs tool in handle_list_tools() function - converts schema to MCP tool definition

  LikedSongs.as_tool(),

• src/spotify_mcp/spotify_api.py:236-253 (helper)

  get_artists_genres helper method - used by get_with_genres action to fetch genre information for all artists in liked songs using parallel fetching

  def get_artists_genres(self, artist_ids: List[str]) -> Dict[str, List[str]]:
      if not self.auth_ok(): self.auth_refresh()
      genres_map = {}
      
      def fetch_artist(aid):
          try:
              artist = self.sp._get(f"artists/{aid}")
              return aid, artist.get('genres', []) if artist else []
          except Exception as e:
              self.logger.error(f"Error fetching artist {aid}: {str(e)}")
              return aid, []
  
      with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
          results = executor.map(fetch_artist, artist_ids)
          for aid, genres in results:
              genres_map[aid] = genres
              
      return genres_map