TIDAL MCP Server

Instructions

Implementation Reference

• src/tidal_mcp/server.py:233-270 (handler)

  Handler function implementing the search_tracks tool: performs authentication check, calls TIDAL search API, constructs TrackList response.

  @mcp.tool()
  async def search_tracks(query: str, limit: int = 10) -> TrackList:
      """
      Search for tracks on TIDAL.
  
      Args:
          query: Search query - artist name, song title, or combination
          limit: Maximum results (1-50, default: 10)
  
      Returns:
          List of matching tracks with id, title, artist, album, duration, and URL
      """
      if not await ensure_authenticated():
          raise ToolError("Not authenticated. Please run the 'login' tool first.")
  
      try:
          limit = min(max(1, limit), 50)
          results = await anyio.to_thread.run_sync(
              lambda: session.search(query, models=[tidalapi.Track], limit=limit)
          )
  
          tracks = []
          for track in results.get("tracks", []):
              tracks.append(
                  Track(
                      id=str(track.id),
                      title=track.name,
                      artist=track.artist.name if track.artist else "Unknown Artist",
                      album=track.album.name if track.album else "Unknown Album",
                      duration_seconds=track.duration,
                      url=f"https://tidal.com/browse/track/{track.id}",
                  )
              )
  
          return TrackList(status="success", query=query, count=len(tracks), tracks=tracks)
      except Exception as e:
          raise ToolError(f"Track search failed: {str(e)}")

• src/tidal_mcp/models.py:62-69 (schema)

  Pydantic output schema for search_tracks response containing list of tracks.

  class TrackList(BaseModel):
      """List of tracks with metadata."""
  
      status: str = Field(description="Operation status (success/error)")
      query: Optional[str] = Field(None, description="Search query used (for search results)")
      count: int = Field(description="Number of tracks returned")
      tracks: List[Track] = Field(description="List of track objects")

• src/tidal_mcp/models.py:16-25 (schema)

  Pydantic model defining structure of individual Track objects returned by search_tracks.

  class Track(BaseModel):
      """Structured representation of a TIDAL track."""
  
      id: str = Field(description="Unique TIDAL track ID")
      title: str = Field(description="Track title")
      artist: str = Field(description="Primary artist name")
      album: str = Field(description="Album name")
      duration_seconds: int = Field(description="Track duration in seconds")
      url: str = Field(description="TIDAL web URL for the track")

• src/tidal_mcp/server.py:233-233 (registration)

  FastMCP decorator registering the search_tracks function as a tool.

  @mcp.tool()

• src/tidal_mcp/server.py:120-151 (helper)

  Helper function to ensure user authentication before executing search_tracks.

  async def ensure_authenticated() -> bool:
      """
      Check if user is authenticated with TIDAL.
      Automatically loads persisted session if available.
      """
      if await anyio.Path(SESSION_FILE).exists():
          try:
              async with await anyio.open_file(SESSION_FILE, "r") as f:
                  content = await f.read()
                  data = json.loads(content)
  
              # Load OAuth session
              result = await anyio.to_thread.run_sync(
                  session.load_oauth_session,
                  data["token_type"]["data"],
                  data["access_token"]["data"],
                  data["refresh_token"]["data"],
                  None,  # expiry time
              )
  
              if result:
                  is_valid = await anyio.to_thread.run_sync(session.check_login)
                  if not is_valid:
                      await anyio.Path(SESSION_FILE).unlink()
                  return is_valid
              return False
          except Exception:
              await anyio.Path(SESSION_FILE).unlink()
              return False
  
      return await anyio.to_thread.run_sync(session.check_login)