[mcp] Add a basic mcp service

pyproject.toml

@@ -9,7 +9,7 @@ python = ">=3.11,<3.15"
 Django = "^4.0.3"
 django-extensions = "^3.1.5"
 python-dateutil = "^2.8.2"
-python-dotenv = "^0.20.0"
+python-dotenv = ">=0.20.0,<2"
 python-json-logger = "^2.0.2"
 colorlog = "^6.6.0"
 httpx = "<=0.27.2"
@@ -43,6 +43,7 @@ ipython = "^8.14.0"
 pendulum = "^3"
 trafilatura = "^1.6.3"
 django-imagekit = "^5.0.0"
+django-mcp-server = "^0.5.7"
 thefuzz = "^0.22.1"
 dataclass-wizard = "^0.35.0"
 webdavclient3 = "^3.14.6"

vrobbler/apps/scrobbles/mcp.py

@@ -0,0 +1,518 @@
+from mcp_server import MCPToolset
+from scrobbles.models import Scrobble
+from scrobbles.constants import LONG_PLAY_MEDIA
+
+
+class ScrobbleToolset(MCPToolset):
+    def list_recent_scrobbles(
+        self,
+        days: int = 7,
+        media_type: str | None = None,
+        limit: int = 50,
+    ) -> list[dict]:
+        """List scrobbles from the last N days, optionally filtered by media type.
+        Valid media_type values: Video, Track, PodcastEpisode, SportEvent, Book,
+        Paper, VideoGame, BoardGame, GeoLocation, Trail, Beer, Puzzle, Food, Task,
+        WebPage, LifeEvent, Mood, BrickSet, Channel, BirdingLocation, DiscGolfCourse
+        """
+        qs = (
+            Scrobble.objects.filter(user=self.request.user)
+            .select_related(
+                "video", "track", "book", "video_game", "board_game",
+                "beer", "puzzle", "food", "trail", "task", "web_page",
+                "life_event", "mood", "brick_set", "podcast_episode",
+                "sport_event", "geo_location", "birding_location",
+                "disc_golf_course", "channel",
+            )
+            .order_by("-timestamp")
+        )
+        from django.utils import timezone
+        import datetime
+        qs = qs.filter(timestamp__gte=timezone.now() - datetime.timedelta(days=days))
+        if media_type:
+            qs = qs.filter(media_type=media_type)
+        qs = qs[:limit]
+        return [_scrobble_to_dict(s) for s in qs]
+
+    def get_scrobble(self, uuid: str) -> dict | None:
+        """Get a single scrobble by its UUID."""
+        try:
+            s = Scrobble.objects.filter(user=self.request.user).get(uuid=uuid)
+        except Scrobble.DoesNotExist:
+            return None
+        return _scrobble_to_dict(s)
+
+    def search_scrobbles(
+        self, query: str, media_type: str | None = None, limit: int = 20
+    ) -> list[dict]:
+        """Search scrobbles by text in their log data or related media titles."""
+        from django.db.models import Q
+        qs = Scrobble.objects.filter(user=self.request.user).order_by("-timestamp")
+        if media_type:
+            qs = qs.filter(media_type=media_type)
+        qs = qs.filter(
+            Q(log__icontains=query)
+            | Q(video__title__icontains=query)
+            | Q(track__title__icontains=query)
+            | Q(book__title__icontains=query)
+            | Q(video_game__title__icontains=query)
+            | Q(board_game__title__icontains=query)
+            | Q(beer__title__icontains=query)
+            | Q(food__title__icontains=query)
+            | Q(trail__title__icontains=query)
+            | Q(task__title__icontains=query)
+            | Q(web_page__title__icontains=query)
+            | Q(life_event__title__icontains=query)
+            | Q(puzzle__title__icontains=query)
+            | Q(brick_set__title__icontains=query)
+            | Q(podcast_episode__title__icontains=query)
+        )[:limit]
+        return [_scrobble_to_dict(s) for s in qs]
+
+    def get_scrobbles_by_date(
+        self, date: str, media_type: str | None = None
+    ) -> list[dict]:
+        """Get scrobbles for a specific date (YYYY-MM-DD format)."""
+        import datetime
+        try:
+            dt = datetime.datetime.strptime(date, "%Y-%m-%d").date()
+        except ValueError:
+            return []
+        qs = Scrobble.objects.filter(
+            user=self.request.user,
+            timestamp__date=dt,
+        ).order_by("-timestamp")
+        if media_type:
+            qs = qs.filter(media_type=media_type)
+        return [_scrobble_to_dict(s) for s in qs]
+
+    def get_in_progress_scrobbles(
+        self, media_type: str | None = None
+    ) -> list[dict]:
+        """Get scrobbles currently in progress (started but not finished).
+        These are long-play items like books, video games, brick sets, or tasks."""
+        qs = Scrobble.objects.filter(
+            user=self.request.user,
+            in_progress=True,
+        ).order_by("-timestamp")
+        if media_type:
+            qs = qs.filter(media_type=media_type)
+        return [_scrobble_to_dict(s) for s in qs]
+
+    def get_long_play_scrobbles(
+        self, status: str = "in_progress", media_type: str | None = None
+    ) -> list[dict]:
+        """Get long-play scrobbles (books, video games, brick sets, tasks).
+        Status can be 'in_progress' or 'completed'."""
+        types = list(LONG_PLAY_MEDIA.values())
+        qs = Scrobble.objects.filter(
+            user=self.request.user,
+            media_type__in=types,
+        ).order_by("-timestamp")
+        if media_type:
+            qs = qs.filter(media_type=media_type)
+        if status == "in_progress":
+            qs = qs.filter(in_progress=True)
+        elif status == "completed":
+            qs = qs.filter(in_progress=False)
+        return [_scrobble_to_dict(s) for s in qs]
+
+
+class MediaToolset(MCPToolset):
+    def get_book(self, uuid: str) -> dict | None:
+        """Get a book by UUID."""
+        from books.models import Book
+        try:
+            b = Book.objects.get(uuid=uuid)
+        except Book.DoesNotExist:
+            return None
+        return _media_to_dict(b, fields=["title", "pages", "language",
+                                          "first_publish_year", "isbn_13",
+                                          "publisher", "summary"])
+
+    def list_books(self, author: str | None = None, limit: int = 20) -> list[dict]:
+        """List books, optionally filtered by author name."""
+        from books.models import Book
+        qs = Book.objects.all().order_by("title")
+        if author:
+            qs = qs.filter(authors__name__icontains=author)
+        return [_media_to_dict(b, fields=["title", "pages", "language",
+                                           "first_publish_year", "isbn_13",
+                                           "publisher"]) for b in qs[:limit]]
+
+    def get_track(self, uuid: str) -> dict | None:
+        """Get a music track by UUID."""
+        from music.models import Track
+        try:
+            t = Track.objects.select_related("artist_fk").get(uuid=uuid)
+        except Track.DoesNotExist:
+            return None
+        return _media_to_dict(t, fields=["title", "base_run_time_seconds",
+                                          "artist_fk__name", "genre"])
+
+    def list_tracks(self, artist: str | None = None, limit: int = 20) -> list[dict]:
+        """List music tracks, optionally filtered by artist name."""
+        from music.models import Track
+        qs = Track.objects.select_related("artist_fk").all().order_by("title")
+        if artist:
+            qs = qs.filter(artist_fk__name__icontains=artist)
+        return [_media_to_dict(t, fields=["title", "base_run_time_seconds",
+                                           "artist_fk__name", "genre"])
+                for t in qs[:limit]]
+
+    def get_video(self, uuid: str) -> dict | None:
+        """Get a video by UUID."""
+        from videos.models import Video
+        try:
+            v = Video.objects.select_related("tv_series", "channel").get(uuid=uuid)
+        except Video.DoesNotExist:
+            return None
+        return _media_to_dict(v, fields=["title", "year", "overview",
+                                          "imdb_id", "imdb_rating",
+                                          "tv_series__name", "channel__title",
+                                          "season_number", "episode_number"])
+
+    def list_videos(self, series: str | None = None, limit: int = 20) -> list[dict]:
+        """List videos, optionally filtered by series name."""
+        from videos.models import Video
+        qs = Video.objects.select_related("tv_series", "channel").all().order_by("title")
+        if series:
+            qs = qs.filter(tv_series__name__icontains=series)
+        return [_media_to_dict(v, fields=["title", "year", "overview",
+                                           "tv_series__name", "channel__title",
+                                           "season_number", "episode_number"])
+                for v in qs[:limit]]
+
+    def get_board_game(self, uuid: str) -> dict | None:
+        """Get a board game by UUID."""
+        from boardgames.models import BoardGame
+        try:
+            bg = BoardGame.objects.get(uuid=uuid)
+        except BoardGame.DoesNotExist:
+            return None
+        return _media_to_dict(bg, fields=["title", "genre"])
+
+    def list_board_games(self, limit: int = 20) -> list[dict]:
+        """List board games."""
+        from boardgames.models import BoardGame
+        qs = BoardGame.objects.all().order_by("title")[:limit]
+        return [_media_to_dict(bg, fields=["title", "genre"]) for bg in qs]
+
+    def get_podcast_episode(self, uuid: str) -> dict | None:
+        """Get a podcast episode by UUID."""
+        from podcasts.models import PodcastEpisode
+        try:
+            pe = PodcastEpisode.objects.select_related("podcast", "producer").get(
+                uuid=uuid
+            )
+        except PodcastEpisode.DoesNotExist:
+            return None
+        return _media_to_dict(pe, fields=["title", "podcast__title",
+                                           "producer__name", "base_run_time_seconds"])
+
+    def get_beer(self, uuid: str) -> dict | None:
+        """Get a beer by UUID."""
+        from beers.models import Beer
+        try:
+            b = Beer.objects.select_related("style", "producer").get(uuid=uuid)
+        except Beer.DoesNotExist:
+            return None
+        return _media_to_dict(b, fields=["title", "style__name",
+                                          "producer__name", "abv"])
+
+    def get_brick_set(self, uuid: str) -> dict | None:
+        """Get a brick set (LEGO) by UUID."""
+        from bricksets.models import BrickSet
+        try:
+            bs = BrickSet.objects.get(uuid=uuid)
+        except BrickSet.DoesNotExist:
+            return None
+        return _media_to_dict(bs, fields=["title", "piece_count", "set_number"])
+
+    def get_video_game(self, uuid: str) -> dict | None:
+        """Get a video game by UUID."""
+        from videogames.models import VideoGame
+        try:
+            vg = VideoGame.objects.get(uuid=uuid)
+        except VideoGame.DoesNotExist:
+            return None
+        return _media_to_dict(vg, fields=["title", "genre",
+                                           "base_run_time_seconds"])
+
+    def get_puzzle(self, uuid: str) -> dict | None:
+        """Get a puzzle by UUID."""
+        from puzzles.models import Puzzle
+        try:
+            p = Puzzle.objects.select_related("manufacturer").get(uuid=uuid)
+        except Puzzle.DoesNotExist:
+            return None
+        return _media_to_dict(p, fields=["title", "piece_count",
+                                          "manufacturer__name"])
+
+    def get_web_page(self, uuid: str) -> dict | None:
+        """Get a web page by UUID."""
+        from webpages.models import WebPage
+        try:
+            wp = WebPage.objects.select_related("domain").get(uuid=uuid)
+        except WebPage.DoesNotExist:
+            return None
+        return _media_to_dict(wp, fields=["title", "url", "domain__name"])
+
+    def get_task(self, uuid: str) -> dict | None:
+        """Get a task by UUID."""
+        from tasks.models import Task
+        try:
+            t = Task.objects.get(uuid=uuid)
+        except Task.DoesNotExist:
+            return None
+        return _media_to_dict(t, fields=["title", "completed"])
+
+    def get_trail(self, uuid: str) -> dict | None:
+        """Get a trail by UUID."""
+        from trails.models import Trail
+        try:
+            t = Trail.objects.get(uuid=uuid)
+        except Trail.DoesNotExist:
+            return None
+        return _media_to_dict(t, fields=["title", "genre", "base_run_time_seconds"])
+
+    def get_geo_location(self, uuid: str) -> dict | None:
+        """Get a geo location by UUID."""
+        from locations.models import GeoLocation
+        try:
+            gl = GeoLocation.objects.get(uuid=uuid)
+        except GeoLocation.DoesNotExist:
+            return None
+        return _media_to_dict(gl, fields=["title", "latitude", "longitude"])
+
+    def get_life_event(self, uuid: str) -> dict | None:
+        """Get a life event by UUID."""
+        from lifeevents.models import LifeEvent
+        try:
+            le = LifeEvent.objects.get(uuid=uuid)
+        except LifeEvent.DoesNotExist:
+            return None
+        return _media_to_dict(le, fields=["title", "event_date", "genre"])
+
+    def get_mood(self, uuid: str) -> dict | None:
+        """Get a mood entry by UUID."""
+        from moods.models import Mood
+        try:
+            m = Mood.objects.get(uuid=uuid)
+        except Mood.DoesNotExist:
+            return None
+        return _media_to_dict(m, fields=["title", "mood_type", "mood_quality"])
+
+    def get_food(self, uuid: str) -> dict | None:
+        """Get a food entry by UUID."""
+        from foods.models import Food
+        try:
+            f = Food.objects.select_related("category").get(uuid=uuid)
+        except Food.DoesNotExist:
+            return None
+        return _media_to_dict(f, fields=["title", "category__name"])
+
+    def get_bird_sighting(self, uuid: str) -> dict | None:
+        """Get a bird sighting by UUID."""
+        from birds.models import BirdSighting
+        try:
+            bs = BirdSighting.objects.select_related("bird").get(uuid=uuid)
+        except BirdSighting.DoesNotExist:
+            return None
+        return _media_to_dict(bs, fields=["title", "bird__common_name",
+                                           "bird__scientific_name", "location"])
+
+    def get_disc_golf_course(self, uuid: str) -> dict | None:
+        """Get a disc golf course by UUID."""
+        from discgolf.models import DiscGolfCourse
+        try:
+            dg = DiscGolfCourse.objects.get(uuid=uuid)
+        except DiscGolfCourse.DoesNotExist:
+            return None
+        return _media_to_dict(dg, fields=["title", "holes", "location"])
+
+
+class StatsToolset(MCPToolset):
+    def get_scrobble_counts(self, days: int = 30) -> list[dict]:
+        """Get scrobble counts grouped by media type for the last N days."""
+        from django.utils import timezone
+        import datetime
+        from django.db.models import Count
+
+        cutoff = timezone.now() - datetime.timedelta(days=days)
+        qs = (
+            Scrobble.objects.filter(user=self.request.user, timestamp__gte=cutoff)
+            .values("media_type")
+            .annotate(count=Count("id"))
+            .order_by("-count")
+        )
+        return list(qs)
+
+    def get_top_media(
+        self, media_type: str, days: int = 30, limit: int = 10
+    ) -> list[dict]:
+        """Get the most-scrobbled items of a given media type in the last N days.
+        Valid media_type values: Video, Track, Book, BoardGame, Beer, etc."""
+        from django.utils import timezone
+        import datetime
+        from django.db.models import Count
+
+        cutoff = timezone.now() - datetime.timedelta(days=days)
+
+        rel_field = _media_type_to_rel_field(media_type)
+        if not rel_field:
+            return []
+
+        qs = (
+            Scrobble.objects.filter(
+                user=self.request.user,
+                media_type=media_type,
+                timestamp__gte=cutoff,
+            )
+            .values(rel_field)
+            .annotate(count=Count("id"))
+            .order_by("-count")
+        )[:limit]
+
+        results = []
+        for row in qs:
+            obj_id = row[rel_field]
+            if obj_id is None:
+                continue
+            results.append({"id": obj_id, "count": row["count"]})
+        return results
+
+
+def _media_type_to_rel_field(media_type: str) -> str | None:
+    mapping = {
+        "Video": "video",
+        "Track": "track",
+        "PodcastEpisode": "podcast_episode",
+        "SportEvent": "sport_event",
+        "Book": "book",
+        "Paper": "paper",
+        "VideoGame": "video_game",
+        "BoardGame": "board_game",
+        "GeoLocation": "geo_location",
+        "Trail": "trail",
+        "Beer": "beer",
+        "Puzzle": "puzzle",
+        "Food": "food",
+        "Task": "task",
+        "WebPage": "web_page",
+        "LifeEvent": "life_event",
+        "Mood": "mood",
+        "BrickSet": "brick_set",
+        "Channel": "channel",
+        "BirdingLocation": "birding_location",
+        "DiscGolfCourse": "disc_golf_course",
+    }
+    return mapping.get(media_type)
+
+
+def _scrobble_to_dict(s: Scrobble) -> dict:
+    result = {
+        "uuid": str(s.uuid),
+        "media_type": s.media_type,
+        "timestamp": s.timestamp.isoformat() if s.timestamp else None,
+        "stop_timestamp": s.stop_timestamp.isoformat() if s.stop_timestamp else None,
+        "in_progress": s.in_progress,
+        "played_to_completion": s.played_to_completion,
+        "source": s.source,
+        "visibility": s.visibility,
+        "timezone": s.timezone,
+    }
+    if s.log:
+        result["log"] = s.log
+
+    rel = _scrobble_related_to_dict(s)
+    if rel:
+        result["media"] = rel
+
+    return result
+
+
+def _scrobble_related_to_dict(s: Scrobble) -> dict | None:
+    if s.video:
+        return _media_to_dict(s.video, fields=["title", "year", "imdb_id",
+                                                "imdb_rating"])
+    if s.track:
+        return _media_to_dict(s.track, fields=["title",
+                                                "base_run_time_seconds"])
+    if s.book:
+        return _media_to_dict(s.book, fields=["title", "pages"])
+    if s.video_game:
+        return _media_to_dict(s.video_game, fields=["title",
+                                                     "base_run_time_seconds"])
+    if s.board_game:
+        return _media_to_dict(s.board_game, fields=["title"])
+    if s.beer:
+        return _media_to_dict(s.beer, fields=["title"])
+    if s.puzzle:
+        return _media_to_dict(s.puzzle, fields=["title", "piece_count"])
+    if s.food:
+        return _media_to_dict(s.food, fields=["title"])
+    if s.trail:
+        return _media_to_dict(s.trail, fields=["title",
+                                                "base_run_time_seconds"])
+    if s.task:
+        return _media_to_dict(s.task, fields=["title"])
+    if s.web_page:
+        return _media_to_dict(s.web_page, fields=["title", "url"])
+    if s.life_event:
+        return _media_to_dict(s.life_event, fields=["title", "event_date"])
+    if s.mood:
+        return _media_to_dict(s.mood, fields=["title"])
+    if s.brick_set:
+        return _media_to_dict(s.brick_set, fields=["title", "set_number"])
+    if s.podcast_episode:
+        return _media_to_dict(s.podcast_episode, fields=["title"])
+    if s.sport_event:
+        return {"title": str(s.sport_event)}
+    if s.geo_location:
+        return _media_to_dict(s.geo_location, fields=["title", "latitude",
+                                                       "longitude"])
+    if s.birding_location:
+        return _media_to_dict(s.birding_location, fields=["title"])
+    if s.disc_golf_course:
+        return _media_to_dict(s.disc_golf_course, fields=["title", "holes"])
+    if s.channel:
+        return _media_to_dict(s.channel, fields=["title"])
+    return None
+
+
+def _media_to_dict(obj, fields: list[str] | None = None) -> dict:
+    if obj is None:
+        return {}
+    result = {}
+    if hasattr(obj, "uuid"):
+        result["uuid"] = str(obj.uuid)
+    if hasattr(obj, "title"):
+        result["title"] = obj.title
+
+    if fields is None:
+        return result
+
+    resolved = _resolve_fields(obj, fields)
+    for k, v in resolved.items():
+        if k not in result:
+            result[k] = v
+
+    return result
+
+
+def _resolve_fields(obj, fields: list[str]) -> dict:
+    result = {}
+    for field in fields:
+        parts = field.split("__")
+        val = obj
+        try:
+            for part in parts:
+                val = getattr(val, part)
+        except AttributeError:
+            continue
+        if val is not None:
+            if hasattr(val, "all"):
+                val = [str(v) for v in val.all()]
+            result[field] = val
+    return result

