spotify.py•4.91 kB
import time
import os
from typing import List
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
from db import get_user_tokens, update_access_token, get_first_user
from spotify_client import SpotifyClient
load_dotenv()
CLIENT_ID = os.getenv("SPOTIFY_CLIENT_ID")
CLIENT_SECRET = os.getenv("SPOTIFY_CLIENT_SECRET")
REDIRECT_URI = "http://127.0.0.1:5000/callback"
# Configure MCP server
mcp = FastMCP("spotify")
def get_current_user_id() -> str:
"""Get the current user's Spotify ID"""
user = get_first_user()
if not user:
raise Exception("No users found in database")
return user['spotify_id']
# Initialize Spotify client
spotify_client = SpotifyClient(
CLIENT_ID,
CLIENT_SECRET,
REDIRECT_URI
)
@mcp.tool("create_playlist")
def create_playlist(name: str, track_uris: List[str], description: str = "", public: bool = False) -> dict:
"""
Create a new playlist on Spotify and add tracks to it.
Args:
name: Name of the playlist
track_uris: List of Spotify track URIs to add to the playlist
description: Optional description for the playlist
public: Whether the playlist should be public (default: False)
"""
user_id = get_current_user_id()
tokens = get_user_tokens(user_id)
if not tokens:
raise Exception("No tokens found for user")
if tokens["token_expiry"] < time.time():
tokens = spotify_client.refresh_access_token(tokens["refresh_token"])
update_access_token(user_id, tokens["access_token"], tokens.get("expires_in", 3600))
return spotify_client.create_playlist_with_tracks(
access_token=tokens["access_token"],
user_id=user_id,
name=name,
track_uris=track_uris,
description=description
)
@mcp.tool("get_track_uris")
def get_track_uris(songs: List[dict]) -> List[str]:
"""
Look up Spotify track URIs for a list of songs.
Args:
songs: List of dictionaries containing song information.
Each dictionary should have 'name' and 'artist' keys.
Example: [{"name": "Yesterday", "artist": "The Beatles"}]
Returns:
List of Spotify track URIs for the found songs.
Songs that couldn't be found will be skipped.
"""
user_id = get_current_user_id()
tokens = get_user_tokens(user_id)
if not tokens:
raise Exception("No tokens found for user")
if tokens["token_expiry"] < time.time():
tokens = spotify_client.refresh_access_token(tokens["refresh_token"])
update_access_token(user_id, tokens["access_token"], tokens.get("expires_in", 3600))
track_uris = []
for song in songs:
uri = spotify_client.get_track_uri(
access_token=tokens["access_token"],
artist=song["artist"],
song_name=song["name"]
)
if uri:
track_uris.append(uri)
return track_uris
@mcp.tool("update_playlist")
def update_playlist(playlist_id: str, name: str = None, track_uris: List[str] = None, description: str = None, public: bool = None) -> dict:
"""
Update an existing Spotify playlist's details and/or tracks.
Args:
playlist_id: The Spotify ID of the playlist to update
name: New name for the playlist (optional)
track_uris: New list of track URIs to replace the playlist's tracks (optional)
description: New description for the playlist (optional)
public: New public/private status for the playlist (optional)
"""
user_id = get_current_user_id()
tokens = get_user_tokens(user_id)
if not tokens:
raise Exception("No tokens found for user")
if tokens["token_expiry"] < time.time():
tokens = spotify_client.refresh_access_token(tokens["refresh_token"])
update_access_token(user_id, tokens["access_token"], tokens.get("expires_in", 3600))
result = {}
# Update playlist details if any are provided
if any([name, description, public is not None]):
details_result = spotify_client.update_playlist_details(
access_token=tokens["access_token"],
playlist_id=playlist_id,
name=name,
description=description,
public=public
)
if details_result:
result.update(details_result)
else:
return {"error": "Failed to update playlist details"}
# Update tracks if provided
if track_uris is not None:
tracks_updated = spotify_client.replace_playlist_tracks(
access_token=tokens["access_token"],
playlist_id=playlist_id,
track_uris=track_uris
)
if not tracks_updated:
return {"error": "Failed to update playlist tracks"}
result["tracks_updated"] = True
return result
if __name__ == "__main__":
mcp.run(transport='stdio')