File size: 10,529 Bytes
93a887a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
"""Pydantic v2 input contracts for all nine public MCP tools."""

from __future__ import annotations

from typing import Literal

from pydantic import Field, model_validator

from spotify_mcp_server.tools.common import (
    FailureMode,
    OffsetPaging,
    PriorActionReference,
    SpotifyReferenceInput,
    SpotifyType,
    StrictModel,
    one_of,
)


class SearchCatalogInput(StrictModel):
    query: str = Field(min_length=1, description="Spotify search query, including field filters.")
    types: list[Literal["album", "artist", "playlist", "track", "show", "episode", "audiobook"]] = (
        Field(min_length=1, description="Distinct Spotify catalog types to search in parallel.")
    )
    market: str | None = Field(default=None, pattern=r"^[A-Z]{2}$")
    limit: int = Field(default=5, ge=1, le=10)
    offset: int = Field(default=0, ge=0, le=1000)
    include_external: Literal["audio"] | None = None
    max_pages_per_type: int = Field(default=1, ge=1, le=10)

    @model_validator(mode="after")
    def unique_types(self) -> SearchCatalogInput:
        if len(self.types) != len(set(self.types)):
            raise ValueError("types must not contain duplicates")
        return self


class GetItemInput(StrictModel):
    items: list[SpotifyReferenceInput] = Field(min_length=1, max_length=25)
    market: str | None = Field(default=None, pattern=r"^[A-Z]{2}$")
    expansions: list[
        Literal["album_tracks", "artist_albums", "show_episodes", "audiobook_chapters"]
    ] = Field(default_factory=list)
    expansion_paging: OffsetPaging = Field(default_factory=OffsetPaging)

    @model_validator(mode="after")
    def unique_expansions(self) -> GetItemInput:
        if len(self.expansions) != len(set(self.expansions)):
            raise ValueError("expansions must not contain duplicates")
        return self


class PlayerStatusInput(StrictModel):
    include: list[Literal["current_playback", "devices", "queue"]] = Field(
        default=["current_playback", "devices", "queue"], min_length=1
    )
    market: str | None = Field(default=None, pattern=r"^[A-Z]{2}$")

    @model_validator(mode="after")
    def unique_sections(self) -> PlayerStatusInput:
        if len(self.include) != len(set(self.include)):
            raise ValueError("include must not contain duplicates")
        return self


class PlayerAction(StrictModel):
    action: Literal[
        "transfer_playback",
        "start_playback",
        "pause_playback",
        "skip_next",
        "skip_previous",
        "seek",
        "set_repeat",
        "set_volume",
        "set_shuffle",
        "add_to_queue",
    ]
    device_id: str | None = None
    device_ids: list[str] | None = Field(default=None, min_length=1, max_length=1)
    play: bool | None = None
    context_uri: str | None = None
    uris: list[str] | None = Field(default=None, min_length=1, max_length=100)
    offset_uri: str | None = None
    offset_position: int | None = Field(default=None, ge=0)
    position_ms: int | None = Field(default=None, ge=0)
    state: Literal["track", "context", "off"] | None = None
    volume_percent: int | None = Field(default=None, ge=0, le=100)
    enabled: bool | None = None
    uri: str | None = None

    @model_validator(mode="after")
    def validate_action_fields(self) -> PlayerAction:
        required: dict[str, tuple[object, str]] = {
            "transfer_playback": (self.device_ids, "device_ids"),
            "seek": (self.position_ms, "position_ms"),
            "set_repeat": (self.state, "state"),
            "set_volume": (self.volume_percent, "volume_percent"),
            "set_shuffle": (self.enabled, "enabled"),
            "add_to_queue": (self.uri, "uri"),
        }
        if self.action in required and required[self.action][0] is None:
            raise ValueError(f"{required[self.action][1]} is required for {self.action}")
        if self.action == "start_playback":
            if self.context_uri is not None and self.uris is not None:
                raise ValueError("start_playback accepts context_uri or uris, not both")
            if self.offset_uri is not None and self.offset_position is not None:
                raise ValueError("start_playback accepts one offset form")
        return self


class PlayerControlInput(StrictModel):
    actions: list[PlayerAction] = Field(min_length=1, max_length=20)
    failure_mode: FailureMode = FailureMode.STOP
    include_final_status: bool = True


class PlaylistReadRequest(StrictModel):
    operation: Literal["list_current_playlists", "get_playlists"]
    playlists: list[SpotifyReferenceInput] = Field(default_factory=list, max_length=20)
    include_items: bool = False
    paging: OffsetPaging = Field(default_factory=OffsetPaging)

    @model_validator(mode="after")
    def playlists_for_get(self) -> PlaylistReadRequest:
        if self.operation == "get_playlists" and not self.playlists:
            raise ValueError("playlists is required for get_playlists")
        if self.operation == "list_current_playlists" and self.playlists:
            raise ValueError("playlists is not used by list_current_playlists")
        return self


