# Directory Structure
```
├── .gitignore
├── .python-version
├── LICENSE
├── media
│ └── take5.mp4
├── pyproject.toml
├── README.md
├── src
│ └── spotify_mcp
│ ├── __init__.py
│ ├── server.py
│ ├── spotify_api.py
│ └── utils.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.12
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
.cache
.DS_Store
__pycache__/
.env
.idea/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# spotify-mcp MCP server
MCP project to connect Claude with Spotify. Built on top of [spotipy-dev's API](https://github.com/spotipy-dev/spotipy/tree/2.24.0).
## Features
- Start, pause, and skip playback
- Search for tracks/albums/artists/playlists
- Get info about a track/album/artist/playlist
- Manage the Spotify queue
- Manage, create, and update playlists
## Demo
<details>
<summary>
Video -- turn on audio
</summary>
https://github.com/user-attachments/assets/20ee1f92-f3e3-4dfa-b945-ca57bc1e0894
</details>
## Configuration
### Getting Spotify API Keys
Create an account on [developer.spotify.com](https://developer.spotify.com/). Navigate to [the dashboard](https://developer.spotify.com/dashboard).
Create an app with redirect_uri as http://127.0.0.1:8080/callback.
You can choose any port you want but you must use http and an explicit loopback address (IPv4 or IPv6).
See [here](https://developer.spotify.com/documentation/web-api/concepts/redirect_uri) for more info/troubleshooting.
You may have to restart your MCP environment (e.g. Claude Desktop) once or twice before it works.
### Locating MCP Config
For Cursor, Claude Desktop, or any other MCP-enabled client you will have to locate your config.
- Claude Desktop location on MacOS: `~/Library/Application\ Support/Claude/claude_desktop_config.json`
- Claude Desktop location on Windows: `%APPDATA%/Claude/claude_desktop_config.json`
### Run this project with uvx
Add this snippet to your MCP Config.
```json
{
"mcpServers": {
"spotify": {
"command": "uvx",
"args": [
"--python", "3.12",
"--from", "git+https://github.com/varunneal/spotify-mcp",
"spotify-mcp"
],
"env": {
"SPOTIFY_CLIENT_ID": YOUR_CLIENT_ID,
"SPOTIFY_CLIENT_SECRET": YOUR_CLIENT_SECRET,
"SPOTIFY_REDIRECT_URI": "http://127.0.0.1:8080/callback"
}
}
}
}
```
### Run this project locally
Using UVX will open the spotify redirect URI for every tool call. To avoid this, you can run this project locally by cloning this repo:
```bash
git clone https://github.com/varunneal/spotify-mcp.git
```
Add it to your MCP Config like this:
```json
"spotify": {
"command": "uv",
"args": [
"--directory",
"/path/to/spotify-mcp",
"run",
"spotify-mcp"
],
"env": {
"SPOTIFY_CLIENT_ID": YOUR_CLIENT_ID,
"SPOTIFY_CLIENT_SECRET": YOUR_CLIENT_SECRET,
"SPOTIFY_REDIRECT_URI": "http://127.0.0.1:8080/callback"
}
}
```
### Troubleshooting
Please open an issue if you can't get this MCP working. Here are some tips:
1. Make sure `uv` is updated. I recommend version `>=0.54`.
2. If cloning locally, enable execution permisisons for the project: `chmod -R 755`.
3. Ensure you have Spotify premium (needed for running developer API).
This MCP will emit logs to std err (as specified in the MCP) spec. On Mac the Claude Desktop app should emit these logs
to `~/Library/Logs/Claude`.
On other platforms [you can find logs here](https://modelcontextprotocol.io/quickstart/user#getting-logs-from-claude-for-desktop).
You can launch the MCP Inspector via [`npm`](https://docs.npmjs.com/downloading-and-installing-node-js-and-npm) with this command:
```bash
npx @modelcontextprotocol/inspector uv --directory /path/to/spotify-mcp run spotify-mcp
```
Upon launching, the Inspector will display a URL that you can access in your browser to begin debugging.
## TODO
Unfortunately, a bunch of cool features have [now been deprecated](https://techcrunch.com/2024/11/27/spotify-cuts-developer-access-to-several-of-its-recommendation-features/)
from the Spotify API. Most new features will be relatively minor or for the health of the project:
- tests.
- ~~adding API support for managing playlists.~~
- adding API support for paginated search results/playlists/albums.
PRs appreciated! Thanks to @jamiew, @davidpadbury, @manncodes, @hyuma7, @aanurraj, @JJGO and others for contributions.
[//]: # (## Deployment)
[//]: # ((todo))
[//]: # (### Building and Publishing)
[//]: # ()
[//]: # (To prepare the package for distribution:)
[//]: # ()
[//]: # (1. Sync dependencies and update lockfile:)
[//]: # ()
[//]: # (```bash)
[//]: # (uv sync)
[//]: # (```)
[//]: # ()
[//]: # (2. Build package distributions:)
[//]: # ()
[//]: # (```bash)
[//]: # (uv build)
[//]: # (```)
[//]: # ()
[//]: # (This will create source and wheel distributions in the `dist/` directory.)
[//]: # ()
[//]: # (3. Publish to PyPI:)
[//]: # ()
[//]: # (```bash)
[//]: # (uv publish)
[//]: # (```)
[//]: # ()
[//]: # (Note: You'll need to set PyPI credentials via environment variables or command flags:)
[//]: # ()
[//]: # (- Token: `--token` or `UV_PUBLISH_TOKEN`)
[//]: # (- Or username/password: `--username`/`UV_PUBLISH_USERNAME` and `--password`/`UV_PUBLISH_PASSWORD`)
```
--------------------------------------------------------------------------------
/src/spotify_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
from . import server
import asyncio
def main():
"""Main entry point for the package."""
asyncio.run(server.main())
# Optionally expose other important items at package level
__all__ = ['main', 'server']
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "spotify-mcp"
version = "0.2.0"
description = "MCP spotify project"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"mcp==1.3.0",
"python-dotenv>=1.0.1",
"spotipy==2.24.0",
]
[[project.authors]]
name = "Varun Srivastava"
email = "[email protected]"
[build-system]
requires = [ "hatchling",]
build-backend = "hatchling.build"
[dependency-groups]
dev = [
]
[tool.uv.sources]
spotify-mcp = { workspace = true }
[project.scripts]
spotify-mcp = "spotify_mcp:main"
```
--------------------------------------------------------------------------------
/src/spotify_mcp/utils.py:
--------------------------------------------------------------------------------
```python
from collections import defaultdict
from typing import Optional, Dict
import functools
from typing import Callable, TypeVar
from typing import Optional, Dict
from urllib.parse import quote, urlparse, urlunparse
from requests import RequestException
T = TypeVar('T')
def normalize_redirect_uri(url: str) -> str:
if not url:
return url
parsed = urlparse(url)
# Convert localhost to 127.0.0.1
if parsed.netloc == 'localhost' or parsed.netloc.startswith('localhost:'):
port = ''
if ':' in parsed.netloc:
port = ':' + parsed.netloc.split(':')[1]
parsed = parsed._replace(netloc=f'127.0.0.1{port}')
return urlunparse(parsed)
def parse_track(track_item: dict, detailed=False) -> Optional[dict]:
if not track_item:
return None
narrowed_item = {
'name': track_item['name'],
'id': track_item['id'],
}
if 'is_playing' in track_item:
narrowed_item['is_playing'] = track_item['is_playing']
if detailed:
narrowed_item['album'] = parse_album(track_item.get('album'))
for k in ['track_number', 'duration_ms']:
narrowed_item[k] = track_item.get(k)
if not track_item.get('is_playable', True):
narrowed_item['is_playable'] = False
artists = [a['name'] for a in track_item['artists']]
if detailed:
artists = [parse_artist(a) for a in track_item['artists']]
if len(artists) == 1:
narrowed_item['artist'] = artists[0]
else:
narrowed_item['artists'] = artists
return narrowed_item
def parse_artist(artist_item: dict, detailed=False) -> Optional[dict]:
if not artist_item:
return None
narrowed_item = {
'name': artist_item['name'],
'id': artist_item['id'],
}
if detailed:
narrowed_item['genres'] = artist_item.get('genres')
return narrowed_item
def parse_playlist(playlist_item: dict, username, detailed=False) -> Optional[dict]:
if not playlist_item:
return None
narrowed_item = {
'name': playlist_item['name'],
'id': playlist_item['id'],
'owner': playlist_item['owner']['display_name'],
'user_is_owner': playlist_item['owner']['display_name'] == username,
'total_tracks': playlist_item['tracks']['total'],
}
if detailed:
narrowed_item['description'] = playlist_item.get('description')
tracks = []
for t in playlist_item['tracks']['items']:
tracks.append(parse_track(t['track']))
narrowed_item['tracks'] = tracks
return narrowed_item
def parse_album(album_item: dict, detailed=False) -> dict:
narrowed_item = {
'name': album_item['name'],
'id': album_item['id'],
}
artists = [a['name'] for a in album_item['artists']]
if detailed:
tracks = []
for t in album_item['tracks']['items']:
tracks.append(parse_track(t))
narrowed_item["tracks"] = tracks
artists = [parse_artist(a) for a in album_item['artists']]
for k in ['total_tracks', 'release_date', 'genres']:
narrowed_item[k] = album_item.get(k)
if len(artists) == 1:
narrowed_item['artist'] = artists[0]
else:
narrowed_item['artists'] = artists
return narrowed_item
def parse_search_results(results: Dict, qtype: str, username: Optional[str] = None):
_results = defaultdict(list)
# potential
# if username:
# _results['User Spotify URI'] = username
for q in qtype.split(","):
match q:
case "track":
for idx, item in enumerate(results['tracks']['items']):
if not item: continue
_results['tracks'].append(parse_track(item))
case "artist":
for idx, item in enumerate(results['artists']['items']):
if not item: continue
_results['artists'].append(parse_artist(item))
case "playlist":
for idx, item in enumerate(results['playlists']['items']):
if not item: continue
_results['playlists'].append(parse_playlist(item, username))
case "album":
for idx, item in enumerate(results['albums']['items']):
if not item: continue
_results['albums'].append(parse_album(item))
case _:
raise ValueError(f"Unknown qtype {qtype}")
return dict(_results)
def parse_tracks(items: Dict) -> list:
"""
Parse a list of track items and return a list of parsed tracks.
Args:
items: List of track items
Returns:
List of parsed tracks
"""
tracks = []
for idx, item in enumerate(items):
if not item:
continue
tracks.append(parse_track(item['track']))
return tracks
def build_search_query(base_query: str,
artist: Optional[str] = None,
track: Optional[str] = None,
album: Optional[str] = None,
year: Optional[str] = None,
year_range: Optional[tuple[int, int]] = None,
# upc: Optional[str] = None,
# isrc: Optional[str] = None,
genre: Optional[str] = None,
is_hipster: bool = False,
is_new: bool = False
) -> str:
"""
Build a search query string with optional filters.
Args:
base_query: Base search term
artist: Artist name filter
track: Track name filter
album: Album name filter
year: Specific year filter
year_range: Tuple of (start_year, end_year) for year range filter
genre: Genre filter
is_hipster: Filter for lowest 10% popularity albums
is_new: Filter for albums released in past two weeks
Returns:
Encoded query string with applied filters
"""
filters = []
if artist:
filters.append(f"artist:{artist}")
if track:
filters.append(f"track:{track}")
if album:
filters.append(f"album:{album}")
if year:
filters.append(f"year:{year}")
if year_range:
filters.append(f"year:{year_range[0]}-{year_range[1]}")
if genre:
filters.append(f"genre:{genre}")
if is_hipster:
filters.append("tag:hipster")
if is_new:
filters.append("tag:new")
query_parts = [base_query] + filters
return quote(" ".join(query_parts))
def validate(func: Callable[..., T]) -> Callable[..., T]:
"""
Decorator for Spotify API methods that handles authentication and device validation.
- Checks and refreshes authentication if needed
- Validates active device and retries with candidate device if needed
"""
@functools.wraps(func)
def wrapper(self, *args, **kwargs):
# Handle authentication
if not self.auth_ok():
self.auth_refresh()
# Handle device validation
if not self.is_active_device():
kwargs['device'] = self._get_candidate_device()
# TODO: try-except RequestException
return func(self, *args, **kwargs)
return wrapper
def ensure_username(func):
"""
Decorator to ensure that the username is set before calling the function.
"""
@functools.wraps(func)
def wrapper(self, *args, **kwargs):
if self.username is None:
self.set_username()
return func(self, *args, **kwargs)
return wrapper
```
--------------------------------------------------------------------------------
/src/spotify_mcp/spotify_api.py:
--------------------------------------------------------------------------------
```python
import logging
import os
from typing import Optional, Dict, List
import spotipy
from dotenv import load_dotenv
from spotipy.cache_handler import CacheFileHandler
from spotipy.oauth2 import SpotifyOAuth
from . import utils
load_dotenv()
CLIENT_ID = os.getenv("SPOTIFY_CLIENT_ID")
CLIENT_SECRET = os.getenv("SPOTIFY_CLIENT_SECRET")
REDIRECT_URI = os.getenv("SPOTIFY_REDIRECT_URI")
# Normalize the redirect URI to meet Spotify's requirements
if REDIRECT_URI:
REDIRECT_URI = utils.normalize_redirect_uri(REDIRECT_URI)
SCOPES = ["user-read-currently-playing", "user-read-playback-state", "user-read-currently-playing", # spotify connect
"app-remote-control", "streaming", # playback
"playlist-read-private", "playlist-read-collaborative", "playlist-modify-private", "playlist-modify-public",
# playlists
"user-read-playback-position", "user-top-read", "user-read-recently-played", # listening history
"user-library-modify", "user-library-read", # library
]
class Client:
def __init__(self, logger: logging.Logger):
"""Initialize Spotify client with necessary permissions"""
self.logger = logger
scope = "user-library-read,user-read-playback-state,user-modify-playback-state,user-read-currently-playing,playlist-read-private,playlist-read-collaborative,playlist-modify-private,playlist-modify-public"
try:
self.sp = spotipy.Spotify(auth_manager=SpotifyOAuth(
scope=scope,
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET,
redirect_uri=REDIRECT_URI))
self.auth_manager: SpotifyOAuth = self.sp.auth_manager
self.cache_handler: CacheFileHandler = self.auth_manager.cache_handler
except Exception as e:
self.logger.error(f"Failed to initialize Spotify client: {str(e)}")
raise
self.username = None
@utils.validate
def set_username(self, device=None):
self.username = self.sp.current_user()['display_name']
@utils.validate
def search(self, query: str, qtype: str = 'track', limit=10, device=None):
"""
Searches based of query term.
- query: query term
- qtype: the types of items to return. One or more of 'artist', 'album', 'track', 'playlist'.
If multiple types are desired, pass in a comma separated string; e.g. 'track,album'
- limit: max # items to return
"""
if self.username is None:
self.set_username()
results = self.sp.search(q=query, limit=limit, type=qtype)
if not results:
raise ValueError("No search results found.")
return utils.parse_search_results(results, qtype, self.username)
def recommendations(self, artists: Optional[List] = None, tracks: Optional[List] = None, limit=20):
# doesnt work
recs = self.sp.recommendations(seed_artists=artists, seed_tracks=tracks, limit=limit)
return recs
def get_info(self, item_uri: str) -> dict:
"""
Returns more info about item.
- item_uri: uri. Looks like 'spotify:track:xxxxxx', 'spotify:album:xxxxxx', etc.
"""
_, qtype, item_id = item_uri.split(":")
match qtype:
case 'track':
return utils.parse_track(self.sp.track(item_id), detailed=True)
case 'album':
album_info = utils.parse_album(self.sp.album(item_id), detailed=True)
return album_info
case 'artist':
artist_info = utils.parse_artist(self.sp.artist(item_id), detailed=True)
albums = self.sp.artist_albums(item_id)
top_tracks = self.sp.artist_top_tracks(item_id)['tracks']
albums_and_tracks = {
'albums': albums,
'tracks': {'items': top_tracks}
}
parsed_info = utils.parse_search_results(albums_and_tracks, qtype="album,track")
artist_info['top_tracks'] = parsed_info['tracks']
artist_info['albums'] = parsed_info['albums']
return artist_info
case 'playlist':
if self.username is None:
self.set_username()
playlist = self.sp.playlist(item_id)
self.logger.info(f"playlist info is {playlist}")
playlist_info = utils.parse_playlist(playlist, self.username, detailed=True)
return playlist_info
raise ValueError(f"Unknown qtype {qtype}")
def get_current_track(self) -> Optional[Dict]:
"""Get information about the currently playing track"""
try:
# current_playback vs current_user_playing_track?
current = self.sp.current_user_playing_track()
if not current:
self.logger.info("No playback session found")
return None
if current.get('currently_playing_type') != 'track':
self.logger.info("Current playback is not a track")
return None
track_info = utils.parse_track(current['item'])
if 'is_playing' in current:
track_info['is_playing'] = current['is_playing']
self.logger.info(
f"Current track: {track_info.get('name', 'Unknown')} by {track_info.get('artist', 'Unknown')}")
return track_info
except Exception as e:
self.logger.error("Error getting current track info.")
raise
@utils.validate
def start_playback(self, spotify_uri=None, device=None):
"""
Starts spotify playback of uri. If spotify_uri is omitted, resumes current playback.
- spotify_uri: ID of resource to play, or None. Typically looks like 'spotify:track:xxxxxx' or 'spotify:album:xxxxxx'.
"""
try:
self.logger.info(f"Starting playback for spotify_uri: {spotify_uri} on {device}")
if not spotify_uri:
if self.is_track_playing():
self.logger.info("No track_id provided and playback already active.")
return
if not self.get_current_track():
raise ValueError("No track_id provided and no current playback to resume.")
if spotify_uri is not None:
if spotify_uri.startswith('spotify:track:'):
uris = [spotify_uri]
context_uri = None
else:
uris = None
context_uri = spotify_uri
else:
uris = None
context_uri = None
device_id = device.get('id') if device else None
self.logger.info(f"Starting playback of on {device}: context_uri={context_uri}, uris={uris}")
result = self.sp.start_playback(uris=uris, context_uri=context_uri, device_id=device_id)
self.logger.info(f"Playback result: {result}")
return result
except Exception as e:
self.logger.error(f"Error starting playback: {str(e)}.")
raise
@utils.validate
def pause_playback(self, device=None):
"""Pauses playback."""
playback = self.sp.current_playback()
if playback and playback.get('is_playing'):
self.sp.pause_playback(device.get('id') if device else None)
@utils.validate
def add_to_queue(self, track_id: str, device=None):
"""
Adds track to queue.
- track_id: ID of track to play.
"""
self.sp.add_to_queue(track_id, device.get('id') if device else None)
@utils.validate
def get_queue(self, device=None):
"""Returns the current queue of tracks."""
queue_info = self.sp.queue()
queue_info['currently_playing'] = self.get_current_track()
queue_info['queue'] = [utils.parse_track(track) for track in queue_info.pop('queue')]
return queue_info
def get_liked_songs(self):
# todo
results = self.sp.current_user_saved_tracks()
for idx, item in enumerate(results['items']):
track = item['track']
print(idx, track['artists'][0]['name'], " – ", track['name'])
def is_track_playing(self) -> bool:
"""Returns if a track is actively playing."""
curr_track = self.get_current_track()
if not curr_track:
return False
if curr_track.get('is_playing'):
return True
return False
def get_current_user_playlists(self, limit=50) -> List[Dict]:
"""
Get current user's playlists.
- limit: Max number of playlists to return.
"""
playlists = self.sp.current_user_playlists()
if not playlists:
raise ValueError("No playlists found.")
return [utils.parse_playlist(playlist, self.username) for playlist in playlists['items']]
@utils.ensure_username
def get_playlist_tracks(self, playlist_id: str, limit=50) -> List[Dict]:
"""
Get tracks from a playlist.
- playlist_id: ID of the playlist to get tracks from.
- limit: Max number of tracks to return.
"""
playlist = self.sp.playlist(playlist_id)
if not playlist:
raise ValueError("No playlist found.")
return utils.parse_tracks(playlist['tracks']['items'])
@utils.ensure_username
def add_tracks_to_playlist(self, playlist_id: str, track_ids: List[str], position: Optional[int] = None):
"""
Add tracks to a playlist.
- playlist_id: ID of the playlist to modify.
- track_ids: List of track IDs to add.
- position: Position to insert the tracks at (optional).
"""
if not playlist_id:
raise ValueError("No playlist ID provided.")
if not track_ids:
raise ValueError("No track IDs provided.")
try:
response = self.sp.playlist_add_items(playlist_id, track_ids, position=position)
self.logger.info(f"Response from adding tracks: {track_ids} to playlist {playlist_id}: {response}")
except Exception as e:
self.logger.error(f"Error adding tracks to playlist: {str(e)}")
@utils.ensure_username
def remove_tracks_from_playlist(self, playlist_id: str, track_ids: List[str]):
"""
Remove tracks from a playlist.
- playlist_id: ID of the playlist to modify.
- track_ids: List of track IDs to remove.
"""
if not playlist_id:
raise ValueError("No playlist ID provided.")
if not track_ids:
raise ValueError("No track IDs provided.")
try:
response = self.sp.playlist_remove_all_occurrences_of_items(playlist_id, track_ids)
self.logger.info(f"Response from removing tracks: {track_ids} from playlist {playlist_id}: {response}")
except Exception as e:
self.logger.error(f"Error removing tracks from playlist: {str(e)}")
@utils.ensure_username
def create_playlist(self, name: str, description: Optional[str] = None, public: bool = True):
"""
Create a new playlist.
- name: Name for the playlist.
- description: Description for the playlist.
- public: Whether the playlist should be public.
"""
if not name:
raise ValueError("Playlist name is required.")
try:
user = self.sp.current_user()
user_id = user['id']
playlist = self.sp.user_playlist_create(
user=user_id,
name=name,
public=public,
description=description
)
self.logger.info(f"Created playlist: {name} (ID: {playlist['id']})")
return utils.parse_playlist(playlist, self.username, detailed=True)
except Exception as e:
self.logger.error(f"Error creating playlist: {str(e)}")
raise
@utils.ensure_username
def change_playlist_details(self, playlist_id: str, name: Optional[str] = None, description: Optional[str] = None):
"""
Change playlist details.
- playlist_id: ID of the playlist to modify.
- name: New name for the playlist.
- public: Whether the playlist should be public.
- description: New description for the playlist.
"""
if not playlist_id:
raise ValueError("No playlist ID provided.")
try:
response = self.sp.playlist_change_details(playlist_id, name=name, description=description)
self.logger.info(f"Response from changing playlist details: {response}")
except Exception as e:
self.logger.error(f"Error changing playlist details: {str(e)}")
def get_devices(self) -> dict:
return self.sp.devices()['devices']
def is_active_device(self):
return any([device.get('is_active') for device in self.get_devices()])
def _get_candidate_device(self):
devices = self.get_devices()
if not devices:
raise ConnectionError("No active device. Is Spotify open?")
for device in devices:
if device.get('is_active'):
return device
self.logger.info(f"No active device, assigning {devices[0]['name']}.")
return devices[0]
def auth_ok(self) -> bool:
try:
token = self.cache_handler.get_cached_token()
if token is None:
self.logger.info("Auth check result: no token exists")
return False
is_expired = self.auth_manager.is_token_expired(token)
self.logger.info(f"Auth check result: {'valid' if not is_expired else 'expired'}")
return not is_expired # Return True if token is NOT expired
except Exception as e:
self.logger.error(f"Error checking auth status: {str(e)}")
return False # Return False on error rather than raising
def auth_refresh(self):
self.auth_manager.validate_token(self.cache_handler.get_cached_token())
def skip_track(self, n=1):
# todo: Better error handling
for _ in range(n):
self.sp.next_track()
def previous_track(self):
self.sp.previous_track()
def seek_to_position(self, position_ms):
self.sp.seek_track(position_ms=position_ms)
def set_volume(self, volume_percent):
self.sp.volume(volume_percent)
```
--------------------------------------------------------------------------------
/src/spotify_mcp/server.py:
--------------------------------------------------------------------------------
```python
import asyncio
import base64
import os
import logging
import sys
from enum import Enum
import json
from typing import List, Optional, Tuple
from datetime import datetime
from pathlib import Path
import mcp.types as types
from mcp.server import NotificationOptions, Server # , stdio_server
import mcp.server.stdio
from pydantic import BaseModel, Field, AnyUrl
from spotipy import SpotifyException
from . import spotify_api
from .utils import normalize_redirect_uri
def setup_logger():
class Logger:
def info(self, message):
print(f"[INFO] {message}", file=sys.stderr)
def error(self, message):
print(f"[ERROR] {message}", file=sys.stderr)
return Logger()
logger = setup_logger()
# Normalize the redirect URI to meet Spotify's requirements
if spotify_api.REDIRECT_URI:
spotify_api.REDIRECT_URI = normalize_redirect_uri(spotify_api.REDIRECT_URI)
spotify_client = spotify_api.Client(logger)
server = Server("spotify-mcp")
# options =
class ToolModel(BaseModel):
@classmethod
def as_tool(cls):
return types.Tool(
name="Spotify" + cls.__name__,
description=cls.__doc__,
inputSchema=cls.model_json_schema()
)
class Playback(ToolModel):
"""Manages the current playback with the following actions:
- get: Get information about user's current track.
- start: Starts playing new item or resumes current playback if called with no uri.
- pause: Pauses current playback.
- skip: Skips current track.
"""
action: str = Field(description="Action to perform: 'get', 'start', 'pause' or 'skip'.")
spotify_uri: Optional[str] = Field(default=None, description="Spotify uri of item to play for 'start' action. " +
"If omitted, resumes current playback.")
num_skips: Optional[int] = Field(default=1, description="Number of tracks to skip for `skip` action.")
class Queue(ToolModel):
"""Manage the playback queue - get the queue or add tracks."""
action: str = Field(description="Action to perform: 'add' or 'get'.")
track_id: Optional[str] = Field(default=None, description="Track ID to add to queue (required for add action)")
class GetInfo(ToolModel):
"""Get detailed information about a Spotify item (track, album, artist, or playlist)."""
item_uri: str = Field(description="URI of the item to get information about. " +
"If 'playlist' or 'album', returns its tracks. " +
"If 'artist', returns albums and top tracks.")
class Search(ToolModel):
"""Search for tracks, albums, artists, or playlists on Spotify."""
query: str = Field(description="query term")
qtype: Optional[str] = Field(default="track",
description="Type of items to search for (track, album, artist, playlist, " +
"or comma-separated combination)")
limit: Optional[int] = Field(default=10, description="Maximum number of items to return")
class Playlist(ToolModel):
"""Manage Spotify playlists.
- get: Get a list of user's playlists.
- get_tracks: Get tracks in a specific playlist.
- add_tracks: Add tracks to a specific playlist.
- remove_tracks: Remove tracks from a specific playlist.
- change_details: Change details of a specific playlist.
- create: Create a new playlist.
"""
action: str = Field(
description="Action to perform: 'get', 'get_tracks', 'add_tracks', 'remove_tracks', 'change_details', 'create'.")
playlist_id: Optional[str] = Field(default=None, description="ID of the playlist to manage.")
track_ids: Optional[List[str]] = Field(default=None, description="List of track IDs to add/remove.")
name: Optional[str] = Field(default=None, description="Name for the playlist (required for create and change_details).")
description: Optional[str] = Field(default=None, description="Description for the playlist.")
public: Optional[bool] = Field(default=True, description="Whether the playlist should be public (for create action).")
@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
return []
@server.list_resources()
async def handle_list_resources() -> list[types.Resource]:
return []
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""List available tools."""
logger.info("Listing available tools")
# await server.request_context.session.send_notification("are you recieving this notification?")
tools = [
Playback.as_tool(),
Search.as_tool(),
Queue.as_tool(),
GetInfo.as_tool(),
Playlist.as_tool(),
]
logger.info(f"Available tools: {[tool.name for tool in tools]}")
return tools
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""Handle tool execution requests."""
logger.info(f"Tool called: {name} with arguments: {arguments}")
assert name[:7] == "Spotify", f"Unknown tool: {name}"
try:
match name[7:]:
case "Playback":
action = arguments.get("action")
match action:
case "get":
logger.info("Attempting to get current track")
curr_track = spotify_client.get_current_track()
if curr_track:
logger.info(f"Current track retrieved: {curr_track.get('name', 'Unknown')}")
return [types.TextContent(
type="text",
text=json.dumps(curr_track, indent=2)
)]
logger.info("No track currently playing")
return [types.TextContent(
type="text",
text="No track playing."
)]
case "start":
logger.info(f"Starting playback with arguments: {arguments}")
spotify_client.start_playback(spotify_uri=arguments.get("spotify_uri"))
logger.info("Playback started successfully")
return [types.TextContent(
type="text",
text="Playback starting."
)]
case "pause":
logger.info("Attempting to pause playback")
spotify_client.pause_playback()
logger.info("Playback paused successfully")
return [types.TextContent(
type="text",
text="Playback paused."
)]
case "skip":
num_skips = int(arguments.get("num_skips", 1))
logger.info(f"Skipping {num_skips} tracks.")
spotify_client.skip_track(n=num_skips)
return [types.TextContent(
type="text",
text="Skipped to next track."
)]
case "Search":
logger.info(f"Performing search with arguments: {arguments}")
search_results = spotify_client.search(
query=arguments.get("query", ""),
qtype=arguments.get("qtype", "track"),
limit=arguments.get("limit", 10)
)
logger.info("Search completed successfully.")
return [types.TextContent(
type="text",
text=json.dumps(search_results, indent=2)
)]
case "Queue":
logger.info(f"Queue operation with arguments: {arguments}")
action = arguments.get("action")
match action:
case "add":
track_id = arguments.get("track_id")
if not track_id:
logger.error("track_id is required for add to queue.")
return [types.TextContent(
type="text",
text="track_id is required for add action"
)]
spotify_client.add_to_queue(track_id)
return [types.TextContent(
type="text",
text=f"Track added to queue."
)]
case "get":
queue = spotify_client.get_queue()
return [types.TextContent(
type="text",
text=json.dumps(queue, indent=2)
)]
case _:
return [types.TextContent(
type="text",
text=f"Unknown queue action: {action}. Supported actions are: add, remove, and get."
)]
case "GetInfo":
logger.info(f"Getting item info with arguments: {arguments}")
item_info = spotify_client.get_info(
item_uri=arguments.get("item_uri")
)
return [types.TextContent(
type="text",
text=json.dumps(item_info, indent=2)
)]
case "Playlist":
logger.info(f"Playlist operation with arguments: {arguments}")
action = arguments.get("action")
match action:
case "get":
logger.info(f"Getting current user's playlists with arguments: {arguments}")
playlists = spotify_client.get_current_user_playlists()
return [types.TextContent(
type="text",
text=json.dumps(playlists, indent=2)
)]
case "get_tracks":
logger.info(f"Getting tracks in playlist with arguments: {arguments}")
if not arguments.get("playlist_id"):
logger.error("playlist_id is required for get_tracks action.")
return [types.TextContent(
type="text",
text="playlist_id is required for get_tracks action."
)]
tracks = spotify_client.get_playlist_tracks(arguments.get("playlist_id"))
return [types.TextContent(
type="text",
text=json.dumps(tracks, indent=2)
)]
case "add_tracks":
logger.info(f"Adding tracks to playlist with arguments: {arguments}")
track_ids = arguments.get("track_ids")
if isinstance(track_ids, str):
try:
track_ids = json.loads(track_ids) # Convert JSON string to Python list
except json.JSONDecodeError:
logger.error("track_ids must be a list or a valid JSON array.")
return [types.TextContent(
type="text",
text="Error: track_ids must be a list or a valid JSON array."
)]
spotify_client.add_tracks_to_playlist(
playlist_id=arguments.get("playlist_id"),
track_ids=track_ids
)
return [types.TextContent(
type="text",
text="Tracks added to playlist."
)]
case "remove_tracks":
logger.info(f"Removing tracks from playlist with arguments: {arguments}")
track_ids = arguments.get("track_ids")
if isinstance(track_ids, str):
try:
track_ids = json.loads(track_ids) # Convert JSON string to Python list
except json.JSONDecodeError:
logger.error("track_ids must be a list or a valid JSON array.")
return [types.TextContent(
type="text",
text="Error: track_ids must be a list or a valid JSON array."
)]
spotify_client.remove_tracks_from_playlist(
playlist_id=arguments.get("playlist_id"),
track_ids=track_ids
)
return [types.TextContent(
type="text",
text="Tracks removed from playlist."
)]
case "change_details":
logger.info(f"Changing playlist details with arguments: {arguments}")
if not arguments.get("playlist_id"):
logger.error("playlist_id is required for change_details action.")
return [types.TextContent(
type="text",
text="playlist_id is required for change_details action."
)]
if not arguments.get("name") and not arguments.get("description"):
logger.error("At least one of name, description or public is required.")
return [types.TextContent(
type="text",
text="At least one of name, description, public, or collaborative is required."
)]
spotify_client.change_playlist_details(
playlist_id=arguments.get("playlist_id"),
name=arguments.get("name"),
description=arguments.get("description")
)
return [types.TextContent(
type="text",
text="Playlist details changed."
)]
case "create":
logger.info(f"Creating playlist with arguments: {arguments}")
if not arguments.get("name"):
logger.error("name is required for create action.")
return [types.TextContent(
type="text",
text="name is required for create action."
)]
playlist = spotify_client.create_playlist(
name=arguments.get("name"),
description=arguments.get("description"),
public=arguments.get("public", True)
)
return [types.TextContent(
type="text",
text=json.dumps(playlist, indent=2)
)]
case _:
return [types.TextContent(
type="text",
text=f"Unknown playlist action: {action}."
"Supported actions are: get, get_tracks, add_tracks, remove_tracks, change_details, create."
)]
case _:
error_msg = f"Unknown tool: {name}"
logger.error(error_msg)
return [types.TextContent(
type="text",
text=error_msg
)]
except SpotifyException as se:
error_msg = f"Spotify Client error occurred: {str(se)}"
logger.error(error_msg)
return [types.TextContent(
type="text",
text=f"An error occurred with the Spotify Client: {str(se)}"
)]
except Exception as e:
error_msg = f"Unexpected error occurred: {str(e)}"
logger.error(error_msg)
return [types.TextContent(
type="text",
text=error_msg
)]
async def main():
try:
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
except Exception as e:
logger.error(f"Server error occurred: {str(e)}")
raise
```