vrobbler/settings.py

@@ -236,6 +236,7 @@ INSTALLED_APPS = [
     "allauth.account",
     "allauth.socialaccount",
     "django_celery_results",
+    "mcp_server",
 ]
 
 SITE_ID = 1
@@ -329,6 +330,13 @@ REST_FRAMEWORK = {
     "PAGE_SIZE": 200,
 }
 
+DJANGO_MCP_AUTHENTICATION_CLASSES = [
+    "oauth2_provider.contrib.rest_framework.OAuth2Authentication",
+    "rest_framework.authentication.TokenAuthentication",
+    "rest_framework.authentication.SessionAuthentication",
+]
+DJANGO_MCP_GET_SERVER_INSTRUCTIONS_TOOL = True
+
 LOGIN_REDIRECT_URL = "/"
 
 AUTH_PASSWORD_VALIDATORS = [

vrobbler/urls.py

@@ -188,6 +188,7 @@ urlpatterns = [
     path("", include(people_urls, namespace="people")),
     path("", include(charts_urls, namespace="charts")),
     path("", include(trends_urls, namespace="trends")),
+    path("", include("mcp_server.urls")),
     path("", scrobbles_views.RecentScrobbleList.as_view(), name="vrobbler-home"),
 ]
 if settings.DEBUG: