diff --git a/beets/autotag/match.py b/beets/autotag/match.py
index 91a315de06..24191c41a9 100644
--- a/beets/autotag/match.py
+++ b/beets/autotag/match.py
@@ -27,7 +27,7 @@
 import lap
 import numpy as np
 
-from beets import config, logging, plugins
+from beets import config, logging, metadata_plugins
 from beets.autotag import (
     AlbumInfo,
     AlbumMatch,
@@ -213,7 +213,7 @@ def track_distance(
         dist.add_expr("medium", item.disc != track_info.medium)
 
     # Plugins.
-    dist.update(plugins.track_distance(item, track_info))
+    dist.update(metadata_plugins.track_distance(item, track_info))
 
     return dist
 
@@ -330,7 +330,7 @@ def distance(
         dist.add("unmatched_tracks", 1.0)
 
     # Plugins.
-    dist.update(plugins.album_distance(items, album_info, mapping))
+    dist.update(metadata_plugins.album_distance(items, album_info, mapping))
 
     return dist
 
@@ -356,7 +356,7 @@ def match_by_id(items: Iterable[Item]) -> AlbumInfo | None:
             return None
     # If all album IDs are equal, look up the album.
     log.debug("Searching for discovered album ID: {0}", first)
-    return plugins.album_for_id(first)
+    return metadata_plugins.album_for_id(first)
 
 
 def _recommendation(
@@ -511,7 +511,7 @@ def tag_album(
     if search_ids:
         for search_id in search_ids:
             log.debug("Searching for album ID: {0}", search_id)
-            if info := plugins.album_for_id(search_id):
+            if info := metadata_plugins.album_for_id(search_id):
                 _add_candidate(items, candidates, info)
 
     # Use existing metadata or text search.
@@ -548,7 +548,7 @@ def tag_album(
         log.debug("Album might be VA: {0}", va_likely)
 
         # Get the results from the data sources.
-        for matched_candidate in plugins.candidates(
+        for matched_candidate in metadata_plugins.candidates(
             items, search_artist, search_album, va_likely
         ):
             _add_candidate(items, candidates, matched_candidate)
@@ -583,7 +583,7 @@ def tag_item(
     if trackids:
         for trackid in trackids:
             log.debug("Searching for track ID: {0}", trackid)
-            if info := plugins.track_for_id(trackid):
+            if info := metadata_plugins.track_for_id(trackid):
                 dist = track_distance(item, info, incl_artist=True)
                 candidates[info.track_id] = hooks.TrackMatch(dist, info)
                 # If this is a good match, then don't keep searching.
@@ -609,7 +609,7 @@ def tag_item(
     log.debug("Item search terms: {0} - {1}", search_artist, search_title)
 
     # Get and evaluate candidate metadata.
-    for track_info in plugins.item_candidates(
+    for track_info in metadata_plugins.item_candidates(
         item, search_artist, search_title
     ):
         dist = track_distance(item, track_info, incl_artist=True)
diff --git a/beets/metadata_plugins.py b/beets/metadata_plugins.py
new file mode 100644
index 0000000000..d669ae6c16
--- /dev/null
+++ b/beets/metadata_plugins.py
@@ -0,0 +1,348 @@
+"""Metadata source plugin interface.
+
+This allows beets to lookup metadata from various sources. We define
+a common interface for all metadata sources which need to be
+implemented as plugins.
+"""
+
+from __future__ import annotations
+
+import abc
+import re
+from typing import TYPE_CHECKING, Generic, Literal, Sequence, TypedDict, TypeVar
+
+from typing_extensions import NotRequired
+
+from .plugins import BeetsPlugin, find_plugins, notify_info_yielded, send
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
+    from confuse import ConfigView
+
+    from .autotag import Distance
+    from .autotag.hooks import AlbumInfo, Item, TrackInfo
+
+
+def find_metadata_source_plugins() -> list[MetadataSourcePlugin]:
+    """Returns a list of MetadataSourcePlugin subclass instances
+
+    Resolved from all currently loaded beets plugins.
+    """
+    return [
+        plugin
+        for plugin in find_plugins()
+        if isinstance(plugin, MetadataSourcePlugin)
+    ]
+
+
+@notify_info_yielded("albuminfo_received")
+def candidates(*args, **kwargs) -> Iterable[AlbumInfo]:
+    """Return matching album candidates from all metadata source plugins."""
+    for plugin in find_metadata_source_plugins():
+        yield from plugin.candidates(*args, **kwargs)
+
+
+@notify_info_yielded("trackinfo_received")
+def item_candidates(*args, **kwargs) -> Iterable[TrackInfo]:
+    """Return matching track candidates fromm all metadata source plugins."""
+    for plugin in find_metadata_source_plugins():
+        yield from plugin.item_candidates(*args, **kwargs)
+
+
+def album_for_id(_id: str) -> AlbumInfo | None:
+    """Get AlbumInfo object for the given ID string.
+
+    A single ID can yield just a single album, so we return the first match.
+    """
+    for plugin in find_metadata_source_plugins():
+        if info := plugin.album_for_id(album_id=_id):
+            send("albuminfo_received", info=info)
+            return info
+
+    return None
+
+
+def track_for_id(_id: str) -> TrackInfo | None:
+    """Get TrackInfo object for the given ID string.
+
+    A single ID can yield just a single track, so we return the first match.
+    """
+    for plugin in find_metadata_source_plugins():
+        if info := plugin.track_for_id(_id):
+            send("trackinfo_received", info=info)
+            return info
+
+    return None
+
+
+def track_distance(item: Item, info: TrackInfo) -> Distance:
+    """Returns the track distance for an item and trackinfo.
+
+    Returns a Distance object is populated by all metadata source plugins
+    that implement the :py:meth:`MetadataSourcePlugin.track_distance` method.
+    """
+    from beets.autotag.hooks import Distance
+
+    dist = Distance()
+    for plugin in find_metadata_source_plugins():
+        dist.update(plugin.track_distance(item, info))
+    return dist
+
+
+def album_distance(
+    items: Sequence[Item],
+    album_info: AlbumInfo,
+    mapping: dict[Item, TrackInfo],
+) -> Distance:
+    """Returns the album distance calculated by plugins."""
+    from beets.autotag.hooks import Distance
+
+    dist = Distance()
+    for plugin in find_metadata_source_plugins():
+        dist.update(plugin.album_distance(items, album_info, mapping))
+    return dist
+
+
+def _get_distance(
+    config: ConfigView, data_source: str, info: AlbumInfo | TrackInfo
+) -> Distance:
+    """Returns the ``data_source`` weight and the maximum source weight
+    for albums or individual tracks.
+    """
+    from beets.autotag.hooks import Distance
+
+    dist = Distance()
+    if info.data_source == data_source:
+        dist.add("source", config["source_weight"].as_number())
+    return dist
+
+
+class MetadataSourcePlugin(BeetsPlugin, metaclass=abc.ABCMeta):
+    """A plugin that provides metadata from a specific source.
+
+    This base class implements a contract for plugins that provide metadata
+    from a specific source. The plugin must implement the methods to search for albums
+    and tracks, and to retrieve album and track information by ID.
+    """
+
+    data_source: str
+
+    def __init__(self, data_source: str, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self.data_source = data_source or self.__class__.__name__
+        self.config.add({"source_weight": 0.5})
+
+    @abc.abstractmethod
+    def album_for_id(self, album_id: str) -> AlbumInfo | None:
+        """Return :py:class:`AlbumInfo` object or None if no matching release was
+        found."""
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def track_for_id(self, track_id: str) -> TrackInfo | None:
+        """Return a :py:class:`AlbumInfo` object or None if no matching release was
+        found.
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def candidates(
+        self,
+        items: Sequence[Item],
+        artist: str,
+        album: str,
+        va_likely: bool,
+    ) -> Iterable[AlbumInfo]:
+        """Return :py:class:`AlbumInfo` candidates that match the given album.
+
+        Used in the autotag functionality to search for albums.
+
+        :param items: List of items in the album
+        :param artist: Album artist
+        :param album: Album name
+        :param va_likely: Whether the album is likely to be by various artists
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def item_candidates(
+        self, item: Item, artist: str, title: str
+    ) -> Iterable[TrackInfo]:
+        """Return :py:class:`TrackInfo` candidates that match the given track.
+
+        Used in the autotag functionality to search for tracks.
+
+        :param item: Track item
+        :param artist: Track artist
+        :param title: Track title
+        """
+        raise NotImplementedError
+
+    def albums_for_ids(self, ids: Sequence[str]) -> Iterable[AlbumInfo | None]:
+        """Batch lookup of album metadata for a list of album IDs.
+
+        Given a list of album identifiers, yields corresponding AlbumInfo
+        objects. Missing albums result in None values in the output iterator.
+        Plugins may implement this for optimized batched lookups instead of
+        single calls to album_for_id.
+        """
+
+        return (self.album_for_id(id) for id in ids)
+
+    def tracks_for_ids(self, ids: Sequence[str]) -> Iterable[TrackInfo | None]:
+        """Batch lookup of track metadata for a list of track IDs.
+
+        Given a list of track identifiers, yields corresponding TrackInfo objects.
+        Missing tracks result in None values in the output iterator. Plugins may
+        implement this for optimized batched lookups instead of single calls to
+        track_for_id.
+        """
+
+        return (self.track_for_id(id) for id in ids)
+
+    def album_distance(
+        self,
+        items: Sequence[Item],
+        album_info: AlbumInfo,
+        mapping: dict[Item, TrackInfo],
+    ) -> Distance:
+        return _get_distance(
+            data_source=self.data_source, info=album_info, config=self.config
+        )
+
+    def track_distance(
+        self,
+        item: Item,
+        info: TrackInfo,
+    ) -> Distance:
+        return _get_distance(
+            data_source=self.data_source, info=info, config=self.config
+        )
+
+
+class IDResponse(TypedDict):
+    """Response from the API containing an ID."""
+
+    id: str
+
+
+class SearchFilter(TypedDict):
+    artist: NotRequired[str]
+    album: NotRequired[str]
+
+
+R = TypeVar("R", bound=IDResponse)
+
+
+class SearchApiMetadataSourcePlugin(
+    Generic[R], MetadataSourcePlugin, metaclass=abc.ABCMeta
+):
+    """Helper class to implement a metadata source plugin with an API.
+
+    Plugins using this ABC must implement an API search method to
+    retrieve album and track information by ID,
+    i.e. `album_for_id` and `track_for_id`, and a search method to
+    perform a search on the API. The search method should return a list
+    of identifiers for the requested type (album or track).
+    """
+
+    @abc.abstractmethod
+    def _search_api(
+        self,
+        query_type: Literal["album", "track"],
+        filters: SearchFilter | None = None,
+        keywords: str = "",
+    ) -> Sequence[R] | None:
+        """Perform a search on the API.
+
+        :param query_type: The type of query to perform.
+        :param filters: A dictionary of filters to apply to the search.
+        :param keywords: Additional keywords to include in the search.
+
+        Should return a list of identifiers for the requested type (album or track).
+        """
+        raise NotImplementedError
+
+    def candidates(
+        self,
+        items: Sequence[Item],
+        artist: str,
+        album: str,
+        va_likely: bool,
+    ) -> Iterable[AlbumInfo]:
+        query_filters: SearchFilter = {"album": album}
+        if not va_likely:
+            query_filters["artist"] = artist
+
+        results = self._search_api("album", query_filters)
+        if not results:
+            return []
+
+        return filter(
+            None, self.albums_for_ids([result["id"] for result in results])
+        )
+
+    def item_candidates(
+        self, item: Item, artist: str, title: str
+    ) -> Iterable[TrackInfo]:
+        results = self._search_api("track", {"artist": artist}, keywords=title)
+        if not results:
+            return []
+
+        return filter(
+            None,
+            self.tracks_for_ids([result["id"] for result in results if result]),
+        )
+
+
+def artists_to_artist_str(
+    artists: Iterable[dict],
+    id_key: str | int = "id",
+    name_key: str | int = "name",
+    join_key: str | int | None = None,
+) -> tuple[str, str | None]:
+    """Returns an artist string (all artists) and an artist_id (the main
+    artist) for a list of artist object dicts.
+
+    For each artist, this function moves articles (such as 'a', 'an',
+    and 'the') to the front and strips trailing disambiguation numbers. It
+    returns a tuple containing the comma-separated string of all
+    normalized artists and the ``id`` of the main/first artist.
+    Alternatively a keyword can be used to combine artists together into a
+    single string by passing the join_key argument.
+
+    :param artists: Iterable of artist dicts or lists returned by API.
+    :type artists: list[dict] or list[list]
+    :param id_key: Key or index corresponding to the value of ``id`` for
+        the main/first artist. Defaults to 'id'.
+    :param name_key: Key or index corresponding to values of names
+        to concatenate for the artist string (containing all artists).
+        Defaults to 'name'.
+    :param join_key: Key or index corresponding to a field containing a
+        keyword to use for combining artists into a single string, for
+        example "Feat.", "Vs.", "And" or similar. The default is None
+        which keeps the default behaviour (comma-separated).
+    :return: Normalized artist string.
+    """
+    artist_id = None
+    artist_string = ""
+    artists = list(artists)  # In case a generator was passed.
+    total = len(artists)
+    for idx, artist in enumerate(artists):
+        if not artist_id:
+            artist_id = artist[id_key]
+        name = artist[name_key]
+        # Strip disambiguation number.
+        name = re.sub(r" \(\d+\)$", "", name)
+        # Move articles to the front.
+        name = re.sub(r"^(.*?), (a|an|the)$", r"\2 \1", name, flags=re.I)
+        # Use a join keyword if requested and available.
+        if idx < (total - 1):  # Skip joining on last.
+            if join_key and artist.get(join_key, None):
+                name += f" {artist[join_key]} "
+            else:
+                name += ", "
+        artist_string += name
+
+    return artist_string, artist_id
diff --git a/beets/plugins.py b/beets/plugins.py
index 63e5d3bde5..1ef3e0616a 100644
--- a/beets/plugins.py
+++ b/beets/plugins.py
@@ -23,21 +23,12 @@
 import traceback
 from collections import defaultdict
 from functools import wraps
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    Callable,
-    Generic,
-    Sequence,
-    TypedDict,
-    TypeVar,
-)
+from typing import TYPE_CHECKING, Any, Callable, Sequence, TypeVar
 
 import mediafile
 
 import beets
 from beets import logging
-from beets.util.id_extractors import extract_release_id
 
 if sys.version_info >= (3, 10):
     from typing import ParamSpec
@@ -50,7 +41,6 @@
 
     from confuse import ConfigView
 
-    from beets.autotag import AlbumInfo, Distance, TrackInfo
     from beets.dbcore import Query
     from beets.dbcore.db import FieldQueryType
     from beets.dbcore.types import Type
@@ -110,7 +100,7 @@ def filter(self, record):
 # Managing the plugins themselves.
 
 
-class BeetsPlugin:
+class BeetsPlugin(metaclass=abc.ABCMeta):
     """The base class for all beets plugins. Plugins provide
     functionality by defining a subclass of BeetsPlugin and overriding
     the abstract methods defined here.
@@ -213,66 +203,6 @@ def queries(self) -> dict[str, type[Query]]:
         """Return a dict mapping prefixes to Query subclasses."""
         return {}
 
-    def track_distance(
-        self,
-        item: Item,
-        info: TrackInfo,
-    ) -> Distance:
-        """Should return a Distance object to be added to the
-        distance for every track comparison.
-        """
-        from beets.autotag.hooks import Distance
-
-        return Distance()
-
-    def album_distance(
-        self,
-        items: Sequence[Item],
-        album_info: AlbumInfo,
-        mapping: dict[Item, TrackInfo],
-    ) -> Distance:
-        """Should return a Distance object to be added to the
-        distance for every album-level comparison.
-        """
-        from beets.autotag.hooks import Distance
-
-        return Distance()
-
-    def candidates(
-        self, items: list[Item], artist: str, album: str, va_likely: bool
-    ) -> Iterable[AlbumInfo]:
-        """Return :py:class:`AlbumInfo` candidates that match the given album.
-
-        :param items: List of items in the album
-        :param artist: Album artist
-        :param album: Album name
-        :param va_likely: Whether the album is likely to be by various artists
-        """
-        yield from ()
-
-    def item_candidates(
-        self, item: Item, artist: str, title: str
-    ) -> Iterable[TrackInfo]:
-        """Return :py:class:`TrackInfo` candidates that match the given track.
-
-        :param item: Track item
-        :param artist: Track artist
-        :param title: Track title
-        """
-        yield from ()
-
-    def album_for_id(self, album_id: str) -> AlbumInfo | None:
-        """Return an AlbumInfo object or None if no matching release was
-        found.
-        """
-        return None
-
-    def track_for_id(self, track_id: str) -> TrackInfo | None:
-        """Return a TrackInfo object or None if no matching release was
-        found.
-        """
-        return None
-
     def add_media_field(
         self, name: str, descriptor: mediafile.MediaField
     ) -> None:
@@ -367,7 +297,7 @@ def load_plugins(names: Sequence[str] = ()) -> None:
                         isinstance(obj, type)
                         and issubclass(obj, BeetsPlugin)
                         and obj != BeetsPlugin
-                        and obj != MetadataSourcePlugin
+                        and not inspect.isabstract(obj)
                         and obj not in _classes
                     ):
                         _classes.add(obj)
@@ -451,32 +381,6 @@ def named_queries(model_cls: type[AnyModel]) -> dict[str, FieldQueryType]:
     return queries
 
 
-def track_distance(item: Item, info: TrackInfo) -> Distance:
-    """Gets the track distance calculated by all loaded plugins.
-    Returns a Distance object.
-    """
-    from beets.autotag.hooks import Distance
-
-    dist = Distance()
-    for plugin in find_plugins():
-        dist.update(plugin.track_distance(item, info))
-    return dist
-
-
-def album_distance(
-    items: Sequence[Item],
-    album_info: AlbumInfo,
-    mapping: dict[Item, TrackInfo],
-) -> Distance:
-    """Returns the album distance calculated by plugins."""
-    from beets.autotag.hooks import Distance
-
-    dist = Distance()
-    for plugin in find_plugins():
-        dist.update(plugin.album_distance(items, album_info, mapping))
-    return dist
-
-
 def notify_info_yielded(event: str) -> Callable[[IterF[P, Ret]], IterF[P, Ret]]:
     """Makes a generator send the event 'event' every time it yields.
     This decorator is supposed to decorate a generator, but any function
@@ -497,46 +401,6 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> Iterable[Ret]:
     return decorator
 
 
-@notify_info_yielded("albuminfo_received")
-def candidates(*args, **kwargs) -> Iterable[AlbumInfo]:
-    """Return matching album candidates from all plugins."""
-    for plugin in find_plugins():
-        yield from plugin.candidates(*args, **kwargs)
-
-
-@notify_info_yielded("trackinfo_received")
-def item_candidates(*args, **kwargs) -> Iterable[TrackInfo]:
-    """Return matching track candidates from all plugins."""
-    for plugin in find_plugins():
-        yield from plugin.item_candidates(*args, **kwargs)
-
-
-def album_for_id(_id: str) -> AlbumInfo | None:
-    """Get AlbumInfo object for the given ID string.
-
-    A single ID can yield just a single album, so we return the first match.
-    """
-    for plugin in find_plugins():
-        if info := plugin.album_for_id(_id):
-            send("albuminfo_received", info=info)
-            return info
-
-    return None
-
-
-def track_for_id(_id: str) -> TrackInfo | None:
-    """Get TrackInfo object for the given ID string.
-
-    A single ID can yield just a single track, so we return the first match.
-    """
-    for plugin in find_plugins():
-        if info := plugin.track_for_id(_id):
-            send("trackinfo_received", info=info)
-            return info
-
-    return None
-
-
 def template_funcs() -> TFuncMap[str]:
     """Get all the template functions declared by plugins as a
     dictionary.
@@ -711,20 +575,6 @@ def sanitize_pairs(
     return res
 
 
-def get_distance(
-    config: ConfigView, data_source: str, info: AlbumInfo | TrackInfo
-) -> Distance:
-    """Returns the ``data_source`` weight and the maximum source weight
-    for albums or individual tracks.
-    """
-    from beets.autotag.hooks import Distance
-
-    dist = Distance()
-    if info.data_source == data_source:
-        dist.add("source", config["source_weight"].as_number())
-    return dist
-
-
 def apply_item_changes(
     lib: Library, item: Item, move: bool, pretend: bool, write: bool
 ) -> None:
@@ -750,149 +600,3 @@ def apply_item_changes(
         item.try_write()
 
     item.store()
-
-
-class Response(TypedDict):
-    """A dictionary with the response of a plugin API call.
-
-    May be extended by plugins to include additional information, but `id`
-    is required.
-    """
-
-    id: str
-
-
-R = TypeVar("R", bound=Response)
-
-
-class MetadataSourcePlugin(Generic[R], BeetsPlugin, metaclass=abc.ABCMeta):
-    def __init__(self):
-        super().__init__()
-        self.config.add({"source_weight": 0.5})
-
-    @property
-    @abc.abstractmethod
-    def data_source(self) -> str:
-        raise NotImplementedError
-
-    @property
-    @abc.abstractmethod
-    def search_url(self) -> str:
-        raise NotImplementedError
-
-    @property
-    @abc.abstractmethod
-    def album_url(self) -> str:
-        raise NotImplementedError
-
-    @property
-    @abc.abstractmethod
-    def track_url(self) -> str:
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def _search_api(
-        self,
-        query_type: str,
-        filters: dict[str, str] | None,
-        keywords: str = "",
-    ) -> Sequence[R]:
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def album_for_id(self, album_id: str) -> AlbumInfo | None:
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def track_for_id(self, track_id: str) -> TrackInfo | None:
-        raise NotImplementedError
-
-    @staticmethod
-    def get_artist(
-        artists,
-        id_key: str | int = "id",
-        name_key: str | int = "name",
-        join_key: str | int | None = None,
-    ) -> tuple[str, str | None]:
-        """Returns an artist string (all artists) and an artist_id (the main
-        artist) for a list of artist object dicts.
-
-        For each artist, this function moves articles (such as 'a', 'an',
-        and 'the') to the front and strips trailing disambiguation numbers. It
-        returns a tuple containing the comma-separated string of all
-        normalized artists and the ``id`` of the main/first artist.
-        Alternatively a keyword can be used to combine artists together into a
-        single string by passing the join_key argument.
-
-        :param artists: Iterable of artist dicts or lists returned by API.
-        :type artists: list[dict] or list[list]
-        :param id_key: Key or index corresponding to the value of ``id`` for
-            the main/first artist. Defaults to 'id'.
-        :param name_key: Key or index corresponding to values of names
-            to concatenate for the artist string (containing all artists).
-            Defaults to 'name'.
-        :param join_key: Key or index corresponding to a field containing a
-            keyword to use for combining artists into a single string, for
-            example "Feat.", "Vs.", "And" or similar. The default is None
-            which keeps the default behaviour (comma-separated).
-        :return: Normalized artist string.
-        """
-        artist_id = None
-        artist_string = ""
-        artists = list(artists)  # In case a generator was passed.
-        total = len(artists)
-        for idx, artist in enumerate(artists):
-            if not artist_id:
-                artist_id = artist[id_key]
-            name = artist[name_key]
-            # Strip disambiguation number.
-            name = re.sub(r" \(\d+\)$", "", name)
-            # Move articles to the front.
-            name = re.sub(r"^(.*?), (a|an|the)$", r"\2 \1", name, flags=re.I)
-            # Use a join keyword if requested and available.
-            if idx < (total - 1):  # Skip joining on last.
-                if join_key and artist.get(join_key, None):
-                    name += f" {artist[join_key]} "
-                else:
-                    name += ", "
-            artist_string += name
-
-        return artist_string, artist_id
-
-    def _get_id(self, id_string: str) -> str | None:
-        """Parse release ID from the given ID string."""
-        return extract_release_id(self.data_source.lower(), id_string)
-
-    def candidates(
-        self, items: list[Item], artist: str, album: str, va_likely: bool
-    ) -> Iterable[AlbumInfo]:
-        query_filters = {"album": album}
-        if not va_likely:
-            query_filters["artist"] = artist
-        for result in self._search_api("album", query_filters):
-            if info := self.album_for_id(result["id"]):
-                yield info
-
-    def item_candidates(
-        self, item: Item, artist: str, title: str
-    ) -> Iterable[TrackInfo]:
-        for result in self._search_api(
-            "track", {"artist": artist}, keywords=title
-        ):
-            if info := self.track_for_id(result["id"]):
-                yield info
-
-    def album_distance(
-        self,
-        items: Sequence[Item],
-        album_info: AlbumInfo,
-        mapping: dict[Item, TrackInfo],
-    ) -> Distance:
-        return get_distance(
-            data_source=self.data_source, info=album_info, config=self.config
-        )
-
-    def track_distance(self, item: Item, info: TrackInfo) -> Distance:
-        return get_distance(
-            data_source=self.data_source, info=info, config=self.config
-        )
diff --git a/beets/test/helper.py b/beets/test/helper.py
index b86db5b23d..1a11252bb0 100644
--- a/beets/test/helper.py
+++ b/beets/test/helper.py
@@ -792,10 +792,12 @@ class AutotagStub:
 
     def install(self):
         self.patchers = [
-            patch("beets.plugins.album_for_id", lambda *_: None),
-            patch("beets.plugins.track_for_id", lambda *_: None),
-            patch("beets.plugins.candidates", self.candidates),
-            patch("beets.plugins.item_candidates", self.item_candidates),
+            patch("beets.metadata_plugins.album_for_id", lambda *_: None),
+            patch("beets.metadata_plugins.track_for_id", lambda *_: None),
+            patch("beets.metadata_plugins.candidates", self.candidates),
+            patch(
+                "beets.metadata_plugins.item_candidates", self.item_candidates
+            ),
         ]
         for p in self.patchers:
             p.start()
diff --git a/beets/util/id_extractors.py b/beets/util/id_extractors.py
index bbe2c32a4f..74eab260e2 100644
--- a/beets/util/id_extractors.py
+++ b/beets/util/id_extractors.py
@@ -43,6 +43,6 @@
 
 
 def extract_release_id(source: str, id_: str) -> str | None:
-    if m := PATTERN_BY_SOURCE[source].search(str(id_)):
+    if m := PATTERN_BY_SOURCE[source.lower()].search(str(id_)):
         return m[1]
     return None
diff --git a/beetsplug/beatport.py b/beetsplug/beatport.py
index 20147b5ccf..a40feeed93 100644
--- a/beetsplug/beatport.py
+++ b/beetsplug/beatport.py
@@ -14,9 +14,19 @@
 
 """Adds Beatport release and track search support to the autotagger"""
 
+from __future__ import annotations
+
 import json
 import re
 from datetime import datetime, timedelta
+from typing import (
+    TYPE_CHECKING,
+    Iterable,
+    Iterator,
+    Literal,
+    Sequence,
+    overload,
+)
 
 import confuse
 from requests_oauthlib import OAuth1Session
@@ -29,7 +39,14 @@
 import beets
 import beets.ui
 from beets.autotag.hooks import AlbumInfo, TrackInfo
-from beets.plugins import BeetsPlugin, MetadataSourcePlugin, get_distance
+from beets.metadata_plugins import MetadataSourcePlugin, artists_to_artist_str
+from beets.util.id_extractors import extract_release_id
+
+if TYPE_CHECKING:
+    from beets.importer import ImportSession
+    from beets.library import Item
+
+    from ._typing import JSONDict
 
 AUTH_ERRORS = (TokenRequestDenied, TokenMissing, VerifierMissing)
 USER_AGENT = f"beets/{beets.__version__} +https://beets.io/"
@@ -39,20 +56,6 @@ class BeatportAPIError(Exception):
     pass
 
 
-class BeatportObject:
-    def __init__(self, data):
-        self.beatport_id = data["id"]
-        self.name = str(data["name"])
-        if "releaseDate" in data:
-            self.release_date = datetime.strptime(
-                data["releaseDate"], "%Y-%m-%d"
-            )
-        if "artists" in data:
-            self.artists = [(x["id"], str(x["name"])) for x in data["artists"]]
-        if "genres" in data:
-            self.genres = [str(x["name"]) for x in data["genres"]]
-
-
 class BeatportClient:
     _api_base = "https://oauth-api.beatport.com"
 
@@ -77,7 +80,7 @@ def __init__(self, c_key, c_secret, auth_key=None, auth_secret=None):
         )
         self.api.headers = {"User-Agent": USER_AGENT}
 
-    def get_authorize_url(self):
+    def get_authorize_url(self) -> str:
         """Generate the URL for the user to authorize the application.
 
         Retrieves a request token from the Beatport API and returns the
@@ -99,15 +102,13 @@ def get_authorize_url(self):
             self._make_url("/identity/1/oauth/authorize")
         )
 
-    def get_access_token(self, auth_data):
+    def get_access_token(self, auth_data: str) -> tuple[str, str]:
         """Obtain the final access token and secret for the API.
 
         :param auth_data:   URL-encoded authorization data as displayed at
                             the authorization url (obtained via
                             :py:meth:`get_authorize_url`) after signing in
-        :type auth_data:    unicode
-        :returns:           OAuth resource owner key and secret
-        :rtype:             (unicode, unicode) tuple
+        :returns:           OAuth resource owner key and secret as unicode
         """
         self.api.parse_authorization_response(
             "https://beets.io/auth?" + auth_data
@@ -117,20 +118,37 @@ def get_access_token(self, auth_data):
         )
         return access_data["oauth_token"], access_data["oauth_token_secret"]
 
-    def search(self, query, release_type="release", details=True):
+    @overload
+    def search(
+        self,
+        query: str,
+        release_type: Literal["release"],
+        details: bool = True,
+    ) -> Iterator[BeatportRelease]: ...
+
+    @overload
+    def search(
+        self,
+        query: str,
+        release_type: Literal["track"],
+        details: bool = True,
+    ) -> Iterator[BeatportTrack]: ...
+
+    def search(
+        self,
+        query: str,
+        release_type: Literal["release", "track"],
+        details=True,
+    ) -> Iterator[BeatportRelease | BeatportTrack]:
         """Perform a search of the Beatport catalogue.
 
         :param query:           Query string
-        :param release_type:    Type of releases to search for, can be
-                                'release' or 'track'
+        :param release_type:    Type of releases to search for.
         :param details:         Retrieve additional information about the
                                 search results. Currently this will fetch
                                 the tracklist for releases and do nothing for
                                 tracks
         :returns:               Search results
-        :rtype:                 generator that yields
-                                py:class:`BeatportRelease` or
-                                :py:class:`BeatportTrack`
         """
         response = self._get(
             "catalog/3/search",
@@ -140,20 +158,18 @@ def search(self, query, release_type="release", details=True):
         )
         for item in response:
             if release_type == "release":
+                release = BeatportRelease(item)
                 if details:
-                    release = self.get_release(item["id"])
-                else:
-                    release = BeatportRelease(item)
+                    release.tracks = self.get_release_tracks(item["id"])
                 yield release
             elif release_type == "track":
                 yield BeatportTrack(item)
 
-    def get_release(self, beatport_id):
+    def get_release(self, beatport_id: str) -> BeatportRelease | None:
         """Get information about a single release.
 
         :param beatport_id:     Beatport ID of the release
         :returns:               The matching release
-        :rtype:                 :py:class:`BeatportRelease`
         """
         response = self._get("/catalog/3/releases", id=beatport_id)
         if response:
@@ -162,35 +178,33 @@ def get_release(self, beatport_id):
             return release
         return None
 
-    def get_release_tracks(self, beatport_id):
+    def get_release_tracks(self, beatport_id: str) -> list[BeatportTrack]:
         """Get all tracks for a given release.
 
         :param beatport_id:     Beatport ID of the release
         :returns:               Tracks in the matching release
-        :rtype:                 list of :py:class:`BeatportTrack`
         """
         response = self._get(
             "/catalog/3/tracks", releaseId=beatport_id, perPage=100
         )
         return [BeatportTrack(t) for t in response]
 
-    def get_track(self, beatport_id):
+    def get_track(self, beatport_id: str) -> BeatportTrack:
         """Get information about a single track.
 
         :param beatport_id:     Beatport ID of the track
         :returns:               The matching track
-        :rtype:                 :py:class:`BeatportTrack`
         """
         response = self._get("/catalog/3/tracks", id=beatport_id)
         return BeatportTrack(response[0])
 
-    def _make_url(self, endpoint):
+    def _make_url(self, endpoint: str) -> str:
         """Get complete URL for a given API endpoint."""
         if not endpoint.startswith("/"):
             endpoint = "/" + endpoint
         return self._api_base + endpoint
 
-    def _get(self, endpoint, **kwargs):
+    def _get(self, endpoint: str, **kwargs) -> list[JSONDict]:
         """Perform a GET request on a given API endpoint.
 
         Automatically extracts result data from the response and converts HTTP
@@ -211,48 +225,81 @@ def _get(self, endpoint, **kwargs):
         return response.json()["results"]
 
 
-class BeatportRelease(BeatportObject):
-    def __str__(self):
-        if len(self.artists) < 4:
-            artist_str = ", ".join(x[1] for x in self.artists)
+class BeatportObject:
+    beatport_id: str
+    name: str
+
+    release_date: datetime | None = None
+
+    artists: list[tuple[str, str]] | None = None
+    # tuple of artist id and artist name
+
+    def __init__(self, data: JSONDict):
+        self.beatport_id = str(data["id"])  # given as int in the response
+        self.name = str(data["name"])
+        if "releaseDate" in data:
+            self.release_date = datetime.strptime(
+                data["releaseDate"], "%Y-%m-%d"
+            )
+        if "artists" in data:
+            self.artists = [(x["id"], str(x["name"])) for x in data["artists"]]
+        if "genres" in data:
+            self.genres = [str(x["name"]) for x in data["genres"]]
+
+    def artists_str(self) -> str | None:
+        if self.artists is not None:
+            if len(self.artists) < 4:
+                artist_str = ", ".join(x[1] for x in self.artists)
+            else:
+                artist_str = "Various Artists"
         else:
-            artist_str = "Various Artists"
-        return "<BeatportRelease: {} - {} ({})>".format(
-            artist_str,
-            self.name,
-            self.catalog_number,
-        )
+            artist_str = None
+
+        return artist_str
+
+
+class BeatportRelease(BeatportObject):
+    catalog_number: str | None
+    label_name: str | None
+    category: str | None
+    url: str | None
+    genre: str | None
+
+    tracks: list[BeatportTrack] | None = None
+
+    def __init__(self, data: JSONDict):
+        super().__init__(data)
+
+        self.catalog_number = data.get("catalogNumber")
+        self.label_name = data.get("label", {}).get("name")
+        self.category = data.get("category")
+        self.genre = data.get("genre")
 
-    def __repr__(self):
-        return str(self).encode("utf-8")
-
-    def __init__(self, data):
-        BeatportObject.__init__(self, data)
-        if "catalogNumber" in data:
-            self.catalog_number = data["catalogNumber"]
-        if "label" in data:
-            self.label_name = data["label"]["name"]
-        if "category" in data:
-            self.category = data["category"]
         if "slug" in data:
             self.url = "https://beatport.com/release/{}/{}".format(
                 data["slug"], data["id"]
             )
-        self.genre = data.get("genre")
 
-
-class BeatportTrack(BeatportObject):
-    def __str__(self):
-        artist_str = ", ".join(x[1] for x in self.artists)
-        return "<BeatportTrack: {} - {} ({})>".format(
-            artist_str, self.name, self.mix_name
+    def __str__(self) -> str:
+        return "<BeatportRelease: {} - {} ({})>".format(
+            self.artists_str(),
+            self.name,
+            self.catalog_number,
         )
 
-    def __repr__(self):
-        return str(self).encode("utf-8")
 
-    def __init__(self, data):
-        BeatportObject.__init__(self, data)
+class BeatportTrack(BeatportObject):
+    title: str | None
+    mix_name: str | None
+    length: timedelta
+    url: str | None
+    track_number: int | None
+    bpm: str | None
+    initial_key: str | None
+    genre: str | None
+
+    def __init__(self, data: JSONDict):
+        super().__init__(data)
         if "title" in data:
             self.title = str(data["title"])
         if "mixName" in data:
@@ -279,11 +326,11 @@ def __init__(self, data):
             self.genre = str(data["genres"][0].get("name"))
 
 
-class BeatportPlugin(BeetsPlugin):
-    data_source = "Beatport"
+class BeatportPlugin(MetadataSourcePlugin):
+    _client: BeatportClient | None = None
 
     def __init__(self):
-        super().__init__()
+        super().__init__(data_source="Beatport")
         self.config.add(
             {
                 "apikey": "57713c3906af6f5def151b33601389176b37b429",
@@ -294,12 +341,19 @@ def __init__(self):
         )
         self.config["apikey"].redact = True
         self.config["apisecret"].redact = True
-        self.client = None
         self.register_listener("import_begin", self.setup)
 
-    def setup(self, session=None):
-        c_key = self.config["apikey"].as_str()
-        c_secret = self.config["apisecret"].as_str()
+    @property
+    def client(self) -> BeatportClient:
+        if self._client is None:
+            raise ValueError(
+                "Beatport client not initialized. Call setup() first."
+            )
+        return self._client
+
+    def setup(self, session: ImportSession):
+        c_key: str = self.config["apikey"].as_str()
+        c_secret: str = self.config["apisecret"].as_str()
 
         # Get the OAuth token from a file or log in.
         try:
@@ -312,9 +366,9 @@ def setup(self, session=None):
             token = tokendata["token"]
             secret = tokendata["secret"]
 
-        self.client = BeatportClient(c_key, c_secret, token, secret)
+        self._client = BeatportClient(c_key, c_secret, token, secret)
 
-    def authenticate(self, c_key, c_secret):
+    def authenticate(self, c_key: str, c_secret: str) -> tuple[str, str]:
         # Get the link for the OAuth page.
         auth_client = BeatportClient(c_key, c_secret)
         try:
@@ -341,44 +395,32 @@ def authenticate(self, c_key, c_secret):
 
         return token, secret
 
-    def _tokenfile(self):
+    def _tokenfile(self) -> str:
         """Get the path to the JSON file for storing the OAuth token."""
         return self.config["tokenfile"].get(confuse.Filename(in_app_dir=True))
 
-    def album_distance(self, items, album_info, mapping):
-        """Returns the Beatport source weight and the maximum source weight
-        for albums.
-        """
-        return get_distance(
-            data_source=self.data_source, info=album_info, config=self.config
-        )
-
-    def track_distance(self, item, track_info):
-        """Returns the Beatport source weight and the maximum source weight
-        for individual tracks.
-        """
-        return get_distance(
-            data_source=self.data_source, info=track_info, config=self.config
-        )
+    # ---------------------------------- search ---------------------------------- #
 
-    def candidates(self, items, artist, release, va_likely):
-        """Returns a list of AlbumInfo objects for beatport search results
-        matching release and artist (if not various).
-        """
+    def candidates(
+        self,
+        items: Sequence[Item],
+        artist: str,
+        album: str,
+        va_likely: bool,
+    ) -> Iterator[AlbumInfo]:
         if va_likely:
-            query = release
+            query = album
         else:
-            query = f"{artist} {release}"
+            query = f"{artist} {album}"
         try:
-            return self._get_releases(query)
+            yield from self._get_releases(query)
         except BeatportAPIError as e:
             self._log.debug("API Error: {0} (query: {1})", e, query)
-            return []
+            return
 
-    def item_candidates(self, item, artist, title):
-        """Returns a list of TrackInfo objects for beatport search results
-        matching title and artist.
-        """
+    def item_candidates(
+        self, item: Item, artist: str, title: str
+    ) -> Iterable[TrackInfo]:
         query = f"{artist} {title}"
         try:
             return self._get_tracks(query)
@@ -386,13 +428,15 @@ def item_candidates(self, item, artist, title):
             self._log.debug("API Error: {0} (query: {1})", e, query)
             return []
 
-    def album_for_id(self, release_id):
+    # --------------------------------- id lookup -------------------------------- #
+
+    def album_for_id(self, album_id: str):
         """Fetches a release by its Beatport ID and returns an AlbumInfo object
         or None if the query is not a valid ID or release is not found.
         """
-        self._log.debug("Searching for release {0}", release_id)
+        self._log.debug("Searching for release {0}", album_id)
 
-        if not (release_id := self._get_id(release_id)):
+        if not (release_id := extract_release_id("beatport", album_id)):
             self._log.debug("Not a valid Beatport release ID.")
             return None
 
@@ -401,11 +445,12 @@ def album_for_id(self, release_id):
             return self._get_album_info(release)
         return None
 
-    def track_for_id(self, track_id):
+    def track_for_id(self, track_id: str):
         """Fetches a track by its Beatport ID and returns a TrackInfo object
         or None if the track is not a valid Beatport ID or track is not found.
         """
         self._log.debug("Searching for track {0}", track_id)
+        # TODO: move to extractor
         match = re.search(r"(^|beatport\.com/track/.+/)(\d+)$", track_id)
         if not match:
             self._log.debug("Not a valid Beatport track ID.")
@@ -415,7 +460,9 @@ def track_for_id(self, track_id):
             return self._get_track_info(bp_track)
         return None
 
-    def _get_releases(self, query):
+    # ------------------------------- parsing utils ------------------------------ #
+
+    def _get_releases(self, query: str) -> Iterator[AlbumInfo]:
         """Returns a list of AlbumInfo objects for a beatport search query."""
         # Strip non-word characters from query. Things like "!" and "-" can
         # cause a query to return no results, even if they match the artist or
@@ -425,16 +472,22 @@ def _get_releases(self, query):
         # Strip medium information from query, Things like "CD1" and "disk 1"
         # can also negate an otherwise positive result.
         query = re.sub(r"\b(CD|disc)\s*\d+", "", query, flags=re.I)
-        albums = [self._get_album_info(x) for x in self.client.search(query)]
-        return albums
+        for beatport_release in self.client.search(query, "release"):
+            if beatport_release is None:
+                continue
+            yield self._get_album_info(beatport_release)
 
-    def _get_album_info(self, release):
+    def _get_album_info(self, release: BeatportRelease) -> AlbumInfo:
         """Returns an AlbumInfo object for a Beatport Release object."""
-        va = len(release.artists) > 3
+        va = release.artists is not None and len(release.artists) > 3
         artist, artist_id = self._get_artist(release.artists)
         if va:
             artist = "Various Artists"
-        tracks = [self._get_track_info(x) for x in release.tracks]
+        tracks: list[TrackInfo] = []
+        if release.tracks is not None:
+            tracks = [self._get_track_info(x) for x in release.tracks]
+
+        release_date = release.release_date
 
         return AlbumInfo(
             album=release.name,
@@ -445,18 +498,18 @@ def _get_album_info(self, release):
             tracks=tracks,
             albumtype=release.category,
             va=va,
-            year=release.release_date.year,
-            month=release.release_date.month,
-            day=release.release_date.day,
             label=release.label_name,
             catalognum=release.catalog_number,
             media="Digital",
             data_source=self.data_source,
             data_url=release.url,
             genre=release.genre,
+            year=release_date.year if release_date else None,
+            month=release_date.month if release_date else None,
+            day=release_date.day if release_date else None,
         )
 
-    def _get_track_info(self, track):
+    def _get_track_info(self, track: BeatportTrack) -> TrackInfo:
         """Returns a TrackInfo object for a Beatport Track object."""
         title = track.name
         if track.mix_name != "Original Mix":
@@ -482,9 +535,7 @@ def _get_artist(self, artists):
         """Returns an artist string (all artists) and an artist_id (the main
         artist) for a list of Beatport release or track artists.
         """
-        return MetadataSourcePlugin.get_artist(
-            artists=artists, id_key=0, name_key=1
-        )
+        return artists_to_artist_str(artists=artists, id_key=0, name_key=1)
 
     def _get_tracks(self, query):
         """Returns a list of TrackInfo objects for a Beatport query."""
diff --git a/beetsplug/chroma.py b/beetsplug/chroma.py
index 518a41776b..cf06f16e2d 100644
--- a/beetsplug/chroma.py
+++ b/beetsplug/chroma.py
@@ -19,12 +19,14 @@
 import re
 from collections import defaultdict
 from functools import cached_property, partial
+from typing import Iterable
 
 import acoustid
 import confuse
 
 from beets import config, plugins, ui, util
 from beets.autotag.hooks import Distance
+from beets.metadata_plugins import MetadataSourcePlugin, TrackInfo
 from beetsplug.musicbrainz import MusicBrainzPlugin
 
 API_KEY = "1vOwZtEn"
@@ -168,9 +170,9 @@ def _all_releases(items):
             yield release_id
 
 
-class AcoustidPlugin(plugins.BeetsPlugin):
+class AcoustidPlugin(MetadataSourcePlugin, plugins.BeetsPlugin):
     def __init__(self):
-        super().__init__()
+        super().__init__("acoustid")
 
         self.config.add(
             {
@@ -210,7 +212,7 @@ def candidates(self, items, artist, album, va_likely):
         self._log.debug("acoustid album candidates: {0}", len(albums))
         return albums
 
-    def item_candidates(self, item, artist, title):
+    def item_candidates(self, item, artist, title) -> Iterable[TrackInfo]:
         if item.path not in _matches:
             return []
 
@@ -223,6 +225,14 @@ def item_candidates(self, item, artist, title):
         self._log.debug("acoustid item candidates: {0}", len(tracks))
         return tracks
 
+    def album_for_id(self, *args, **kwargs):
+        # Lookup by fingerprint ID does not make too much sense.
+        return None
+
+    def track_for_id(self, *args, **kwargs):
+        # Lookup by fingerprint ID does not make too much sense.
+        return None
+
     def commands(self):
         submit_cmd = ui.Subcommand(
             "submit", help="submit Acoustid fingerprints"
diff --git a/beetsplug/deezer.py b/beetsplug/deezer.py
index 2e5d8473af..7795097555 100644
--- a/beetsplug/deezer.py
+++ b/beetsplug/deezer.py
@@ -18,6 +18,7 @@
 
 import collections
 import time
+from typing import TYPE_CHECKING, Literal, Sequence
 
 import requests
 import unidecode
@@ -26,12 +27,19 @@
 from beets.autotag import AlbumInfo, TrackInfo
 from beets.dbcore import types
 from beets.library import DateType
-from beets.plugins import BeetsPlugin, MetadataSourcePlugin
+from beets.metadata_plugins import (
+    IDResponse,
+    SearchApiMetadataSourcePlugin,
+    SearchFilter,
+    artists_to_artist_str,
+)
+from beets.util.id_extractors import extract_release_id
 
+if TYPE_CHECKING:
+    from ._typing import JSONDict
 
-class DeezerPlugin(MetadataSourcePlugin, BeetsPlugin):
-    data_source = "Deezer"
 
+class DeezerPlugin(SearchApiMetadataSourcePlugin):
     item_types = {
         "deezer_track_rank": types.INTEGER,
         "deezer_track_id": types.INTEGER,
@@ -45,7 +53,7 @@ class DeezerPlugin(MetadataSourcePlugin, BeetsPlugin):
     track_url = "https://api.deezer.com/track/"
 
     def __init__(self):
-        super().__init__()
+        super().__init__("Deezer")
 
     def commands(self):
         """Add beet UI commands to interact with Deezer."""
@@ -61,31 +69,20 @@ def func(lib, opts, args):
 
         return [deezer_update_cmd]
 
-    def fetch_data(self, url):
-        try:
-            response = requests.get(url, timeout=10)
-            response.raise_for_status()
-            data = response.json()
-        except requests.exceptions.RequestException as e:
-            self._log.error("Error fetching data from {}\n Error: {}", url, e)
-            return None
-        if "error" in data:
-            self._log.debug("Deezer API error: {}", data["error"]["message"])
-            return None
-        return data
+    # --------------------------------- id lookup -------------------------------- #
 
     def album_for_id(self, album_id: str) -> AlbumInfo | None:
         """Fetch an album by its Deezer ID or URL."""
-        if not (deezer_id := self._get_id(album_id)):
+        if not (deezer_album_id := extract_release_id("deezer", album_id)):
             return None
 
-        album_url = f"{self.album_url}{deezer_id}"
+        album_url = f"{self.album_url}{deezer_album_id}"
         if not (album_data := self.fetch_data(album_url)):
             return None
 
         contributors = album_data.get("contributors")
         if contributors is not None:
-            artist, artist_id = self.get_artist(contributors)
+            artist, artist_id = artists_to_artist_str(contributors)
         else:
             artist, artist_id = None, None
 
@@ -107,13 +104,17 @@ def album_for_id(self, album_id: str) -> AlbumInfo | None:
                 f"Invalid `release_date` returned by {self.data_source} API: "
                 f"{release_date!r}"
             )
-        tracks_obj = self.fetch_data(self.album_url + deezer_id + "/tracks")
+        tracks_obj = self.fetch_data(
+            self.album_url + deezer_album_id + "/tracks"
+        )
         if tracks_obj is None:
             return None
         try:
             tracks_data = tracks_obj["data"]
         except KeyError:
-            self._log.debug("Error fetching album tracks for {}", deezer_id)
+            self._log.debug(
+                "Error fetching album tracks for {}", deezer_album_id
+            )
             tracks_data = None
         if not tracks_data:
             return None
@@ -136,10 +137,10 @@ def album_for_id(self, album_id: str) -> AlbumInfo | None:
 
         return AlbumInfo(
             album=album_data["title"],
-            album_id=deezer_id,
-            deezer_album_id=deezer_id,
+            album_id=deezer_album_id,
+            deezer_album_id=deezer_album_id,
             artist=artist,
-            artist_credit=self.get_artist([album_data["artist"]])[0],
+            artist_credit=artists_to_artist_str([album_data["artist"]])[0],
             artist_id=artist_id,
             tracks=tracks,
             albumtype=album_data["record_type"],
@@ -157,63 +158,34 @@ def album_for_id(self, album_id: str) -> AlbumInfo | None:
             cover_art_url=album_data.get("cover_xl"),
         )
 
-    def _get_track(self, track_data):
-        """Convert a Deezer track object dict to a TrackInfo object.
-
-        :param track_data: Deezer Track object dict
-        :type track_data: dict
-        :return: TrackInfo object for track
-        :rtype: beets.autotag.hooks.TrackInfo
-        """
-        artist, artist_id = self.get_artist(
-            track_data.get("contributors", [track_data["artist"]])
-        )
-        return TrackInfo(
-            title=track_data["title"],
-            track_id=track_data["id"],
-            deezer_track_id=track_data["id"],
-            isrc=track_data.get("isrc"),
-            artist=artist,
-            artist_id=artist_id,
-            length=track_data["duration"],
-            index=track_data.get("track_position"),
-            medium=track_data.get("disk_number"),
-            deezer_track_rank=track_data.get("rank"),
-            medium_index=track_data.get("track_position"),
-            data_source=self.data_source,
-            data_url=track_data["link"],
-            deezer_updated=time.time(),
-        )
-
-    def track_for_id(self, track_id=None, track_data=None):
+    def track_for_id(self, track_id: str) -> None | TrackInfo:
         """Fetch a track by its Deezer ID or URL and return a
         TrackInfo object or None if the track is not found.
 
         :param track_id: (Optional) Deezer ID or URL for the track. Either
             ``track_id`` or ``track_data`` must be provided.
-        :type track_id: str
-        :param track_data: (Optional) Simplified track object dict. May be
-            provided instead of ``track_id`` to avoid unnecessary API calls.
-        :type track_data: dict
-        :return: TrackInfo object for track
-        :rtype: beets.autotag.hooks.TrackInfo or None
+
         """
-        if track_data is None:
-            if not (deezer_id := self._get_id(track_id)) or not (
-                track_data := self.fetch_data(f"{self.track_url}{deezer_id}")
-            ):
-                return None
+        if not (deezer_id := extract_release_id("deezer", track_id)):
+            self._log.debug("Invalid Deezer track_id: {}", track_id)
+            return None
+
+        if not (track_data := self.fetch_data(f"{self.track_url}{deezer_id}")):
+            self._log.debug("Track not found: {}", track_id)
+            return None
 
         track = self._get_track(track_data)
 
         # Get album's tracks to set `track.index` (position on the entire
         # release) and `track.medium_total` (total number of tracks on
         # the track's disc).
-        album_tracks_obj = self.fetch_data(
-            self.album_url + str(track_data["album"]["id"]) + "/tracks"
-        )
-        if album_tracks_obj is None:
+        if not (
+            album_tracks_obj := self.fetch_data(
+                self.album_url + str(track_data["album"]["id"]) + "/tracks"
+            )
+        ):
             return None
+
         try:
             album_tracks_data = album_tracks_obj["data"]
         except KeyError:
@@ -230,6 +202,51 @@ def track_for_id(self, track_id=None, track_data=None):
         track.medium_total = medium_total
         return track
 
+    # ---------------------------------- search ---------------------------------- #
+    # implemented in parent SearchApiMetadataSourcePluginNext
+
+    # ------------------------------- parsing utils ------------------------------ #
+
+    def _get_track(self, track_data: JSONDict) -> TrackInfo:
+        """Convert a Deezer track object dict to a TrackInfo object.
+
+        :param track_data: Deezer Track object dict
+        """
+        artist, artist_id = artists_to_artist_str(
+            track_data.get("contributors", [track_data["artist"]])
+        )
+        return TrackInfo(
+            title=track_data["title"],
+            track_id=track_data["id"],
+            deezer_track_id=track_data["id"],
+            isrc=track_data.get("isrc"),
+            artist=artist,
+            artist_id=artist_id,
+            length=track_data["duration"],
+            index=track_data.get("track_position"),
+            medium=track_data.get("disk_number"),
+            deezer_track_rank=track_data.get("rank"),
+            medium_index=track_data.get("track_position"),
+            data_source=self.data_source,
+            data_url=track_data["link"],
+            deezer_updated=time.time(),
+        )
+
+    # -------------------------------- fetch data -------------------------------- #
+
+    def fetch_data(self, url) -> JSONDict | None:
+        try:
+            response = requests.get(url, timeout=10)
+            response.raise_for_status()
+            data = response.json()
+        except requests.exceptions.RequestException as e:
+            self._log.error("Error fetching data from {}\n Error: {}", url, e)
+            return None
+        if "error" in data:
+            self._log.debug("Deezer API error: {}", data["error"]["message"])
+            return None
+        return data
+
     @staticmethod
     def _construct_search_query(filters=None, keywords=""):
         """Construct a query string with the specified filters and keywords to
@@ -252,14 +269,18 @@ def _construct_search_query(filters=None, keywords=""):
             query = query.decode("utf8")
         return unidecode.unidecode(query)
 
-    def _search_api(self, query_type, filters=None, keywords=""):
+    def _search_api(
+        self,
+        query_type: Literal["album", "track"],
+        filters: SearchFilter | None = None,
+        keywords="",
+    ) -> None | Sequence[IDResponse]:
         """Query the Deezer Search API for the specified ``keywords``, applying
         the provided ``filters``.
 
-        :param query_type: The Deezer Search API method to use. Valid types
-            are: 'album', 'artist', 'history', 'playlist', 'podcast',
-            'radio', 'track', 'user', and 'track'.
-        :type query_type: str
+        :param query_type: Valid types are: 'album', 'artist', 'history',
+            'playlist', 'podcast', 'radio', 'track', 'user', and 'track'.
+            But only `track` and `album` are used.
         :param filters: (Optional) Field filters to apply.
         :type filters: dict
         :param keywords: (Optional) Query keywords to use.
@@ -286,7 +307,7 @@ def _search_api(self, query_type, filters=None, keywords=""):
                 e,
             )
             return None
-        response_data = response.json().get("data", [])
+        response_data: Sequence[IDResponse] = response.json().get("data", [])
         self._log.debug(
             "Found {} result(s) from {} for '{}'",
             len(response_data),
diff --git a/beetsplug/discogs.py b/beetsplug/discogs.py
index 696f1d1acb..f459f6f715 100644
--- a/beetsplug/discogs.py
+++ b/beetsplug/discogs.py
@@ -27,11 +27,15 @@
 import traceback
 from functools import cache
 from string import ascii_lowercase
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Sequence
 
 import confuse
 from discogs_client import Client, Master, Release
+from discogs_client.exceptions import (
+    AuthorizationError as DiscogsAuthorizationError,
+)
 from discogs_client.exceptions import DiscogsAPIError
+from discogs_client.exceptions import HTTPError as DiscogsHTTPError
 from requests.exceptions import ConnectionError
 from typing_extensions import TypedDict
 
@@ -39,7 +43,7 @@
 import beets.ui
 from beets import config
 from beets.autotag.hooks import AlbumInfo, TrackInfo, string_dist
-from beets.plugins import BeetsPlugin, MetadataSourcePlugin, get_distance
+from beets.metadata_plugins import MetadataSourcePlugin, artists_to_artist_str
 from beets.util.id_extractors import extract_release_id
 
 if TYPE_CHECKING:
@@ -83,9 +87,9 @@ class ReleaseFormat(TypedDict):
     descriptions: list[str] | None
 
 
-class DiscogsPlugin(BeetsPlugin):
+class DiscogsPlugin(MetadataSourcePlugin):
     def __init__(self):
-        super().__init__()
+        super().__init__(data_source="Discogs")
         self.config.add(
             {
                 "apikey": API_KEY,
@@ -135,7 +139,7 @@ def reset_auth(self):
         os.remove(self._tokenfile())
         self.setup()
 
-    def _tokenfile(self):
+    def _tokenfile(self) -> str:
         """Get the path to the JSON file for storing the OAuth token."""
         return self.config["tokenfile"].get(confuse.Filename(in_app_dir=True))
 
@@ -168,20 +172,8 @@ def authenticate(self, c_key, c_secret):
 
         return token, secret
 
-    def album_distance(self, items, album_info, mapping):
-        """Returns the album distance."""
-        return get_distance(
-            data_source="Discogs", info=album_info, config=self.config
-        )
-
-    def track_distance(self, item, track_info):
-        """Returns the track distance."""
-        return get_distance(
-            data_source="Discogs", info=track_info, config=self.config
-        )
-
     def candidates(
-        self, items: list[Item], artist: str, album: str, va_likely: bool
+        self, items: Sequence[Item], artist: str, album: str, va_likely: bool
     ) -> Iterable[AlbumInfo]:
         return self.get_albums(f"{artist} {album}" if va_likely else album)
 
@@ -225,16 +217,24 @@ def album_for_id(self, album_id: str) -> AlbumInfo | None:
         # Try to obtain title to verify that we indeed have a valid Release
         try:
             getattr(result, "title")
-        except DiscogsAPIError as e:
+        except DiscogsAuthorizationError as e:
+            self._log.debug(
+                "Authorization Error: {0} (query: {1})",
+                e,
+                result.data["resource_url"],
+            )
+            self.reset_auth()
+            return self.album_for_id(album_id)
+        except DiscogsHTTPError as e:
             if e.status_code != 404:
                 self._log.debug(
                     "API Error: {0} (query: {1})",
                     e,
                     result.data["resource_url"],
                 )
-                if e.status_code == 401:
-                    self.reset_auth()
-                    return self.album_for_id(album_id)
+            if e.status_code == 401:
+                self.reset_auth()
+                return self.album_for_id(album_id)
             return None
         except CONNECTION_ERRORS:
             self._log.debug("Connection error in album lookup", exc_info=True)
@@ -271,7 +271,7 @@ def get_albums(self, query: str) -> Iterable[AlbumInfo]:
                 exc_info=True,
             )
             return []
-        return map(self.get_album_info, releases)
+        return filter(None, map(self.get_album_info, releases))
 
     @cache
     def get_master_year(self, master_id: str) -> int | None:
@@ -283,16 +283,16 @@ def get_master_year(self, master_id: str) -> int | None:
 
         try:
             return result.fetch("year")
-        except DiscogsAPIError as e:
+        except DiscogsHTTPError as e:
             if e.status_code != 404:
                 self._log.debug(
                     "API Error: {0} (query: {1})",
                     e,
                     result.data["resource_url"],
                 )
-                if e.status_code == 401:
-                    self.reset_auth()
-                    return self.get_master_year(master_id)
+            if e.status_code == 401:
+                self.reset_auth()
+                return self.get_master_year(master_id)
             return None
         except CONNECTION_ERRORS:
             self._log.debug(
@@ -312,7 +312,7 @@ def get_media_and_albumtype(
 
         return media, albumtype
 
-    def get_album_info(self, result):
+    def get_album_info(self, result) -> None | AlbumInfo:
         """Returns an AlbumInfo object for a discogs Release object."""
         # Explicitly reload the `Release` fields, as they might not be yet
         # present if the result is from a `discogs_client.search()`.
@@ -333,7 +333,7 @@ def get_album_info(self, result):
             self._log.warning("Release does not contain the required fields")
             return None
 
-        artist, artist_id = MetadataSourcePlugin.get_artist(
+        artist, artist_id = artists_to_artist_str(
             [a.data for a in result.artists], join_key="join"
         )
         album = re.sub(r" +", " ", result.title)
@@ -376,7 +376,7 @@ def get_album_info(self, result):
 
         # Additional cleanups (various artists name, catalog number, media).
         if va:
-            artist = config["va_name"].as_str()
+            artist: str = config["va_name"].as_str()
         if catalogno == "none":
             catalogno = None
         # Explicitly set the `media` for the tracks, since it is expected by
@@ -444,8 +444,11 @@ def format(self, classification):
         else:
             return None
 
-    def get_tracks(self, tracklist):
-        """Returns a list of TrackInfo objects for a discogs tracklist."""
+    def get_tracks(self, tracklist) -> list[TrackInfo]:
+        """Returns a list of TrackInfo objects for a discogs tracklist.
+
+        FIXME: This needs a look at and type hinting.
+        """
         try:
             clean_tracklist = self.coalesce_tracks(tracklist)
         except Exception as exc:
@@ -628,7 +631,7 @@ def add_merged_subtracks(tracklist, subtracks):
 
         return tracklist
 
-    def get_track_info(self, track, index, divisions):
+    def get_track_info(self, track, index, divisions) -> TrackInfo:
         """Returns a TrackInfo object for a discogs track."""
         title = track["title"]
         if self.config["index_tracks"]:
@@ -637,7 +640,7 @@ def get_track_info(self, track, index, divisions):
                 title = f"{prefix}: {title}"
         track_id = None
         medium, medium_index, _ = self.get_track_index(track["position"])
-        artist, artist_id = MetadataSourcePlugin.get_artist(
+        artist, artist_id = artists_to_artist_str(
             track.get("artists", []), join_key="join"
         )
         length = self.get_track_length(track["duration"])
diff --git a/beetsplug/mbsync.py b/beetsplug/mbsync.py
index 94870232c7..457454b29f 100644
--- a/beetsplug/mbsync.py
+++ b/beetsplug/mbsync.py
@@ -16,7 +16,7 @@
 
 from collections import defaultdict
 
-from beets import autotag, library, plugins, ui, util
+from beets import autotag, library, metadata_plugins, ui, util
 from beets.plugins import BeetsPlugin, apply_item_changes
 
 
@@ -79,7 +79,9 @@ def singletons(self, lib, query, move, pretend, write):
                 )
                 continue
 
-            if not (track_info := plugins.track_for_id(item.mb_trackid)):
+            if not (
+                track_info := metadata_plugins.track_for_id(item.mb_trackid)
+            ):
                 self._log.info(
                     "Recording ID not found: {0.mb_trackid} for track {0}", item
                 )
@@ -100,7 +102,9 @@ def albums(self, lib, query, move, pretend, write):
                 self._log.info("Skipping album with no mb_albumid: {}", album)
                 continue
 
-            if not (album_info := plugins.album_for_id(album.mb_albumid)):
+            if not (
+                album_info := metadata_plugins.album_for_id(album.mb_albumid)
+            ):
                 self._log.info(
                     "Release ID {0.mb_albumid} not found for album {0}", album
                 )
diff --git a/beetsplug/missing.py b/beetsplug/missing.py
index c4bbb83fda..9b66bd88e4 100644
--- a/beetsplug/missing.py
+++ b/beetsplug/missing.py
@@ -21,7 +21,7 @@
 import musicbrainzngs
 from musicbrainzngs.musicbrainz import MusicBrainzError
 
-from beets import config, plugins
+from beets import config, metadata_plugins
 from beets.dbcore import types
 from beets.library import Album, Item, Library
 from beets.plugins import BeetsPlugin
@@ -222,7 +222,7 @@ def _missing(self, album: Album) -> Iterator[Item]:
         item_mbids = {x.mb_trackid for x in album.items()}
         # fetch missing items
         # TODO: Implement caching that without breaking other stuff
-        if album_info := plugins.album_for_id(album.mb_albumid):
+        if album_info := metadata_plugins.album_for_id(album.mb_albumid):
             for track_info in album_info.tracks:
                 if track_info.track_id not in item_mbids:
                     self._log.debug(
diff --git a/beetsplug/musicbrainz.py b/beetsplug/musicbrainz.py
index ceb9311797..173ae57a83 100644
--- a/beetsplug/musicbrainz.py
+++ b/beetsplug/musicbrainz.py
@@ -20,7 +20,7 @@
 from collections import Counter
 from functools import cached_property
 from itertools import product
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Iterable, Sequence
 from urllib.parse import urljoin
 
 import musicbrainzngs
@@ -28,11 +28,11 @@
 import beets
 import beets.autotag.hooks
 from beets import config, plugins, util
-from beets.plugins import BeetsPlugin
+from beets.metadata_plugins import MetadataSourcePlugin
 from beets.util.id_extractors import extract_release_id
 
 if TYPE_CHECKING:
-    from collections.abc import Iterator, Sequence
+    from collections.abc import Sequence
     from typing import Literal
 
     from beets.library import Item
@@ -362,14 +362,12 @@ def _merge_pseudo_and_actual_album(
     return merged
 
 
-class MusicBrainzPlugin(BeetsPlugin):
-    data_source = "Musicbrainz"
-
+class MusicBrainzPlugin(MetadataSourcePlugin):
     def __init__(self):
         """Set up the python-musicbrainz-ngs module according to settings
         from the beets configuration. This should be called at startup.
         """
-        super().__init__()
+        super().__init__(data_source="Musicbrainz")
         self.config.add(
             {
                 "host": "musicbrainz.org",
@@ -767,7 +765,7 @@ def extra_mb_field_by_tag(self) -> dict[str, str]:
         return mb_field_by_tag
 
     def get_album_criteria(
-        self, items: list[Item], artist: str, album: str, va_likely: bool
+        self, items: Sequence[Item], artist: str, album: str, va_likely: bool
     ) -> dict[str, str]:
         criteria = {
             "release": album,
@@ -812,12 +810,11 @@ def _search_api(
 
     def candidates(
         self,
-        items: list[Item],
+        items: Sequence[Item],
         artist: str,
         album: str,
         va_likely: bool,
-        extra_tags: dict[str, Any] | None = None,
-    ) -> Iterator[beets.autotag.hooks.AlbumInfo]:
+    ) -> Iterable[beets.autotag.hooks.AlbumInfo]:
         criteria = self.get_album_criteria(items, artist, album, va_likely)
         release_ids = (r["id"] for r in self._search_api("release", criteria))
 
@@ -825,7 +822,7 @@ def candidates(
 
     def item_candidates(
         self, item: Item, artist: str, title: str
-    ) -> Iterator[beets.autotag.hooks.TrackInfo]:
+    ) -> Iterable[beets.autotag.hooks.TrackInfo]:
         criteria = {"artist": artist, "recording": title}
 
         yield from filter(
diff --git a/beetsplug/spotify.py b/beetsplug/spotify.py
index c0d2129714..7e78dbc039 100644
--- a/beetsplug/spotify.py
+++ b/beetsplug/spotify.py
@@ -25,6 +25,7 @@
 import re
 import time
 import webbrowser
+from typing import TYPE_CHECKING, Literal, Sequence
 
 import confuse
 import requests
@@ -34,7 +35,17 @@
 from beets.autotag.hooks import AlbumInfo, TrackInfo
 from beets.dbcore import types
 from beets.library import DateType
-from beets.plugins import BeetsPlugin, MetadataSourcePlugin
+from beets.metadata_plugins import (
+    IDResponse,
+    SearchApiMetadataSourcePlugin,
+    SearchFilter,
+    artists_to_artist_str,
+)
+from beets.util.id_extractors import extract_release_id
+
+if TYPE_CHECKING:
+    from ._typing import JSONDict
+
 
 DEFAULT_WAITING_TIME = 5
 
@@ -43,9 +54,7 @@ class SpotifyAPIError(Exception):
     pass
 
 
-class SpotifyPlugin(MetadataSourcePlugin, BeetsPlugin):
-    data_source = "Spotify"
-
+class SpotifyPlugin(SearchApiMetadataSourcePlugin):
     item_types = {
         "spotify_track_popularity": types.INTEGER,
         "spotify_acousticness": types.FLOAT,
@@ -88,7 +97,7 @@ class SpotifyPlugin(MetadataSourcePlugin, BeetsPlugin):
     }
 
     def __init__(self):
-        super().__init__()
+        super().__init__(data_source="Spotify")
         self.config.add(
             {
                 "mode": "list",
@@ -107,33 +116,28 @@ def __init__(self):
         self.config["client_id"].redact = True
         self.config["client_secret"].redact = True
 
-        self.tokenfile = self.config["tokenfile"].get(
-            confuse.Filename(in_app_dir=True)
-        )  # Path to the JSON file for storing the OAuth access token.
         self.setup()
 
     def setup(self):
         """Retrieve previously saved OAuth token or generate a new one."""
+        c_id: str = self.config["client_id"].as_str()
+        c_secret: str = self.config["client_secret"].as_str()
+
         try:
-            with open(self.tokenfile) as f:
+            with open(self._tokenfile()) as f:
                 token_data = json.load(f)
         except OSError:
-            self._authenticate()
+            self.authenticate(c_id, c_secret)
         else:
             self.access_token = token_data["access_token"]
 
-    def _authenticate(self):
+    def authenticate(self, c_id: str, c_secret: str):
         """Request an access token via the Client Credentials Flow:
         https://developer.spotify.com/documentation/general/guides/authorization-guide/#client-credentials-flow
         """
         headers = {
             "Authorization": "Basic {}".format(
-                base64.b64encode(
-                    ":".join(
-                        self.config[k].as_str()
-                        for k in ("client_id", "client_secret")
-                    ).encode()
-                ).decode()
+                base64.b64encode(":".join([c_id, c_secret]).encode()).decode()
             )
         }
         response = requests.post(
@@ -154,103 +158,33 @@ def _authenticate(self):
         self._log.debug(
             "{} access token: {}", self.data_source, self.access_token
         )
-        with open(self.tokenfile, "w") as f:
+        with open(self._tokenfile(), "w") as f:
             json.dump({"access_token": self.access_token}, f)
 
-    def _handle_response(
-        self, request_type, url, params=None, retry_count=0, max_retries=3
-    ):
-        """Send a request, reauthenticating if necessary.
+    def _tokenfile(self) -> str:
+        """Get the path to the JSON file for storing the OAuth token."""
+        return self.config["tokenfile"].get(confuse.Filename(in_app_dir=True))
 
-        :param request_type: Type of :class:`Request` constructor,
-            e.g. ``requests.get``, ``requests.post``, etc.
-        :type request_type: function
-        :param url: URL for the new :class:`Request` object.
-        :type url: str
-        :param params: (optional) list of tuples or bytes to send
-            in the query string for the :class:`Request`.
-        :type params: dict
-        :return: JSON data for the class:`Response <Response>` object.
-        :rtype: dict
-        """
-        try:
-            response = request_type(
-                url,
-                headers={"Authorization": f"Bearer {self.access_token}"},
-                params=params,
-                timeout=10,
-            )
-            response.raise_for_status()
-            return response.json()
-        except requests.exceptions.ReadTimeout:
-            self._log.error("ReadTimeout.")
-            raise SpotifyAPIError("Request timed out.")
-        except requests.exceptions.ConnectionError as e:
-            self._log.error(f"Network error: {e}")
-            raise SpotifyAPIError("Network error.")
-        except requests.exceptions.RequestException as e:
-            if e.response.status_code == 401:
-                self._log.debug(
-                    f"{self.data_source} access token has expired. "
-                    f"Reauthenticating."
-                )
-                self._authenticate()
-                return self._handle_response(request_type, url, params=params)
-            elif e.response.status_code == 404:
-                raise SpotifyAPIError(
-                    f"API Error: {e.response.status_code}\n"
-                    f"URL: {url}\nparams: {params}"
-                )
-            elif e.response.status_code == 429:
-                if retry_count >= max_retries:
-                    raise SpotifyAPIError("Maximum retries reached.")
-                seconds = response.headers.get(
-                    "Retry-After", DEFAULT_WAITING_TIME
-                )
-                self._log.debug(
-                    f"Too many API requests. Retrying after {seconds} seconds."
-                )
-                time.sleep(int(seconds) + 1)
-                return self._handle_response(
-                    request_type,
-                    url,
-                    params=params,
-                    retry_count=retry_count + 1,
-                )
-            elif e.response.status_code == 503:
-                self._log.error("Service Unavailable.")
-                raise SpotifyAPIError("Service Unavailable.")
-            elif e.response.status_code == 502:
-                self._log.error("Bad Gateway.")
-                raise SpotifyAPIError("Bad Gateway.")
-            elif e.response is not None:
-                raise SpotifyAPIError(
-                    f"{self.data_source} API error:\n{e.response.text}\n"
-                    f"URL:\n{url}\nparams:\n{params}"
-                )
-            else:
-                self._log.error(f"Request failed. Error: {e}")
-                raise SpotifyAPIError("Request failed.")
+    def reset_auth(self):
+        """Redo the auth steps."""
+        self.setup()
 
-    def album_for_id(self, album_id: str) -> AlbumInfo | None:
-        """Fetch an album by its Spotify ID or URL and return an
-        AlbumInfo object or None if the album is not found.
+    # ---------------------------------- search ---------------------------------- #
+    # implemented in parent SearchApiMetadataSourcePluginNext
 
-        :param album_id: Spotify ID or URL for the album
-        :type album_id: str
-        :return: AlbumInfo object for album
-        :rtype: beets.autotag.hooks.AlbumInfo or None
-        """
-        if not (spotify_id := self._get_id(album_id)):
+    # --------------------------------- id lookup -------------------------------- #
+
+    def album_for_id(self, album_id: str) -> AlbumInfo | None:
+        if not (spotify_album_id := extract_release_id("spotify", album_id)):
             return None
 
         album_data = self._handle_response(
-            requests.get, self.album_url + spotify_id
+            "get", self.album_url + spotify_album_id
         )
         if album_data["name"] == "":
             self._log.debug("Album removed from Spotify: {}", album_id)
             return None
-        artist, artist_id = self.get_artist(album_data["artists"])
+        artist, artist_id = artists_to_artist_str(album_data["artists"])
 
         date_parts = [
             int(part) for part in album_data["release_date"].split("-")
@@ -277,9 +211,7 @@ def album_for_id(self, album_id: str) -> AlbumInfo | None:
         tracks_data = album_data["tracks"]
         tracks_items = tracks_data["items"]
         while tracks_data["next"]:
-            tracks_data = self._handle_response(
-                requests.get, tracks_data["next"]
-            )
+            tracks_data = self._handle_response("get", tracks_data["next"])
             tracks_items.extend(tracks_data["items"])
 
         tracks = []
@@ -294,8 +226,8 @@ def album_for_id(self, album_id: str) -> AlbumInfo | None:
 
         return AlbumInfo(
             album=album_data["name"],
-            album_id=spotify_id,
-            spotify_album_id=spotify_id,
+            album_id=spotify_album_id,
+            spotify_album_id=spotify_album_id,
             artist=artist,
             artist_id=artist_id,
             spotify_artist_id=artist_id,
@@ -312,7 +244,39 @@ def album_for_id(self, album_id: str) -> AlbumInfo | None:
             data_url=album_data["external_urls"]["spotify"],
         )
 
-    def _get_track(self, track_data):
+    def track_for_id(self, track_id: str) -> None | TrackInfo:
+        if not (spotify_track_id := extract_release_id("spotify", track_id)):
+            self._log.debug("Invalid Spotify ID: {}", track_id)
+            return None
+
+        if not (
+            track_data := self._handle_response(
+                "get", f"{self.track_url}{spotify_track_id}"
+            )
+        ):
+            self._log.debug("Track not found: {}", track_id)
+            return None
+
+        track = self._get_track(track_data)
+
+        # Get album's tracks to set `track.index` (position on the entire
+        # release) and `track.medium_total` (total number of tracks on
+        # the track's disc).
+        album_data = self._handle_response(
+            "get", self.album_url + track_data["album"]["id"]
+        )
+        medium_total = 0
+        for i, track_data in enumerate(album_data["tracks"]["items"], start=1):
+            if track_data["disc_number"] == track.medium:
+                medium_total += 1
+                if track_data["id"] == track.track_id:
+                    track.index = i
+        track.medium_total = medium_total
+        return track
+
+    # ------------------------------- parsing utils ------------------------------ #
+
+    def _get_track(self, track_data: JSONDict) -> TrackInfo:
         """Convert a Spotify track object dict to a TrackInfo object.
 
         :param track_data: Simplified track object
@@ -321,7 +285,7 @@ def _get_track(self, track_data):
         :return: TrackInfo object for track
         :rtype: beets.autotag.hooks.TrackInfo
         """
-        artist, artist_id = self.get_artist(track_data["artists"])
+        artist, artist_id = artists_to_artist_str(track_data["artists"])
 
         # Get album information for spotify tracks
         try:
@@ -344,43 +308,89 @@ def _get_track(self, track_data):
             data_url=track_data["external_urls"]["spotify"],
         )
 
-    def track_for_id(self, track_id=None, track_data=None):
-        """Fetch a track by its Spotify ID or URL and return a
-        TrackInfo object or None if the track is not found.
+    # ---------------------------------------------------------------------------- #
 
-        :param track_id: (Optional) Spotify ID or URL for the track. Either
-            ``track_id`` or ``track_data`` must be provided.
-        :type track_id: str
-        :param track_data: (Optional) Simplified track object dict. May be
-            provided instead of ``track_id`` to avoid unnecessary API calls.
-        :type track_data: dict
-        :return: TrackInfo object for track
-        :rtype: beets.autotag.hooks.TrackInfo or None
+    def _handle_response(
+        self,
+        method: Literal["get", "post", "put", "delete"],
+        url,
+        params=None,
+        retry_count=0,
+        max_retries=3,
+    ) -> JSONDict:
+        """Send a request, reauthenticating if necessary.
+
+        :param url: URL for the new :class:`Request` object.
+        :type url: str
+        :param params: (optional) list of tuples or bytes to send
+            in the query string for the :class:`Request`.
+        :type params: dict
+        :return: JSON data for the class:`Response <Response>` object.
+        :rtype: dict
         """
-        if not track_data:
-            if not (spotify_id := self._get_id(track_id)) or not (
-                track_data := self._handle_response(
-                    requests.get, f"{self.track_url}{spotify_id}"
+        try:
+            response = requests.request(
+                method,
+                url,
+                headers={"Authorization": f"Bearer {self.access_token}"},
+                params=params,
+                timeout=10,
+            )
+            response.raise_for_status()
+            return response.json()
+        except requests.exceptions.ReadTimeout:
+            self._log.error("ReadTimeout.")
+            raise SpotifyAPIError("Request timed out.")
+        except requests.exceptions.ConnectionError as e:
+            self._log.error(f"Network error: {e}")
+            raise SpotifyAPIError("Network error.")
+        except requests.exceptions.RequestException as e:
+            if e.response is None:
+                self._log.error(f"Request failed: {e}")
+                raise SpotifyAPIError("Request failed.")
+            if e.response.status_code == 401:
+                self._log.debug(
+                    f"{self.data_source} access token has expired. "
+                    f"Reauthenticating."
+                )
+                self.reset_auth()
+                return self._handle_response(method, url, params=params)
+            elif e.response.status_code == 404:
+                raise SpotifyAPIError(
+                    f"API Error: {e.response.status_code}\n"
+                    f"URL: {url}\nparams: {params}"
+                )
+            elif e.response.status_code == 429:
+                if retry_count >= max_retries:
+                    raise SpotifyAPIError("Maximum retries reached.")
+                seconds = response.headers.get(
+                    "Retry-After", DEFAULT_WAITING_TIME
+                )
+                self._log.debug(
+                    f"Too many API requests. Retrying after {seconds} seconds."
                 )
-            ):
-                return None
-
-        track = self._get_track(track_data)
 
-        # Get album's tracks to set `track.index` (position on the entire
-        # release) and `track.medium_total` (total number of tracks on
-        # the track's disc).
-        album_data = self._handle_response(
-            requests.get, self.album_url + track_data["album"]["id"]
-        )
-        medium_total = 0
-        for i, track_data in enumerate(album_data["tracks"]["items"], start=1):
-            if track_data["disc_number"] == track.medium:
-                medium_total += 1
-                if track_data["id"] == track.track_id:
-                    track.index = i
-        track.medium_total = medium_total
-        return track
+                time.sleep(int(seconds) + 1)
+                return self._handle_response(
+                    method,
+                    url,
+                    params=params,
+                    retry_count=retry_count + 1,
+                )
+            elif e.response.status_code == 503:
+                self._log.error("Service Unavailable.")
+                raise SpotifyAPIError("Service Unavailable.")
+            elif e.response.status_code == 502:
+                self._log.error("Bad Gateway.")
+                raise SpotifyAPIError("Bad Gateway.")
+            elif e.response is not None:
+                raise SpotifyAPIError(
+                    f"{self.data_source} API error:\n{e.response.text}\n"
+                    f"URL:\n{url}\nparams:\n{params}"
+                )
+            else:
+                self._log.error(f"Request failed. Error: {e}")
+                raise SpotifyAPIError("Request failed.")
 
     @staticmethod
     def _construct_search_query(filters=None, keywords=""):
@@ -404,20 +414,21 @@ def _construct_search_query(filters=None, keywords=""):
             query = query.decode("utf8")
         return unidecode.unidecode(query)
 
-    def _search_api(self, query_type, filters=None, keywords=""):
+    def _search_api(
+        self,
+        query_type: Literal["album", "track"],
+        filters: SearchFilter | None = None,
+        keywords="",
+    ) -> Sequence[IDResponse] | None:
         """Query the Spotify Search API for the specified ``keywords``,
         applying the provided ``filters``.
 
         :param query_type: Item type to search across. Valid types are:
             'album', 'artist', 'playlist', and 'track'.
-        :type query_type: str
         :param filters: (Optional) Field filters to apply.
-        :type filters: dict
         :param keywords: (Optional) Query keywords to use.
-        :type keywords: str
         :return: JSON data for the class:`Response <Response>` object or None
             if no search results are returned.
-        :rtype: dict or None
         """
         query = self._construct_search_query(keywords=keywords, filters=filters)
         if not query:
@@ -425,14 +436,16 @@ def _search_api(self, query_type, filters=None, keywords=""):
         self._log.debug(f"Searching {self.data_source} for '{query}'")
         try:
             response = self._handle_response(
-                requests.get,
+                "get",
                 self.search_url,
                 params={"q": query, "type": query_type},
             )
         except SpotifyAPIError as e:
             self._log.debug("Spotify API error: {}", e)
             return []
-        response_data = response.get(query_type + "s", {}).get("items", [])
+        response_data: Sequence[IDResponse] = response.get(
+            query_type + "s", {}
+        ).get("items", [])
         self._log.debug(
             "Found {} result(s) from {} for '{}'",
             len(response_data),
@@ -553,7 +566,7 @@ def _match_library_tracks(self, library, keywords):
             keywords = item[self.config["track_field"].get()]
 
             # Query the Web API for each track, look for the items' JSON data
-            query_filters = {"artist": artist, "album": album}
+            query_filters: SearchFilter = {"artist": artist, "album": album}
             response_data_tracks = self._search_api(
                 query_type="track", keywords=keywords, filters=query_filters
             )
@@ -686,25 +699,25 @@ def _fetch_info(self, items, write, force):
     def track_info(self, track_id=None):
         """Fetch a track's popularity and external IDs using its Spotify ID."""
         track_data = self._handle_response(
-            requests.get, self.track_url + track_id
+            "get", self.track_url + (track_id or "")
         )
         self._log.debug(
             "track_popularity: {} and track_isrc: {}",
             track_data.get("popularity"),
-            track_data.get("external_ids").get("isrc"),
+            track_data.get("external_ids", {}).get("isrc"),
         )
         return (
             track_data.get("popularity"),
-            track_data.get("external_ids").get("isrc"),
-            track_data.get("external_ids").get("ean"),
-            track_data.get("external_ids").get("upc"),
+            track_data.get("external_ids", {}).get("isrc"),
+            track_data.get("external_ids", {}).get("ean"),
+            track_data.get("external_ids", {}).get("upc"),
         )
 
     def track_audio_features(self, track_id=None):
         """Fetch track audio features by its Spotify ID."""
         try:
             return self._handle_response(
-                requests.get, self.audio_features_url + track_id
+                "get", self.audio_features_url + (track_id or "")
             )
         except SpotifyAPIError as e:
             self._log.debug("Spotify API error: {}", e)
diff --git a/docs/changelog.rst b/docs/changelog.rst
index 9bc065419a..1334bb06ce 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -42,9 +42,20 @@ For plugin developers:
 * The `fetchart` plugins has seen a few changes to function signatures and
   source registration in the process of introducing typings to the code.
   Custom art sources might need to be adapted.
-
+* We split the responsibilities of plugins into two base classes
+    #. :class:`beets.plugins.BeetsPlugin` 
+        is the base class for all plugins, any plugin needs to inherit from this class.
+    #. :class:`beets.metadata_plugin.MetadataSourcePlugin`
+        allows plugins to act like metadata sources. E.g. used by the MusicBrainz plugin. All plugins
+        in the beets repo are opted into this class where applicable. If you are maintaining a plugin
+        that acts like a metadata source, i.e. you expose any of `track_for_id,
+        album_for_id, candidates, item_candidates, album_distance, track_distance` methods,
+        please update your plugin to inherit from the new baseclass, as otherwise it will
+        not be registered as a metadata source and wont be usable going forward.
+        
 Other changes:
 
+* Refactor: Split responsibilities of Plugins into MetaDataPlugins and general Plugins.
 * Documentation structure for auto generated API references changed slightly.
   Autogenerated API references are now located in the `docs/api` subdirectory.
 
diff --git a/docs/dev/index.rst b/docs/dev/index.rst
index 10b3566c21..b6f02f7cc7 100644
--- a/docs/dev/index.rst
+++ b/docs/dev/index.rst
@@ -4,15 +4,17 @@ For Developers
 This section contains information for developers. Read on if you're interested
 in hacking beets itself or creating plugins for it.
 
-See also the documentation for `MediaFile`_, the library used by beets to read
-and write metadata tags in media files.
+
+See also the documentation for the `MediaFile`_ and `Confuse`_ libraries. These are maintained by the beets team and used to read and write metadata tags and manage configuration files, respectively.
 
 .. _MediaFile: https://mediafile.readthedocs.io/en/latest/
+.. _Confuse: https://confuse.readthedocs.io/en/latest/
 
 .. toctree::
     :maxdepth: 1
+    :caption: Guides
 
-    plugins
+    plugins/index
     library
     importer
     cli
diff --git a/docs/dev/plugins.rst b/docs/dev/plugins.rst
deleted file mode 100644
index c24a940938..0000000000
--- a/docs/dev/plugins.rst
+++ /dev/null
@@ -1,660 +0,0 @@
-Plugin Development Guide
-========================
-
-Beets plugins are Python modules or packages that extend the core functionality
-of beets. The plugin system is designed to be flexible, allowing developers to
-add virtually any type of features.
-
-
-.. _writing-plugins:
-
-Writing Plugins
----------------
-
-A beets plugin is just a Python module or package inside the ``beetsplug``
-namespace package. (Check out `this article`_ and `this Stack Overflow
-question`_ if you haven't heard about namespace packages.) So, to make one,
-create a directory called ``beetsplug`` and add either your plugin module::
-
-    beetsplug/
-        myawesomeplugin.py
-
-or your plugin subpackage::
-
-    beetsplug/
-        myawesomeplugin/
-            __init__.py
-            myawesomeplugin.py
-
-.. attention::
-
-    You do not anymore need to add a ``__init__.py`` file to the ``beetsplug``
-    directory. Python treats your plugin as a namespace package automatically,
-    thus we do not depend on ``pkgutil``-based setup in the ``__init__.py``
-    file anymore.
-
-The meat of your plugin goes in ``myawesomeplugin.py``. There, you'll have to
-import ``BeetsPlugin`` from ``beets.plugins`` and subclass it, for example
-
-.. code-block:: python
-
-    from beets.plugins import BeetsPlugin
-
-    class MyAwesomePlugin(BeetsPlugin):
-        pass
-
-Once you have your ``BeetsPlugin`` subclass, there's a variety of things your
-plugin can do. (Read on!)
-
-To use your new plugin, package your plugin (see how to do this with `poetry`_
-or `setuptools`_, for example) and install it into your ``beets`` virtual
-environment. Then, add your plugin to beets configuration
-
-.. code-block:: yaml
-
-    # config.yaml
-    plugins:
-      - myawesomeplugin
-
-and you're good to go!
-
-.. _this article: https://realpython.com/python-namespace-package/#setting-up-some-namespace-packages
-.. _this Stack Overflow question: https://stackoverflow.com/a/27586272/9582674
-.. _poetry: https://python-poetry.org/docs/pyproject/#packages
-.. _setuptools: https://setuptools.pypa.io/en/latest/userguide/package_discovery.html#finding-simple-packages
-
-.. _add_subcommands:
-
-Add Commands to the CLI
-^^^^^^^^^^^^^^^^^^^^^^^
-
-Plugins can add new subcommands to the ``beet`` command-line interface. Define
-the plugin class' ``commands()`` method to return a list of ``Subcommand``
-objects. (The ``Subcommand`` class is defined in the ``beets.ui`` module.)
-Here's an example plugin that adds a simple command::
-
-    from beets.plugins import BeetsPlugin
-    from beets.ui import Subcommand
-
-    my_super_command = Subcommand('super', help='do something super')
-    def say_hi(lib, opts, args):
-        print "Hello everybody! I'm a plugin!"
-    my_super_command.func = say_hi
-
-    class SuperPlug(BeetsPlugin):
-        def commands(self):
-            return [my_super_command]
-
-To make a subcommand, invoke the constructor like so: ``Subcommand(name, parser,
-help, aliases)``. The ``name`` parameter is the only required one and should
-just be the name of your command. ``parser`` can be an `OptionParser instance`_,
-but it defaults to an empty parser (you can extend it later). ``help`` is a
-description of your command, and ``aliases`` is a list of shorthand versions of
-your command name.
-
-.. _OptionParser instance: https://docs.python.org/library/optparse.html
-
-You'll need to add a function to your command by saying ``mycommand.func =
-myfunction``. This function should take the following parameters: ``lib`` (a
-beets ``Library`` object) and ``opts`` and ``args`` (command-line options and
-arguments as returned by `OptionParser.parse_args`_).
-
-.. _OptionParser.parse_args:
-    https://docs.python.org/library/optparse.html#parsing-arguments
-
-The function should use any of the utility functions defined in ``beets.ui``.
-Try running ``pydoc beets.ui`` to see what's available.
-
-You can add command-line options to your new command using the ``parser`` member
-of the ``Subcommand`` class, which is a ``CommonOptionsParser`` instance. Just
-use it like you would a normal ``OptionParser`` in an independent script. Note
-that it offers several methods to add common options: ``--album``, ``--path``
-and ``--format``. This feature is versatile and extensively documented, try
-``pydoc beets.ui.CommonOptionsParser`` for more information.
-
-.. _plugin_events:
-
-Listen for Events
-^^^^^^^^^^^^^^^^^
-
-Event handlers allow plugins to run code whenever something happens in beets'
-operation. For instance, a plugin could write a log message every time an album
-is successfully autotagged or update MPD's index whenever the database is
-changed.
-
-You can "listen" for events using ``BeetsPlugin.register_listener``. Here's
-an example::
-
-    from beets.plugins import BeetsPlugin
-
-    def loaded():
-        print 'Plugin loaded!'
-
-    class SomePlugin(BeetsPlugin):
-      def __init__(self):
-        super().__init__()
-        self.register_listener('pluginload', loaded)
-
-Note that if you want to access an attribute of your plugin (e.g. ``config`` or
-``log``) you'll have to define a method and not a function. Here is the usual
-registration process in this case::
-
-    from beets.plugins import BeetsPlugin
-
-    class SomePlugin(BeetsPlugin):
-      def __init__(self):
-        super().__init__()
-        self.register_listener('pluginload', self.loaded)
-
-      def loaded(self):
-        self._log.info('Plugin loaded!')
-
-The events currently available are:
-
-* `pluginload`: called after all the plugins have been loaded after the ``beet``
-  command starts
-
-* `import`: called after a ``beet import`` command finishes (the ``lib`` keyword
-  argument is a Library object; ``paths`` is a list of paths (strings) that were
-  imported)
-
-* `album_imported`: called with an ``Album`` object every time the ``import``
-  command finishes adding an album to the library. Parameters: ``lib``,
-  ``album``
-
-* `album_removed`: called with an ``Album`` object every time an album is
-  removed from the library (even when its file is not deleted from disk).
-
-* `item_copied`: called with an ``Item`` object whenever its file is copied.
-  Parameters: ``item``, ``source`` path, ``destination`` path
-
-* `item_imported`: called with an ``Item`` object every time the importer adds a
-  singleton to the library (not called for full-album imports). Parameters:
-  ``lib``, ``item``
-
-* `before_item_moved`: called with an ``Item`` object immediately before its
-  file is moved. Parameters: ``item``, ``source`` path, ``destination`` path
-
-* `item_moved`: called with an ``Item`` object whenever its file is moved.
-  Parameters: ``item``, ``source`` path, ``destination`` path
-
-* `item_linked`: called with an ``Item`` object whenever a symlink is created
-  for a file.
-  Parameters: ``item``, ``source`` path, ``destination`` path
-
-* `item_hardlinked`: called with an ``Item`` object whenever a hardlink is
-  created for a file.
-  Parameters: ``item``, ``source`` path, ``destination`` path
-
-* `item_reflinked`: called with an ``Item`` object whenever a reflink is
-  created for a file.
-  Parameters: ``item``, ``source`` path, ``destination`` path
-
-* `item_removed`: called with an ``Item`` object every time an item (singleton
-  or album's part) is removed from the library (even when its file is not
-  deleted from disk).
-
-* `write`: called with an ``Item`` object, a ``path``, and a ``tags``
-  dictionary just before a file's metadata is written to disk (i.e.,
-  just before the file on disk is opened). Event handlers may change
-  the ``tags`` dictionary to customize the tags that are written to the
-  media file. Event handlers may also raise a
-  ``library.FileOperationError`` exception to abort the write
-  operation. Beets will catch that exception, print an error message
-  and continue.
-
-* `after_write`: called with an ``Item`` object after a file's metadata is
-  written to disk (i.e., just after the file on disk is closed).
-
-* `import_task_created`: called immediately after an import task is
-  initialized. Plugins can use this to, for example, change imported files of a
-  task before anything else happens. It's also possible to replace the task
-  with another task by returning a list of tasks. This list can contain zero
-  or more `ImportTask`s. Returning an empty list will stop the task.
-  Parameters: ``task`` (an `ImportTask`) and ``session`` (an `ImportSession`).
-
-* `import_task_start`: called when before an import task begins processing.
-  Parameters: ``task`` and ``session``.
-
-* `import_task_apply`: called after metadata changes have been applied in an
-  import task. This is called on the same thread as the UI, so use this
-  sparingly and only for tasks that can be done quickly. For most plugins, an
-  import pipeline stage is a better choice (see :ref:`plugin-stage`).
-  Parameters: ``task`` and ``session``.
-
-* `import_task_before_choice`: called after candidate search for an import task
-  before any decision is made about how/if to import or tag. Can be used to
-  present information about the task or initiate interaction with the user
-  before importing occurs. Return an importer action to take a specific action.
-  Only one handler may return a non-None result.
-  Parameters: ``task`` and ``session``
-
-* `import_task_choice`: called after a decision has been made about an import
-  task. This event can be used to initiate further interaction with the user.
-  Use ``task.choice_flag`` to determine or change the action to be
-  taken. Parameters: ``task`` and ``session``.
-
-* `import_task_files`: called after an import task finishes manipulating the
-  filesystem (copying and moving files, writing metadata tags). Parameters:
-  ``task`` and ``session``.
-
-* `library_opened`: called after beets starts up and initializes the main
-  Library object. Parameter: ``lib``.
-
-* `database_change`: a modification has been made to the library database. The
-  change might not be committed yet. Parameters: ``lib`` and ``model``.
-
-* `cli_exit`: called just before the ``beet`` command-line program exits.
-  Parameter: ``lib``.
-
-* `import_begin`: called just before a ``beet import`` session starts up.
-  Parameter: ``session``.
-
-* `trackinfo_received`: called after metadata for a track item has been
-  fetched from a data source, such as MusicBrainz. You can modify the tags
-  that the rest of the pipeline sees on a ``beet import`` operation or during
-  later adjustments, such as ``mbsync``. Slow handlers of the event can impact
-  the operation, since the event is fired for any fetched possible match
-  `before` the user (or the autotagger machinery) gets to see the match.
-  Parameter: ``info``.
-
-* `albuminfo_received`: like `trackinfo_received`, the event indicates new
-  metadata for album items. The parameter is an ``AlbumInfo`` object instead
-  of a ``TrackInfo``.
-  Parameter: ``info``.
-
-* `before_choose_candidate`: called before the user is prompted for a decision
-  during a ``beet import`` interactive session. Plugins can use this event for
-  :ref:`appending choices to the prompt <append_prompt_choices>` by returning a
-  list of ``PromptChoices``. Parameters: ``task`` and ``session``.
-
-* `mb_track_extract`: called after the metadata is obtained from
-  MusicBrainz. The parameter is a ``dict`` containing the tags retrieved from
-  MusicBrainz for a track. Plugins must return a new (potentially empty)
-  ``dict`` with additional ``field: value`` pairs, which the autotagger will
-  apply to the item, as flexible attributes if ``field`` is not a hardcoded
-  field. Fields already present on the track are overwritten.
-  Parameter: ``data``
-
-* `mb_album_extract`: Like `mb_track_extract`, but for album tags. Overwrites
-  tags set at the track level, if they have the same ``field``.
-  Parameter: ``data``
-
-The included ``mpdupdate`` plugin provides an example use case for event listeners.
-
-Extend the Autotagger
-^^^^^^^^^^^^^^^^^^^^^
-
-Plugins can also enhance the functionality of the autotagger. For a
-comprehensive example, try looking at the ``chroma`` plugin, which is included
-with beets.
-
-A plugin can extend three parts of the autotagger's process: the track distance
-function, the album distance function, and the initial MusicBrainz search. The
-distance functions determine how "good" a match is at the track and album
-levels; the initial search controls which candidates are presented to the
-matching algorithm. Plugins implement these extensions by implementing four
-methods on the plugin class:
-
-* ``track_distance(self, item, info)``: adds a component to the distance
-  function (i.e., the similarity metric) for individual tracks. ``item`` is the
-  track to be matched (an Item object) and ``info`` is the TrackInfo object
-  that is proposed as a match. Should return a ``(dist, dist_max)`` pair
-  of floats indicating the distance.
-
-* ``album_distance(self, items, album_info, mapping)``: like the above, but
-  compares a list of items (representing an album) to an album-level MusicBrainz
-  entry. ``items`` is a list of Item objects; ``album_info`` is an AlbumInfo
-  object; and ``mapping`` is a dictionary that maps Items to their corresponding
-  TrackInfo objects.
-
-* ``candidates(self, items, artist, album, va_likely)``: given a list of items
-  comprised by an album to be matched, return a list of ``AlbumInfo`` objects
-  for candidate albums to be compared and matched.
-
-* ``item_candidates(self, item, artist, album)``: given a *singleton* item,
-  return a list of ``TrackInfo`` objects for candidate tracks to be compared and
-  matched.
-
-* ``album_for_id(self, album_id)``: given an ID from user input or an album's
-  tags, return a candidate AlbumInfo object (or None).
-
-* ``track_for_id(self, track_id)``: given an ID from user input or a file's
-  tags, return a candidate TrackInfo object (or None).
-
-When implementing these functions, you may want to use the functions from the
-``beets.autotag`` and ``beets.autotag.mb`` modules, both of which have
-somewhat helpful docstrings.
-
-Read Configuration Options
-^^^^^^^^^^^^^^^^^^^^^^^^^^
-
-Plugins can configure themselves using the ``config.yaml`` file. You can read
-configuration values in two ways. The first is to use `self.config` within
-your plugin class. This gives you a view onto the configuration values in a
-section with the same name as your plugin's module. For example, if your plugin
-is in ``greatplugin.py``, then `self.config` will refer to options under the
-``greatplugin:`` section of the config file.
-
-For example, if you have a configuration value called "foo", then users can put
-this in their ``config.yaml``::
-
-    greatplugin:
-        foo: bar
-
-To access this value, say ``self.config['foo'].get()`` at any point in your
-plugin's code. The `self.config` object is a *view* as defined by the `Confuse`_
-library.
-
-.. _Confuse: https://confuse.readthedocs.io/en/latest/
-
-If you want to access configuration values *outside* of your plugin's section,
-import the `config` object from the `beets` module. That is, just put ``from
-beets import config`` at the top of your plugin and access values from there.
-
-If your plugin provides configuration values for sensitive data (e.g.,
-passwords, API keys, ...), you should add these to the config so they can be
-redacted automatically when users dump their config. This can be done by
-setting each value's `redact` flag, like so::
-
-    self.config['password'].redact = True
-
-
-Add Path Format Functions and Fields
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-
-Beets supports *function calls* in its path format syntax (see
-:doc:`/reference/pathformat`). Beets includes a few built-in functions, but
-plugins can register new functions by adding them to the ``template_funcs``
-dictionary.
-
-Here's an example::
-
-    class MyPlugin(BeetsPlugin):
-        def __init__(self):
-            super().__init__()
-            self.template_funcs['initial'] = _tmpl_initial
-
-    def _tmpl_initial(text: str) -> str:
-        if text:
-            return text[0].upper()
-        else:
-            return u''
-
-This plugin provides a function ``%initial`` to path templates where
-``%initial{$artist}`` expands to the artist's initial (its capitalized first
-character).
-
-Plugins can also add template *fields*, which are computed values referenced
-as ``$name`` in templates. To add a new field, add a function that takes an
-``Item`` object to the ``template_fields`` dictionary on the plugin object.
-Here's an example that adds a ``$disc_and_track`` field::
-
-    class MyPlugin(BeetsPlugin):
-        def __init__(self):
-            super().__init__()
-            self.template_fields['disc_and_track'] = _tmpl_disc_and_track
-
-    def _tmpl_disc_and_track(item: Item) -> str:
-        """Expand to the disc number and track number if this is a
-        multi-disc release. Otherwise, just expands to the track
-        number.
-        """
-        if item.disctotal > 1:
-            return u'%02i.%02i' % (item.disc, item.track)
-        else:
-            return u'%02i' % (item.track)
-
-With this plugin enabled, templates can reference ``$disc_and_track`` as they
-can any standard metadata field.
-
-This field works for *item* templates. Similarly, you can register *album*
-template fields by adding a function accepting an ``Album`` argument to the
-``album_template_fields`` dict.
-
-Extend MediaFile
-^^^^^^^^^^^^^^^^
-
-`MediaFile`_ is the file tag abstraction layer that beets uses to make
-cross-format metadata manipulation simple. Plugins can add fields to MediaFile
-to extend the kinds of metadata that they can easily manage.
-
-The ``MediaFile`` class uses ``MediaField`` descriptors to provide
-access to file tags. If you have created a descriptor you can add it through
-your plugins :py:meth:`beets.plugins.BeetsPlugin.add_media_field()`` method.
-
-.. _MediaFile: https://mediafile.readthedocs.io/en/latest/
-
-
-Here's an example plugin that provides a meaningless new field "foo"::
-
-    class FooPlugin(BeetsPlugin):
-        def __init__(self):
-            field = mediafile.MediaField(
-                mediafile.MP3DescStorageStyle(u'foo'),
-                mediafile.StorageStyle(u'foo')
-            )
-            self.add_media_field('foo', field)
-
-    FooPlugin()
-    item = Item.from_path('/path/to/foo/tag.mp3')
-    assert item['foo'] == 'spam'
-
-    item['foo'] == 'ham'
-    item.write()
-    # The "foo" tag of the file is now "ham"
-
-
-.. _plugin-stage:
-
-Add Import Pipeline Stages
-^^^^^^^^^^^^^^^^^^^^^^^^^^
-
-Many plugins need to add high-latency operations to the import workflow. For
-example, a plugin that fetches lyrics from the Web would, ideally, not block the
-progress of the rest of the importer. Beets allows plugins to add stages to the
-parallel import pipeline.
-
-Each stage is run in its own thread. Plugin stages run after metadata changes
-have been applied to a unit of music (album or track) and before file
-manipulation has occurred (copying and moving files, writing tags to disk).
-Multiple stages run in parallel but each stage processes only one task at a time
-and each task is processed by only one stage at a time.
-
-Plugins provide stages as functions that take two arguments: ``config`` and
-``task``, which are ``ImportSession`` and ``ImportTask`` objects (both defined in
-``beets.importer``). Add such a function to the plugin's ``import_stages`` field
-to register it::
-
-    from beets.plugins import BeetsPlugin
-    class ExamplePlugin(BeetsPlugin):
-        def __init__(self):
-            super().__init__()
-            self.import_stages = [self.stage]
-        def stage(self, session, task):
-            print('Importing something!')
-
-It is also possible to request your function to run early in the pipeline by
-adding the function to the plugin's ``early_import_stages`` field instead::
-
-    self.early_import_stages = [self.stage]
-
-.. _extend-query:
-
-Extend the Query Syntax
-^^^^^^^^^^^^^^^^^^^^^^^
-
-You can add new kinds of queries to beets' :doc:`query syntax
-</reference/query>`. There are two ways to add custom queries: using a prefix
-and using a name. Prefix-based query extension can apply to *any* field, while
-named queries are not associated with any field. For example, beets already
-supports regular expression queries, which are indicated by a colon
-prefix---plugins can do the same.
-
-For either kind of query extension, define a subclass of the ``Query`` type
-from the ``beets.dbcore.query`` module. Then:
-
-- To define a prefix-based query, define a ``queries`` method in your plugin
-  class. Return from this method a dictionary mapping prefix strings to query
-  classes.
-- To define a named query, defined dictionaries named either ``item_queries``
-  or ``album_queries``. These should map names to query types. So if you
-  use ``{ "foo": FooQuery }``, then the query ``foo:bar`` will construct a
-  query like ``FooQuery("bar")``.
-
-For prefix-based queries, you will want to extend ``FieldQuery``, which
-implements string comparisons on fields. To use it, create a subclass
-inheriting from that class and override the ``value_match`` class method.
-(Remember the ``@classmethod`` decorator!) The following example plugin
-declares a query using the ``@`` prefix to delimit exact string matches. The
-plugin will be used if we issue a command like ``beet ls @something`` or
-``beet ls artist:@something``::
-
-    from beets.plugins import BeetsPlugin
-    from beets.dbcore import FieldQuery
-
-    class ExactMatchQuery(FieldQuery):
-        @classmethod
-        def value_match(self, pattern, val):
-            return pattern == val
-
-    class ExactMatchPlugin(BeetsPlugin):
-        def queries(self):
-            return {
-                '@': ExactMatchQuery
-            }
-
-
-Flexible Field Types
-^^^^^^^^^^^^^^^^^^^^
-
-If your plugin uses flexible fields to store numbers or other
-non-string values, you can specify the types of those fields. A rating
-plugin, for example, might want to declare that the ``rating`` field
-should have an integer type::
-
-    from beets.plugins import BeetsPlugin
-    from beets.dbcore import types
-
-    class RatingPlugin(BeetsPlugin):
-        item_types = {'rating': types.INTEGER}
-
-        @property
-        def album_types(self):
-            return {'rating': types.INTEGER}
-
-A plugin may define two attributes: `item_types` and `album_types`.
-Each of those attributes is a dictionary mapping a flexible field name
-to a type instance. You can find the built-in types in the
-`beets.dbcore.types` and `beets.library` modules or implement your own
-type by inheriting from the `Type` class.
-
-Specifying types has several advantages:
-
-* Code that accesses the field like ``item['my_field']`` gets the right
-  type (instead of just a string).
-
-* You can use advanced queries (like :ref:`ranges <numericquery>`)
-  from the command line.
-
-* User input for flexible fields may be validated and converted.
-
-* Items missing the given field can use an appropriate null value for
-  querying and sorting purposes.
-
-
-.. _plugin-logging:
-
-Logging
-^^^^^^^
-
-Each plugin object has a ``_log`` attribute, which is a ``Logger`` from the
-`standard Python logging module`_. The logger is set up to `PEP 3101`_,
-str.format-style string formatting. So you can write logging calls like this::
-
-    self._log.debug(u'Processing {0.title} by {0.artist}', item)
-
-.. _PEP 3101: https://www.python.org/dev/peps/pep-3101/
-.. _standard Python logging module: https://docs.python.org/2/library/logging.html
-
-When beets is in verbose mode, plugin messages are prefixed with the plugin
-name to make them easier to see.
-
-Which messages will be logged depends on the logging level and the action
-performed:
-
-* Inside import stages and event handlers, the default is ``WARNING`` messages
-  and above.
-* Everywhere else, the default is ``INFO`` or above.
-
-The verbosity can be increased with ``--verbose`` (``-v``) flags: each flags
-lowers the level by a notch. That means that, with a single ``-v`` flag, event
-handlers won't have their ``DEBUG`` messages displayed, but command functions
-(for example) will. With ``-vv`` on the command line, ``DEBUG`` messages will
-be displayed everywhere.
-
-This addresses a common pattern where plugins need to use the same code for a
-command and an import stage, but the command needs to print more messages than
-the import stage. (For example, you'll want to log "found lyrics for this song"
-when you're run explicitly as a command, but you don't want to noisily
-interrupt the importer interface when running automatically.)
-
-.. _append_prompt_choices:
-
-Append Prompt Choices
-^^^^^^^^^^^^^^^^^^^^^
-
-Plugins can also append choices to the prompt presented to the user during
-an import session.
-
-To do so, add a listener for the ``before_choose_candidate`` event, and return
-a list of ``PromptChoices`` that represent the additional choices that your
-plugin shall expose to the user::
-
-    from beets.plugins import BeetsPlugin
-    from beets.ui.commands import PromptChoice
-
-    class ExamplePlugin(BeetsPlugin):
-        def __init__(self):
-            super().__init__()
-            self.register_listener('before_choose_candidate',
-                                   self.before_choose_candidate_event)
-
-        def before_choose_candidate_event(self, session, task):
-            return [PromptChoice('p', 'Print foo', self.foo),
-                    PromptChoice('d', 'Do bar', self.bar)]
-
-        def foo(self, session, task):
-            print('User has chosen "Print foo"!')
-
-        def bar(self, session, task):
-            print('User has chosen "Do bar"!')
-
-The previous example modifies the standard prompt::
-
-    # selection (default 1), Skip, Use as-is, as Tracks, Group albums,
-    Enter search, enter Id, aBort?
-
-by appending two additional options (``Print foo`` and ``Do bar``)::
-
-    # selection (default 1), Skip, Use as-is, as Tracks, Group albums,
-    Enter search, enter Id, aBort, Print foo, Do bar?
-
-If the user selects a choice, the ``callback`` attribute of the corresponding
-``PromptChoice`` will be called. It is the responsibility of the plugin to
-check for the status of the import session and decide the choices to be
-appended: for example, if a particular choice should only be presented if the
-album has no candidates, the relevant checks against ``task.candidates`` should
-be performed inside the plugin's ``before_choose_candidate_event`` accordingly.
-
-Please make sure that the short letter for each of the choices provided by the
-plugin is not already in use: the importer will emit a warning and discard
-all but one of the choices using the same letter, giving priority to the
-core importer prompt choices. As a reference, the following characters are used
-by the choices on the core importer prompt, and hence should not be used:
-``a``, ``s``, ``u``, ``t``, ``g``, ``e``, ``i``, ``b``.
-
-Additionally, the callback function can optionally specify the next action to
-be performed by returning a ``importer.Action`` value. It may also return a
-``autotag.Proposal`` value to update the set of current proposals to be
-considered.
diff --git a/docs/dev/plugins/commands.rst b/docs/dev/plugins/commands.rst
new file mode 100644
index 0000000000..5dd5094062
--- /dev/null
+++ b/docs/dev/plugins/commands.rst
@@ -0,0 +1,50 @@
+.. _add_subcommands:
+
+Add Commands to the CLI
+-----------------------
+
+Plugins can add new subcommands to the ``beet`` command-line interface. Define
+the plugin class' ``commands()`` method to return a list of ``Subcommand``
+objects. (The ``Subcommand`` class is defined in the ``beets.ui`` module.)
+Here's an example plugin that adds a simple command:
+
+.. code-block:: python
+
+    from beets.plugins import BeetsPlugin
+    from beets.ui import Subcommand
+
+    my_super_command = Subcommand('super', help='do something super')
+    def say_hi(lib, opts, args):
+        print "Hello everybody! I'm a plugin!"
+    my_super_command.func = say_hi
+
+    class SuperPlug(BeetsPlugin):
+        def commands(self):
+            return [my_super_command]
+
+To make a subcommand, invoke the constructor like so: ``Subcommand(name, parser,
+help, aliases)``. The ``name`` parameter is the only required one and should
+just be the name of your command. ``parser`` can be an `OptionParser instance`_,
+but it defaults to an empty parser (you can extend it later). ``help`` is a
+description of your command, and ``aliases`` is a list of shorthand versions of
+your command name.
+
+.. _OptionParser instance: https://docs.python.org/library/optparse.html
+
+You'll need to add a function to your command by saying ``mycommand.func =
+myfunction``. This function should take the following parameters: ``lib`` (a
+beets ``Library`` object) and ``opts`` and ``args`` (command-line options and
+arguments as returned by `OptionParser.parse_args`_).
+
+.. _OptionParser.parse_args:
+    https://docs.python.org/library/optparse.html#parsing-arguments
+
+The function should use any of the utility functions defined in ``beets.ui``.
+Try running ``pydoc beets.ui`` to see what's available.
+
+You can add command-line options to your new command using the ``parser`` member
+of the ``Subcommand`` class, which is a ``CommonOptionsParser`` instance. Just
+use it like you would a normal ``OptionParser`` in an independent script. Note
+that it offers several methods to add common options: ``--album``, ``--path``
+and ``--format``. This feature is versatile and extensively documented, try
+``pydoc beets.ui.CommonOptionsParser`` for more information.
\ No newline at end of file
diff --git a/docs/dev/plugins/events.rst b/docs/dev/plugins/events.rst
new file mode 100644
index 0000000000..a52a26d9d3
--- /dev/null
+++ b/docs/dev/plugins/events.rst
@@ -0,0 +1,175 @@
+.. _plugin_events:
+
+Listen for Events
+-----------------
+
+.. currentmodule:: beets.plugins
+
+Event handlers allow plugins to hook into whenever something happens in beets'
+operations. For instance, a plugin could write a log message every time an album
+is successfully autotagged or update MPD's index whenever the database is
+changed.
+
+You can "listen" for events using :py:meth:`BeetsPlugin.register_listener`. Here's
+an example:
+
+.. code-block:: python
+
+    from beets.plugins import BeetsPlugin
+
+    def loaded():
+        print('Plugin loaded!')
+
+    class SomePlugin(BeetsPlugin):
+        def __init__(self):
+            super().__init__()
+            self.register_listener('pluginload', loaded)
+
+Note that if you want to access an attribute of your plugin (e.g. ``config`` or
+``log``) you'll have to define a method and not a function. Here is the usual
+registration process in this case:
+
+.. code-block:: python
+
+    from beets.plugins import BeetsPlugin
+
+    class SomePlugin(BeetsPlugin):
+      def __init__(self):
+        super().__init__()
+        self.register_listener('pluginload', self.loaded)
+
+      def loaded(self):
+        self._log.info('Plugin loaded!')
+
+The events currently available are:
+
+* `pluginload`: called after all the plugins have been loaded after the ``beet``
+  command starts
+
+* `import`: called after a ``beet import`` command finishes (the ``lib`` keyword
+  argument is a Library object; ``paths`` is a list of paths (strings) that were
+  imported)
+
+* `album_imported`: called with an ``Album`` object every time the ``import``
+  command finishes adding an album to the library. Parameters: ``lib``,
+  ``album``
+
+* `album_removed`: called with an ``Album`` object every time an album is
+  removed from the library (even when its file is not deleted from disk).
+
+* `item_copied`: called with an ``Item`` object whenever its file is copied.
+  Parameters: ``item``, ``source`` path, ``destination`` path
+
+* `item_imported`: called with an ``Item`` object every time the importer adds a
+  singleton to the library (not called for full-album imports). Parameters:
+  ``lib``, ``item``
+
+* `before_item_moved`: called with an ``Item`` object immediately before its
+  file is moved. Parameters: ``item``, ``source`` path, ``destination`` path
+
+* `item_moved`: called with an ``Item`` object whenever its file is moved.
+  Parameters: ``item``, ``source`` path, ``destination`` path
+
+* `item_linked`: called with an ``Item`` object whenever a symlink is created
+  for a file.
+  Parameters: ``item``, ``source`` path, ``destination`` path
+
+* `item_hardlinked`: called with an ``Item`` object whenever a hardlink is
+  created for a file.
+  Parameters: ``item``, ``source`` path, ``destination`` path
+
+* `item_reflinked`: called with an ``Item`` object whenever a reflink is
+  created for a file.
+  Parameters: ``item``, ``source`` path, ``destination`` path
+
+* `item_removed`: called with an ``Item`` object every time an item (singleton
+  or album's part) is removed from the library (even when its file is not
+  deleted from disk).
+
+* `write`: called with an ``Item`` object, a ``path``, and a ``tags``
+  dictionary just before a file's metadata is written to disk (i.e.,
+  just before the file on disk is opened). Event handlers may change
+  the ``tags`` dictionary to customize the tags that are written to the
+  media file. Event handlers may also raise a
+  ``library.FileOperationError`` exception to abort the write
+  operation. Beets will catch that exception, print an error message
+  and continue.
+
+* `after_write`: called with an ``Item`` object after a file's metadata is
+  written to disk (i.e., just after the file on disk is closed).
+
+* `import_task_created`: called immediately after an import task is
+  initialized. Plugins can use this to, for example, change imported files of a
+  task before anything else happens. It's also possible to replace the task
+  with another task by returning a list of tasks. This list can contain zero
+  or more `ImportTask`s. Returning an empty list will stop the task.
+  Parameters: ``task`` (an `ImportTask`) and ``session`` (an `ImportSession`).
+
+* `import_task_start`: called when before an import task begins processing.
+  Parameters: ``task`` and ``session``.
+
+* `import_task_apply`: called after metadata changes have been applied in an
+  import task. This is called on the same thread as the UI, so use this
+  sparingly and only for tasks that can be done quickly. For most plugins, an
+  import pipeline stage is a better choice (see :ref:`plugin-stage`).
+  Parameters: ``task`` and ``session``.
+
+* `import_task_before_choice`: called after candidate search for an import task
+  before any decision is made about how/if to import or tag. Can be used to
+  present information about the task or initiate interaction with the user
+  before importing occurs. Return an importer action to take a specific action.
+  Only one handler may return a non-None result.
+  Parameters: ``task`` and ``session``
+
+* `import_task_choice`: called after a decision has been made about an import
+  task. This event can be used to initiate further interaction with the user.
+  Use ``task.choice_flag`` to determine or change the action to be
+  taken. Parameters: ``task`` and ``session``.
+
+* `import_task_files`: called after an import task finishes manipulating the
+  filesystem (copying and moving files, writing metadata tags). Parameters:
+  ``task`` and ``session``.
+
+* `library_opened`: called after beets starts up and initializes the main
+  Library object. Parameter: ``lib``.
+
+* `database_change`: a modification has been made to the library database. The
+  change might not be committed yet. Parameters: ``lib`` and ``model``.
+
+* `cli_exit`: called just before the ``beet`` command-line program exits.
+  Parameter: ``lib``.
+
+* `import_begin`: called just before a ``beet import`` session starts up.
+  Parameter: ``session``.
+
+* `trackinfo_received`: called after metadata for a track item has been
+  fetched from a data source, such as MusicBrainz. You can modify the tags
+  that the rest of the pipeline sees on a ``beet import`` operation or during
+  later adjustments, such as ``mbsync``. Slow handlers of the event can impact
+  the operation, since the event is fired for any fetched possible match
+  `before` the user (or the autotagger machinery) gets to see the match.
+  Parameter: ``info``.
+
+* `albuminfo_received`: like `trackinfo_received`, the event indicates new
+  metadata for album items. The parameter is an ``AlbumInfo`` object instead
+  of a ``TrackInfo``.
+  Parameter: ``info``.
+
+* `before_choose_candidate`: called before the user is prompted for a decision
+  during a ``beet import`` interactive session. Plugins can use this event for
+  :ref:`appending choices to the prompt <append_prompt_choices>` by returning a
+  list of ``PromptChoices``. Parameters: ``task`` and ``session``.
+
+* `mb_track_extract`: called after the metadata is obtained from
+  MusicBrainz. The parameter is a ``dict`` containing the tags retrieved from
+  MusicBrainz for a track. Plugins must return a new (potentially empty)
+  ``dict`` with additional ``field: value`` pairs, which the autotagger will
+  apply to the item, as flexible attributes if ``field`` is not a hardcoded
+  field. Fields already present on the track are overwritten.
+  Parameter: ``data``
+
+* `mb_album_extract`: Like `mb_track_extract`, but for album tags. Overwrites
+  tags set at the track level, if they have the same ``field``.
+  Parameter: ``data``
+
+The included ``mpdupdate`` plugin provides an example use case for event listeners.
\ No newline at end of file
diff --git a/docs/dev/plugins/index.rst b/docs/dev/plugins/index.rst
new file mode 100644
index 0000000000..8c311203f7
--- /dev/null
+++ b/docs/dev/plugins/index.rst
@@ -0,0 +1,92 @@
+Plugin Development
+==================
+
+Beets plugins are Python modules or packages that extend the core functionality
+of beets. The plugin system is designed to be flexible, allowing developers to
+add virtually any type of features to beets. 
+
+For instance you can create plugins that add new commands to the command-line interface,
+listen for events in the beets lifecycle or extend the autotagger with new metadata sources.
+
+
+.. _writing-plugins:
+
+Basic Plugin Setup
+------------------
+
+A beets plugin is just a Python module or package inside the ``beetsplug``
+namespace [#namespace]_ package. To create the basic plugin layout,
+create a directory called ``beetsplug`` and add either your plugin module:
+
+.. code-block:: shell
+
+    beetsplug/
+    └── myawesomeplugin.py
+
+
+or your plugin subpackage
+
+.. code-block:: shell
+
+    beetsplug/
+    └── myawesomeplugin/
+        ├── __init__.py
+        └── myawesomeplugin.py
+
+.. attention::
+
+    You do not need to add an ``__init__.py`` file to the ``beetsplug``
+    directory. Python treats your plugin as a namespace package automatically,
+    thus we do not depend on ``pkgutil``-based setup in the ``__init__.py``
+    file anymore.
+
+The meat of your plugin goes in ``myawesomeplugin.py``. Every plugin has to extend the
+:class:`beets.plugins.BeetsPlugin` abstract base class [#baseclass]_. For instance,
+a minimal plugin without any functionality would look like this:
+
+.. code-block:: python
+
+    # beetsplug/myawesomeplugin.py
+    from beets.plugins import BeetsPlugin
+
+    class MyAwesomePlugin(BeetsPlugin):
+        pass
+
+
+To use your new plugin, you need to package [#packaging]_ your plugin and install it into your ``beets`` (virtual) environment. To enable your plugin, add it it to the beets configuration
+
+.. code-block:: yaml
+
+    # config.yaml
+    plugins:
+      - myawesomeplugin
+
+and you're good to go! 
+
+
+.. [#namespace] Check out `this article`_ and `this Stack Overflow question`_ if you haven't heard about namespace packages.
+.. [#baseclass] Abstract base classes allow us to define a contract which any plugin must follow. This is a common paradigm in object-oriented programming, and it helps to ensure that plugins are implemented in a consistent way. For more information, see for example `pep-3119`_.
+.. [#packaging] There are a variety of packaging tools available for python, for example you can use `poetry`_, `setuptools`_ or `hatchling`_.
+
+
+.. _this article: https://realpython.com/python-namespace-package/#setting-up-some-namespace-packages
+.. _this Stack Overflow question: https://stackoverflow.com/a/27586272/9582674
+.. _poetry: https://python-poetry.org/docs/pyproject/#packages
+.. _setuptools: https://setuptools.pypa.io/en/latest/userguide/package_discovery.html#finding-simple-packages
+.. _hatchling: https://hatch.pypa.io/latest/config/build/#build-system
+.. _pep-3119: https://peps.python.org/pep-3119/#rationale
+
+
+Further Reading
+---------------
+
+For more information on writing plugins, feel free to check out 
+the following resources:
+
+.. toctree::
+    :maxdepth: 2
+
+    commands
+    events
+    other
+
diff --git a/docs/dev/plugins/other.rst b/docs/dev/plugins/other.rst
new file mode 100644
index 0000000000..3d552d5032
--- /dev/null
+++ b/docs/dev/plugins/other.rst
@@ -0,0 +1,346 @@
+Extending the Autotagger
+------------------------
+
+Plugins can also be used to extend the autotagger and allow metadata lookup
+from additional sources. For this your plugin has to extend the :class:`beets.metadata_plugin.MetadataSourcePlugin` class and implement all abstract methods.
+
+.. currentmodule:: beets.metadata_plugin
+
+On metadata lookup, the autotagger will first call the :py:meth:`MetadataSourcePlugin.candidates` (or :py:meth:`MetadataSourcePlugin.item_candidates` ) method of all enabled MetadataSourcePlugins to get a list of available candidates. Than we will rank all candidates by using each plugins :py:meth:`MetadataSourcePlugin.track_distance` and :py:meth:`MetadataSourcePlugin.album_distance` methods. 
+
+Please have a look at the ``beets.autotag`` and especially the ``beets.metadata_plugin`` modules for more information. Additionally, for a  comprehensive example, see the ``musicbrainz`` or ``chroma`` plugins, which are included with beets.
+
+
+
+Read Configuration Options
+--------------------------
+
+Plugins can configure themselves using the ``config.yaml`` file. You can read
+configuration values in two ways. The first is to use `self.config` within
+your plugin class. This gives you a view onto the configuration values in a
+section with the same name as your plugin's module. For example, if your plugin
+is in ``greatplugin.py``, then `self.config` will refer to options under the
+``greatplugin:`` section of the config file.
+
+For example, if you have a configuration value called "foo", then users can put
+this in their ``config.yaml``::
+
+    greatplugin:
+        foo: bar
+
+To access this value, say ``self.config['foo'].get()`` at any point in your
+plugin's code. The `self.config` object is a *view* as defined by the `Confuse`_
+library.
+
+.. _Confuse: https://confuse.readthedocs.io/en/latest/
+
+If you want to access configuration values *outside* of your plugin's section,
+import the `config` object from the `beets` module. That is, just put ``from
+beets import config`` at the top of your plugin and access values from there.
+
+If your plugin provides configuration values for sensitive data (e.g.,
+passwords, API keys, ...), you should add these to the config so they can be
+redacted automatically when users dump their config. This can be done by
+setting each value's `redact` flag, like so::
+
+    self.config['password'].redact = True
+
+
+Add Path Format Functions and Fields
+------------------------------------
+
+Beets supports *function calls* in its path format syntax (see
+:doc:`/reference/pathformat`). Beets includes a few built-in functions, but
+plugins can register new functions by adding them to the ``template_funcs``
+dictionary.
+
+Here's an example::
+
+    class MyPlugin(BeetsPlugin):
+        def __init__(self):
+            super().__init__()
+            self.template_funcs['initial'] = _tmpl_initial
+
+    def _tmpl_initial(text: str) -> str:
+        if text:
+            return text[0].upper()
+        else:
+            return u''
+
+This plugin provides a function ``%initial`` to path templates where
+``%initial{$artist}`` expands to the artist's initial (its capitalized first
+character).
+
+Plugins can also add template *fields*, which are computed values referenced
+as ``$name`` in templates. To add a new field, add a function that takes an
+``Item`` object to the ``template_fields`` dictionary on the plugin object.
+Here's an example that adds a ``$disc_and_track`` field::
+
+    class MyPlugin(BeetsPlugin):
+        def __init__(self):
+            super().__init__()
+            self.template_fields['disc_and_track'] = _tmpl_disc_and_track
+
+    def _tmpl_disc_and_track(item: Item) -> str:
+        """Expand to the disc number and track number if this is a
+        multi-disc release. Otherwise, just expands to the track
+        number.
+        """
+        if item.disctotal > 1:
+            return u'%02i.%02i' % (item.disc, item.track)
+        else:
+            return u'%02i' % (item.track)
+
+With this plugin enabled, templates can reference ``$disc_and_track`` as they
+can any standard metadata field.
+
+This field works for *item* templates. Similarly, you can register *album*
+template fields by adding a function accepting an ``Album`` argument to the
+``album_template_fields`` dict.
+
+Extend MediaFile
+----------------
+
+`MediaFile`_ is the file tag abstraction layer that beets uses to make
+cross-format metadata manipulation simple. Plugins can add fields to MediaFile
+to extend the kinds of metadata that they can easily manage.
+
+The ``MediaFile`` class uses ``MediaField`` descriptors to provide
+access to file tags. If you have created a descriptor you can add it through
+your plugins :py:meth:`beets.plugins.BeetsPlugin.add_media_field()`` method.
+
+.. _MediaFile: https://mediafile.readthedocs.io/en/latest/
+
+
+Here's an example plugin that provides a meaningless new field "foo"::
+
+    class FooPlugin(BeetsPlugin):
+        def __init__(self):
+            field = mediafile.MediaField(
+                mediafile.MP3DescStorageStyle(u'foo'),
+                mediafile.StorageStyle(u'foo')
+            )
+            self.add_media_field('foo', field)
+
+    FooPlugin()
+    item = Item.from_path('/path/to/foo/tag.mp3')
+    assert item['foo'] == 'spam'
+
+    item['foo'] == 'ham'
+    item.write()
+    # The "foo" tag of the file is now "ham"
+
+
+.. _plugin-stage:
+
+Add Import Pipeline Stages
+--------------------------
+
+Many plugins need to add high-latency operations to the import workflow. For
+example, a plugin that fetches lyrics from the Web would, ideally, not block the
+progress of the rest of the importer. Beets allows plugins to add stages to the
+parallel import pipeline.
+
+Each stage is run in its own thread. Plugin stages run after metadata changes
+have been applied to a unit of music (album or track) and before file
+manipulation has occurred (copying and moving files, writing tags to disk).
+Multiple stages run in parallel but each stage processes only one task at a time
+and each task is processed by only one stage at a time.
+
+Plugins provide stages as functions that take two arguments: ``config`` and
+``task``, which are ``ImportSession`` and ``ImportTask`` objects (both defined in
+``beets.importer``). Add such a function to the plugin's ``import_stages`` field
+to register it::
+
+    from beets.plugins import BeetsPlugin
+    class ExamplePlugin(BeetsPlugin):
+        def __init__(self):
+            super().__init__()
+            self.import_stages = [self.stage]
+        def stage(self, session, task):
+            print('Importing something!')
+
+It is also possible to request your function to run early in the pipeline by
+adding the function to the plugin's ``early_import_stages`` field instead::
+
+    self.early_import_stages = [self.stage]
+
+.. _extend-query:
+
+Extend the Query Syntax
+^^^^^^^^^^^^^^^^^^^^^^^
+
+You can add new kinds of queries to beets' :doc:`query syntax
+</reference/query>`. There are two ways to add custom queries: using a prefix
+and using a name. Prefix-based query extension can apply to *any* field, while
+named queries are not associated with any field. For example, beets already
+supports regular expression queries, which are indicated by a colon
+prefix---plugins can do the same.
+
+For either kind of query extension, define a subclass of the ``Query`` type
+from the ``beets.dbcore.query`` module. Then:
+
+- To define a prefix-based query, define a ``queries`` method in your plugin
+  class. Return from this method a dictionary mapping prefix strings to query
+  classes.
+- To define a named query, defined dictionaries named either ``item_queries``
+  or ``album_queries``. These should map names to query types. So if you
+  use ``{ "foo": FooQuery }``, then the query ``foo:bar`` will construct a
+  query like ``FooQuery("bar")``.
+
+For prefix-based queries, you will want to extend ``FieldQuery``, which
+implements string comparisons on fields. To use it, create a subclass
+inheriting from that class and override the ``value_match`` class method.
+(Remember the ``@classmethod`` decorator!) The following example plugin
+declares a query using the ``@`` prefix to delimit exact string matches. The
+plugin will be used if we issue a command like ``beet ls @something`` or
+``beet ls artist:@something``::
+
+    from beets.plugins import BeetsPlugin
+    from beets.dbcore import FieldQuery
+
+    class ExactMatchQuery(FieldQuery):
+        @classmethod
+        def value_match(self, pattern, val):
+            return pattern == val
+
+    class ExactMatchPlugin(BeetsPlugin):
+        def queries(self):
+            return {
+                '@': ExactMatchQuery
+            }
+
+
+Flexible Field Types
+--------------------
+
+If your plugin uses flexible fields to store numbers or other
+non-string values, you can specify the types of those fields. A rating
+plugin, for example, might want to declare that the ``rating`` field
+should have an integer type::
+
+    from beets.plugins import BeetsPlugin
+    from beets.dbcore import types
+
+    class RatingPlugin(BeetsPlugin):
+        item_types = {'rating': types.INTEGER}
+
+        @property
+        def album_types(self):
+            return {'rating': types.INTEGER}
+
+A plugin may define two attributes: `item_types` and `album_types`.
+Each of those attributes is a dictionary mapping a flexible field name
+to a type instance. You can find the built-in types in the
+`beets.dbcore.types` and `beets.library` modules or implement your own
+type by inheriting from the `Type` class.
+
+Specifying types has several advantages:
+
+* Code that accesses the field like ``item['my_field']`` gets the right
+  type (instead of just a string).
+
+* You can use advanced queries (like :ref:`ranges <numericquery>`)
+  from the command line.
+
+* User input for flexible fields may be validated and converted.
+
+* Items missing the given field can use an appropriate null value for
+  querying and sorting purposes.
+
+
+.. _plugin-logging:
+
+Logging
+-------
+
+Each plugin object has a ``_log`` attribute, which is a ``Logger`` from the
+`standard Python logging module`_. The logger is set up to `PEP 3101`_,
+str.format-style string formatting. So you can write logging calls like this::
+
+    self._log.debug(u'Processing {0.title} by {0.artist}', item)
+
+.. _PEP 3101: https://www.python.org/dev/peps/pep-3101/
+.. _standard Python logging module: https://docs.python.org/2/library/logging.html
+
+When beets is in verbose mode, plugin messages are prefixed with the plugin
+name to make them easier to see.
+
+Which messages will be logged depends on the logging level and the action
+performed:
+
+* Inside import stages and event handlers, the default is ``WARNING`` messages
+  and above.
+* Everywhere else, the default is ``INFO`` or above.
+
+The verbosity can be increased with ``--verbose`` (``-v``) flags: each flags
+lowers the level by a notch. That means that, with a single ``-v`` flag, event
+handlers won't have their ``DEBUG`` messages displayed, but command functions
+(for example) will. With ``-vv`` on the command line, ``DEBUG`` messages will
+be displayed everywhere.
+
+This addresses a common pattern where plugins need to use the same code for a
+command and an import stage, but the command needs to print more messages than
+the import stage. (For example, you'll want to log "found lyrics for this song"
+when you're run explicitly as a command, but you don't want to noisily
+interrupt the importer interface when running automatically.)
+
+.. _append_prompt_choices:
+
+Append Prompt Choices
+---------------------
+
+Plugins can also append choices to the prompt presented to the user during
+an import session.
+
+To do so, add a listener for the ``before_choose_candidate`` event, and return
+a list of ``PromptChoices`` that represent the additional choices that your
+plugin shall expose to the user::
+
+    from beets.plugins import BeetsPlugin
+    from beets.ui.commands import PromptChoice
+
+    class ExamplePlugin(BeetsPlugin):
+        def __init__(self):
+            super().__init__()
+            self.register_listener('before_choose_candidate',
+                                   self.before_choose_candidate_event)
+
+        def before_choose_candidate_event(self, session, task):
+            return [PromptChoice('p', 'Print foo', self.foo),
+                    PromptChoice('d', 'Do bar', self.bar)]
+
+        def foo(self, session, task):
+            print('User has chosen "Print foo"!')
+
+        def bar(self, session, task):
+            print('User has chosen "Do bar"!')
+
+The previous example modifies the standard prompt::
+
+    # selection (default 1), Skip, Use as-is, as Tracks, Group albums,
+    Enter search, enter Id, aBort?
+
+by appending two additional options (``Print foo`` and ``Do bar``)::
+
+    # selection (default 1), Skip, Use as-is, as Tracks, Group albums,
+    Enter search, enter Id, aBort, Print foo, Do bar?
+
+If the user selects a choice, the ``callback`` attribute of the corresponding
+``PromptChoice`` will be called. It is the responsibility of the plugin to
+check for the status of the import session and decide the choices to be
+appended: for example, if a particular choice should only be presented if the
+album has no candidates, the relevant checks against ``task.candidates`` should
+be performed inside the plugin's ``before_choose_candidate_event`` accordingly.
+
+Please make sure that the short letter for each of the choices provided by the
+plugin is not already in use: the importer will emit a warning and discard
+all but one of the choices using the same letter, giving priority to the
+core importer prompt choices. As a reference, the following characters are used
+by the choices on the core importer prompt, and hence should not be used:
+``a``, ``s``, ``u``, ``t``, ``g``, ``e``, ``i``, ``b``.
+
+Additionally, the callback function can optionally specify the next action to
+be performed by returning a ``importer.Action`` value. It may also return a
+``autotag.Proposal`` value to update the set of current proposals to be
+considered.
\ No newline at end of file
diff --git a/docs/plugins/index.rst b/docs/plugins/index.rst
index 82fa942818..954706c3a1 100644
--- a/docs/plugins/index.rst
+++ b/docs/plugins/index.rst
@@ -5,7 +5,7 @@ Plugins extend beets' core functionality. They add new commands, fetch
 additional data during import, provide new metadata sources, and much more. If
 beets by itself doesn't do what you want it to, you may just need to enable a
 plugin---or, if you want to do something new, :doc:`writing a plugin
-</dev/plugins>` is easy if you know a little Python.
+</dev/plugins/index>` is easy if you know a little Python.
 
 .. _using-plugins:
 
diff --git a/setup.cfg b/setup.cfg
index e3472b04c8..e999b55d33 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -49,3 +49,8 @@ disallow_untyped_decorators = true
 disallow_any_generics = true
 check_untyped_defs = true
 allow_redefinition = true
+
+[[mypy-beets.metadata_plugins]]
+disallow_untyped_decorators = true
+disallow_any_generics = true
+check_untyped_defs = true
diff --git a/test/plugins/test_mbsync.py b/test/plugins/test_mbsync.py
index 088165ef5d..bb88e5e631 100644
--- a/test/plugins/test_mbsync.py
+++ b/test/plugins/test_mbsync.py
@@ -23,7 +23,7 @@ class MbsyncCliTest(PluginTestCase):
     plugin = "mbsync"
 
     @patch(
-        "beets.plugins.album_for_id",
+        "beets.metadata_plugins.album_for_id",
         Mock(
             side_effect=lambda *_: AlbumInfo(
                 album_id="album id",
@@ -33,7 +33,7 @@ class MbsyncCliTest(PluginTestCase):
         ),
     )
     @patch(
-        "beets.plugins.track_for_id",
+        "beets.metadata_plugins.track_for_id",
         Mock(
             side_effect=lambda *_: TrackInfo(
                 track_id="singleton id", title="new title"
diff --git a/test/test_importer.py b/test/test_importer.py
index 9bb0e8a632..cac3f6882b 100644
--- a/test/test_importer.py
+++ b/test/test_importer.py
@@ -1031,7 +1031,9 @@ def album_candidates_mock(*args, **kwargs):
     )
 
 
-@patch("beets.plugins.candidates", Mock(side_effect=album_candidates_mock))
+@patch(
+    "beets.metadata_plugins.candidates", Mock(side_effect=album_candidates_mock)
+)
 class ImportDuplicateAlbumTest(PluginMixin, ImportTestCase):
     plugin = "musicbrainz"
 
@@ -1149,7 +1151,10 @@ def item_candidates_mock(*args, **kwargs):
     )
 
 
-@patch("beets.plugins.item_candidates", Mock(side_effect=item_candidates_mock))
+@patch(
+    "beets.metadata_plugins.item_candidates",
+    Mock(side_effect=item_candidates_mock),
+)
 class ImportDuplicateSingletonTest(ImportTestCase):
     def setUp(self):
         super().setUp()
@@ -1687,8 +1692,14 @@ def mocked_get_track_by_id(id_):
     )
 
 
-@patch("beets.plugins.track_for_id", Mock(side_effect=mocked_get_track_by_id))
-@patch("beets.plugins.album_for_id", Mock(side_effect=mocked_get_album_by_id))
+@patch(
+    "beets.metadata_plugins.track_for_id",
+    Mock(side_effect=mocked_get_track_by_id),
+)
+@patch(
+    "beets.metadata_plugins.album_for_id",
+    Mock(side_effect=mocked_get_album_by_id),
+)
 class ImportIdTest(ImportTestCase):
     ID_RELEASE_0 = "00000000-0000-0000-0000-000000000000"
     ID_RELEASE_1 = "11111111-1111-1111-1111-111111111111"
diff --git a/test/test_ui.py b/test/test_ui.py
index 8bb0218d5e..229a45bbad 100644
--- a/test/test_ui.py
+++ b/test/test_ui.py
@@ -1059,7 +1059,9 @@ def test_cli_config_file_loads_plugin_commands(self):
             file.write("plugins: test")
 
         self.run_command("--config", self.cli_config_path, "plugin", lib=None)
-        assert plugins.find_plugins()[0].is_test_plugin
+        plugs = plugins.find_plugins()
+        assert len(plugs) == 1
+        assert plugs[0].is_test_plugin
         self.unload_plugins()
 
     def test_beetsdir_config(self):