class PlaylistReadInput(StrictModel):
    requests: list[PlaylistReadRequest] = Field(min_length=1, max_length=20)
    market: str | None = Field(default=None, pattern=r"^[A-Z]{2}$")


class PlaylistAction(StrictModel):
    action: Literal[
        "create_playlist",
        "change_playlist_details",
        "add_items",
        "remove_items",
        "replace_items",
        "reorder_items",
    ]
    playlist: SpotifyReferenceInput | PriorActionReference | None = None
    name: str | None = Field(default=None, min_length=1)
    public: bool | None = None
    collaborative: bool | None = None
    description: str | None = None
    uris: list[str] | None = Field(default=None, max_length=100)
    position: int | None = Field(default=None, ge=0)
    snapshot_id: str | None = None
    range_start: int | None = Field(default=None, ge=0)
    insert_before: int | None = Field(default=None, ge=0)
    range_length: int | None = Field(default=None, ge=1)

    @model_validator(mode="after")
    def validate_playlist_action(self) -> PlaylistAction:
        if self.action == "create_playlist":
            if self.name is None:
                raise ValueError("name is required for create_playlist")
        elif self.playlist is None:
            raise ValueError("playlist is required for this action")
        if self.action == "change_playlist_details" and all(
            value is None
            for value in (self.name, self.public, self.collaborative, self.description)
        ):
            raise ValueError("at least one detail is required for change_playlist_details")
        if self.action in {"add_items", "remove_items", "replace_items"} and self.uris is None:
            raise ValueError("uris is required for item mutations")
        if self.action in {"add_items", "remove_items"} and not self.uris:
            raise ValueError("uris must not be empty for add_items or remove_items")
        if self.action == "reorder_items" and one_of(self.range_start, self.insert_before) != 2:
            raise ValueError("range_start and insert_before are required for reorder_items")
        return self


class PlaylistModifyInput(StrictModel):
    actions: list[PlaylistAction] = Field(min_length=1, max_length=20)
    failure_mode: FailureMode = FailureMode.STOP


class LibraryReadRequest(StrictModel):
    operation: Literal["list_saved", "list_followed_artists", "contains"]
    type: Literal["track", "album", "show", "episode", "audiobook"] | None = None
    items: list[SpotifyReferenceInput] = Field(default_factory=list, max_length=40)
    paging: OffsetPaging = Field(default_factory=OffsetPaging)
    after: str | None = None

    @model_validator(mode="after")
    def validate_library_read(self) -> LibraryReadRequest:
        if self.operation == "list_saved" and self.type is None:
            raise ValueError("type is required for list_saved")
        if self.operation == "contains" and not self.items:
            raise ValueError("items is required for contains")
        return self


class LibraryReadInput(StrictModel):
    requests: list[LibraryReadRequest] = Field(min_length=1, max_length=10)
    market: str | None = Field(default=None, pattern=r"^[A-Z]{2}$")


class LibraryAction(StrictModel):
    action: Literal["save", "remove", "follow", "unfollow"]
    items: list[SpotifyReferenceInput] = Field(min_length=1, max_length=40)

    @model_validator(mode="after")
    def validate_item_types(self) -> LibraryAction:
        types = {item.normalized().type for item in self.items}
        save_types = {
            SpotifyType.TRACK,
            SpotifyType.ALBUM,
            SpotifyType.EPISODE,
            SpotifyType.SHOW,
            SpotifyType.AUDIOBOOK,
        }
        follow_types = {SpotifyType.ARTIST, SpotifyType.PLAYLIST}
        allowed = save_types if self.action in {"save", "remove"} else follow_types
        if not types <= allowed:
            raise ValueError(f"{self.action} does not support one or more supplied item types")
        return self


class LibraryModifyInput(StrictModel):
    actions: list[LibraryAction] = Field(min_length=1, max_length=20)
    failure_mode: FailureMode = FailureMode.STOP


class ListeningActivityInput(StrictModel):
    include: list[Literal["recent_tracks", "top_tracks", "top_artists"]] = Field(
        default=["recent_tracks", "top_tracks", "top_artists"],
        min_length=1,
    )
    time_ranges: list[Literal["short_term", "medium_term", "long_term"]] = Field(
        default=["short_term", "medium_term", "long_term"], min_length=1
    )
    recent_limit: int = Field(default=20, ge=1, le=50)
    top_limit: int = Field(default=20, ge=1, le=50)
    before: int | None = Field(default=None, ge=0)
    after: int | None = Field(default=None, ge=0)
    top_offset: int = Field(default=0, ge=0)
    max_pages: int = Field(default=1, ge=1, le=10)

    @model_validator(mode="after")
    def cursor_choice(self) -> ListeningActivityInput:
        if self.before is not None and self.after is not None:
            raise ValueError("before and after are mutually exclusive")
        if len(self.include) != len(set(self.include)):
            raise ValueError("include must not contain duplicates")
        if len(self.time_ranges) != len(set(self.time_ranges)):
            raise ValueError("time_ranges must not contain duplicates")
        return self