Add curated Spotify prompts and profile resource (#4)

* feat: add curated Spotify prompts and profile resource

* fix: validate Spotify prompt and resource contracts

README.md

@@ -22,6 +22,22 @@ Streamable HTTP on loopback only.
 | `library_modify` | Save/remove/follow/unfollow up to 40 URIs per action | Changes library |
 | `listening_activity` | Read recent tracks and top tracks/artists | None |
 
+## Prompts and resources
+
+The server exposes four curated workflows that compose the nine tools without adding another API
+surface:
+
+| Prompt | Purpose |
+| --- | --- |
+| `catch_up_on_podcasts` | Prioritize unfinished or unplayed episodes from saved shows |
+| `weekly_music_recap` | Summarize patterns in recent plays and top music |
+| `build_playlist_for_mood` | Search and create a private playlist for a mood or activity |
+| `now_playing_briefing` | Produce a compact playback, device, progress, and queue summary |
+
+The intentionally small resource catalog contains `spotify://me`. It returns the current user's
+Spotify display name and stable `account_id`; dynamic playback, library, and playlist state remains
+behind tools.
+
 Podcast transcripts and inferred podcast listening history are intentionally out of scope because
 Spotify does not expose them through the supported Web API. The server does not embed, train on,
 download, or persist Spotify content.

src/spotify_mcp_server/prompts/__init__.py

@@ -0,0 +1,5 @@
+"""Curated Spotify workflows exposed as MCP prompts."""
+
+from spotify_mcp_server.prompts.workflows import register_prompts
+
+__all__ = ["register_prompts"]

src/spotify_mcp_server/prompts/workflows.py

@@ -0,0 +1,87 @@
+"""Prompt text for the four curated Spotify workflows."""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+from mcp.server.mcpserver import MCPServer
+from pydantic import Field
+
+
+def catch_up_on_podcasts(
+    show_filter: Annotated[
+        str | None,
+        Field(description="Optional show name or topic to limit the podcast catch-up."),
+    ] = None,
+) -> str:
+    scope = f" Limit the review to shows matching {show_filter!r}." if show_filter else ""
+    return (
+        "Help me catch up on podcasts using Spotify data only."
+        f"{scope} Use library_read to retrieve saved shows and episodes, then use get_item with "
+        "show_episodes expansion for relevant shows when more episode context is needed. Treat "
+        "missing completion or resume-position data as unknown, not unplayed. Rank a concise "
+        "shortlist using available release dates, durations, descriptions, and progress. Explain "
+        "that Spotify provides neither transcripts nor complete chronological podcast history, "
+        "and do not imply knowledge beyond the returned metadata."
+    )
+
+
+def weekly_music_recap() -> str:
+    return (
+        "Create a concise music recap from Spotify. Call listening_activity for recent tracks and "
+        "short-, medium-, and long-term top tracks and artists. Identify repetitions, contrasts, "
+        "and changes that are directly supported by the returned ordering and timestamps. Do not "
+        "describe the data as a complete week of listening, invent play counts, or include podcast "
+        "history. Clearly distinguish recent plays from Spotify's longer-term affinity lists."
+    )
+
+
+def build_playlist_for_mood(
+    mood: Annotated[
+        str,
+        Field(description="Mood, setting, or activity the new playlist should match."),
+    ],
+) -> str:
+    return (
+        f"Build a Spotify playlist for this mood or activity: {mood!r}. Use search_catalog with "
+        "several focused keyword queries and bounded pagination to assemble candidates. You may "
+        "use listening_activity to align choices with my established taste. Do not claim to use "
+        "Spotify recommendations, audio features, or similarity vectors because those surfaces are "
+        "not available. Choose a coherent ordered track list, then use one playlist_modify plan to "
+        "create a private playlist and add the tracks. Report the created playlist and final track "
+        "list, including any partial-result warnings."
+    )
+
+
+def now_playing_briefing() -> str:
+    return (
+        "Give me a compact now-playing briefing. Call player_status with its default complete "
+        "snapshot, then summarize the active item, playback state, progress, device, and next "
+        "queue items when present. If nothing is playing or a section is unavailable, say so "
+        "directly and preserve any partial-result warning."
+    )
+
+
+def register_prompts(server: MCPServer) -> None:
+    """Register the approved curated workflows on an MCP server."""
+
+    server.prompt(
+        name="catch_up_on_podcasts",
+        title="Catch up on podcasts",
+        description="Prioritize unfinished or unplayed episodes from saved Spotify podcasts.",
+    )(catch_up_on_podcasts)
+    server.prompt(
+        name="weekly_music_recap",
+        title="Weekly music recap",
+        description="Summarize patterns in Spotify's recent plays and top music affinity.",
+    )(weekly_music_recap)
+    server.prompt(
+        name="build_playlist_for_mood",
+        title="Build a playlist for a mood",
+        description="Search Spotify and create a private playlist for a mood or activity.",
+    )(build_playlist_for_mood)
+    server.prompt(
+        name="now_playing_briefing",
+        title="Now-playing briefing",
+        description="Summarize current playback, device, progress, and queue in a compact form.",
+    )(now_playing_briefing)

src/spotify_mcp_server/resources/__init__.py

@@ -0,0 +1,5 @@
+"""Low-computation Spotify context exposed as MCP resources."""
+
+from spotify_mcp_server.resources.profile import register_resources
+
+__all__ = ["register_resources"]

src/spotify_mcp_server/resources/profile.py

@@ -0,0 +1,41 @@
+"""Current-user profile resource."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.server.mcpserver import MCPServer
+
+from spotify_mcp_server.spotify.client import SpotifyClient
+
+
+async def current_user_profile(client: SpotifyClient) -> dict[str, Any]:
+    """Return only the stable current-user context approved for auto-attachment."""
+
+    profile = await client.request("GET", "/me")
+    if not isinstance(profile, dict):
+        raise TypeError("Spotify current-user profile was not a JSON object")
+    account_id = profile.get("account_id")
+    display_name = profile.get("display_name")
+    if not isinstance(account_id, str) or not account_id:
+        raise TypeError("Spotify current-user profile has no valid account_id")
+    if display_name is not None and not isinstance(display_name, str):
+        raise TypeError("Spotify current-user profile has an invalid display_name")
+    return {
+        "display_name": display_name,
+        "account_id": account_id,
+    }
+
+
+def register_resources(server: MCPServer, client: SpotifyClient) -> None:
+    """Register the intentionally small Spotify resource catalog."""
+
+    @server.resource(
+        "spotify://me",
+        name="spotify_current_user",
+        title="Spotify current user",
+        description="Current Spotify display name and stable pseudoanonymous account identifier.",
+        mime_type="application/json",
+    )
+    async def spotify_me() -> dict[str, Any]:
+        return await current_user_profile(client)

src/spotify_mcp_server/server.py

@@ -5,6 +5,8 @@ from __future__ import annotations
 from mcp.server.mcpserver import MCPServer
 from mcp.types import ToolAnnotations
 
+from spotify_mcp_server.prompts import register_prompts
+from spotify_mcp_server.resources import register_resources
 from spotify_mcp_server.spotify.auth import SpotifyTokenProvider
 from spotify_mcp_server.spotify.client import SpotifyClient
 from spotify_mcp_server.spotify.config import Settings
@@ -190,6 +192,9 @@ def create_server(service: SpotifyService | None = None) -> MCPServer:
     async def listening_activity(request: ListeningActivityInput) -> ToolResponse:
         return await spotify.listening_activity(request)
 
+    register_prompts(server)
+    register_resources(server, spotify.client)
+
     return server
 
 

src/spotify_mcp_server/spotify/client.py

@@ -29,6 +29,7 @@ ALLOWED_OPERATIONS: tuple[tuple[str, re.Pattern[str]], ...] = tuple(
     (method, re.compile(pattern))
     for method, pattern in (
         ("GET", r"/search"),
+        ("GET", r"/me"),
         ("GET", r"/(tracks|albums|artists|shows|episodes|audiobooks|chapters)/[^/]+"),
         ("GET", r"/albums/[^/]+/tracks"),
         ("GET", r"/artists/[^/]+/albums"),

tests/test_auth.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import socket
 import threading
 from urllib.error import HTTPError
+from urllib.parse import parse_qs, urlparse
 from urllib.request import urlopen
 
 import httpx2
@@ -16,7 +17,7 @@ from spotify_mcp_server.spotify.auth import (
     _pkce_pair,
     authorize,
 )
-from spotify_mcp_server.spotify.config import Settings
+from spotify_mcp_server.spotify.config import SCOPES, Settings
 
 pytestmark = pytest.mark.anyio
 
@@ -144,8 +145,11 @@ async def test_provider_normalizes_refresh_failure() -> None:
             await provider.access_token()
 
 
-async def test_authorize_persists_refresh_token_only(monkeypatch: pytest.MonkeyPatch) -> None:
+async def test_authorize_requests_all_scopes_and_persists_refresh_token_only(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
     store = MemoryStore()
+    opened_urls: list[str] = []
 
     class FakeClient:
         async def __aenter__(self) -> FakeClient:
@@ -169,7 +173,9 @@ async def test_authorize_persists_refresh_token_only(monkeypatch: pytest.MonkeyP
 
     monkeypatch.setattr(auth_module, "_receive_callback", lambda *_: "callback-code")
     monkeypatch.setattr(auth_module.httpx2, "AsyncClient", lambda **_: FakeClient())
-    await authorize(settings(), open_browser=False, store=store)
+    monkeypatch.setattr(auth_module.webbrowser, "open", opened_urls.append)
+    await authorize(settings(), open_browser=True, store=store)
+    assert parse_qs(urlparse(opened_urls[0]).query)["scope"][0].split() == list(SCOPES)
     assert store.saved == ["persist-me"]
     assert "must-not-persist" not in store.saved
 

tests/test_prompts_resources.py

@@ -0,0 +1,145 @@
+"""Discovery and rendering tests for curated prompts and the profile resource."""
+
+from __future__ import annotations
+
+import json
+
+import httpx2
+import pytest
+from mcp import Client
+
+from spotify_mcp_server.resources.profile import current_user_profile
+from spotify_mcp_server.server import create_server
+from spotify_mcp_server.spotify.client import SpotifyClient
+from spotify_mcp_server.tools.service import SpotifyService
+
+pytestmark = pytest.mark.anyio
+
+PROMPT_NAMES = [
+    "catch_up_on_podcasts",
+    "weekly_music_recap",
+    "build_playlist_for_mood",
+    "now_playing_briefing",
+]
+PROMPT_METADATA = {
+    "catch_up_on_podcasts": (
+        "Catch up on podcasts",
+        "Prioritize unfinished or unplayed episodes from saved Spotify podcasts.",
+    ),
+    "weekly_music_recap": (
+        "Weekly music recap",
+        "Summarize patterns in Spotify's recent plays and top music affinity.",
+    ),
+    "build_playlist_for_mood": (
+        "Build a playlist for a mood",
+        "Search Spotify and create a private playlist for a mood or activity.",
+    ),
+    "now_playing_briefing": (
+        "Now-playing briefing",
+        "Summarize current playback, device, progress, and queue in a compact form.",
+    ),
+}
+
+
+class Tokens:
+    async def access_token(self, *, force_refresh: bool = False) -> str:
+        return "token"
+
+
+async def test_prompt_catalog_and_arguments_are_discoverable() -> None:
+    server = create_server()
+    async with Client(server) as client:
+        prompts = (await client.list_prompts()).prompts
+
+    assert [prompt.name for prompt in prompts] == PROMPT_NAMES
+    assert {
+        prompt.name: (prompt.title, prompt.description) for prompt in prompts
+    } == PROMPT_METADATA
+    arguments = {prompt.name: prompt.arguments or [] for prompt in prompts}
+    assert [argument.name for argument in arguments["catch_up_on_podcasts"]] == ["show_filter"]
+    assert arguments["catch_up_on_podcasts"][0].required is False
+    assert [argument.name for argument in arguments["build_playlist_for_mood"]] == ["mood"]
+    assert arguments["build_playlist_for_mood"][0].required is True
+
+
+async def test_prompts_render_current_tool_workflows_and_boundaries() -> None:
+    server = create_server()
+    async with Client(server) as client:
+        unfiltered_podcast = await client.get_prompt("catch_up_on_podcasts")
+        podcast = await client.get_prompt("catch_up_on_podcasts", {"show_filter": "design"})
+        playlist = await client.get_prompt("build_playlist_for_mood", {"mood": "late-night focus"})
+        recap = await client.get_prompt("weekly_music_recap")
+        now_playing = await client.get_prompt("now_playing_briefing")
+
+    unfiltered_podcast_text = unfiltered_podcast.messages[0].content.text
+    podcast_text = podcast.messages[0].content.text
+    playlist_text = playlist.messages[0].content.text
+    recap_text = recap.messages[0].content.text
+    now_playing_text = now_playing.messages[0].content.text
+    assert "library_read" in podcast_text
+    assert "get_item" in podcast_text
+    assert "design" in podcast_text
+    assert "transcripts" in podcast_text
+    assert "Limit the review to shows matching" not in unfiltered_podcast_text
+    assert "search_catalog" in playlist_text
+    assert "playlist_modify" in playlist_text
+    assert "late-night focus" in playlist_text
+    assert "audio features" in playlist_text
+    assert "listening_activity" in recap_text
+    assert "complete week" in recap_text
+    assert "player_status" in now_playing_text
+    assert "partial-result warning" in now_playing_text
+
+
+async def test_spotify_me_resource_reads_and_filters_current_profile() -> None:
+    async def handler(request: httpx2.Request) -> httpx2.Response:
+        assert request.method == "GET"
+        assert request.url.path == "/v1/me"
+        assert request.headers["Authorization"] == "Bearer token"
+        return httpx2.Response(
+            200,
+            json={
+                "display_name": "Leo",
+                "account_id": "account-123",
+                "id": "legacy-user-id",
+                "email": "not-exposed@example.com",
+            },
+        )
+
+    async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http:
+        spotify = SpotifyClient(Tokens(), client=http)
+        server = create_server(SpotifyService(spotify))
+        async with Client(server) as client:
+            resources = (await client.list_resources()).resources
+            result = await client.read_resource("spotify://me")
+
+    assert [str(resource.uri) for resource in resources] == ["spotify://me"]
+    assert resources[0].mime_type == "application/json"
+    content = result.contents[0]
+    assert content.mime_type == "application/json"
+    assert json.loads(content.text) == {
+        "display_name": "Leo",
+        "account_id": "account-123",
+    }
+
+
+@pytest.mark.parametrize(
+    ("profile", "message"),
+    [
+        ({}, "account_id"),
+        ({"account_id": None}, "account_id"),
+        ({"account_id": ""}, "account_id"),
+        ({"account_id": 123}, "account_id"),
+        ({"account_id": "account-123", "display_name": 123}, "display_name"),
+    ],
+)
+async def test_spotify_me_resource_rejects_invalid_profile_fields(
+    profile: dict[str, object], message: str
+) -> None:
+    async def handler(_: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(200, json=profile)
+
+    async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http:
+        spotify = SpotifyClient(Tokens(), client=http)
+        with pytest.raises(TypeError, match=message):
+            await current_user_profile(spotify)