feat: add Quizzical Beats MCP automation interface

Add comprehensive test suite to reach 30%+ code coverage

.flake8

@@ -0,0 +1,10 @@
+[flake8]
+max-line-length = 100
+# Hard errors only (syntax errors and undefined names) are enforced in CI.
+# Style warnings are reported with --exit-zero.
+exclude =
+    .git,
+    __pycache__,
+    migrations,
+    .venv,
+    venv

.github/workflows/ci.yml

@@ -0,0 +1,98 @@
+name: CI
+
+on:
+  push:
+    branches: ["main"]
+    tags: ["v*.*.*"]
+  pull_request:
+    branches: ["main"]
+
+jobs:
+  lint:
+    name: Lint
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: "pip"
+
+      - name: Install dependencies
+        run: pip install flake8
+
+      - name: Lint with flake8
+        run: |
+          # Stop the build on syntax errors or undefined names
+          flake8 musicround/ --count --select=E9,F63,F7,F82 --show-source --statistics
+          # Warn on style issues (exit-zero means the step won't fail on style warnings)
+          flake8 musicround/ --count --exit-zero --statistics
+
+  test:
+    name: Test
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      actions: write
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: "pip"
+
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y ffmpeg
+
+      - name: Install Python dependencies
+        run: pip install -r requirements.txt
+
+      - name: Run tests with coverage
+        env:
+          FLASK_ENV: testing
+          SECRET_KEY: ci-test-secret-key
+          AUTOMATION_TOKEN: ci-test-automation-token
+        run: |
+          pytest tests/ -v --cov=musicround --cov-report=xml --cov-report=term-missing
+
+      - name: Upload coverage report
+        uses: actions/upload-artifact@v4
+        with:
+          name: coverage-report
+          path: coverage.xml
+          retention-days: 14
+
+  docker-build:
+    name: Docker Build (validation)
+    runs-on: ubuntu-latest
+    needs: [lint, test]
+    permissions:
+      contents: read
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Build Docker image
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          push: false
+          tags: quizzicalbeats:ci
+          cache-from: type=gha
+          cache-to: type=gha,mode=max

.github/workflows/docker-publish.yml

@@ -6,6 +6,7 @@ name: Docker
 # documentation.
 
 on:
+  workflow_dispatch:
   schedule:
     - cron: '30 21 * * *'
   push:
@@ -23,8 +24,56 @@ env:
 
 
 jobs:
-  build:
+  ci:
+    name: Tests & Lint
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      actions: write
 
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: "pip"
+
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y ffmpeg
+
+      - name: Install Python dependencies
+        run: pip install -r requirements.txt
+
+      - name: Lint with flake8
+        run: |
+          pip install flake8
+          # Stop the build on syntax errors or undefined names
+          flake8 musicround/ --count --select=E9,F63,F7,F82 --show-source --statistics
+          # Warn on style issues (exit-zero means the step won't fail on style warnings)
+          flake8 musicround/ --count --exit-zero --statistics
+
+      - name: Run tests with coverage
+        env:
+          FLASK_ENV: testing
+          SECRET_KEY: ci-test-secret-key
+          AUTOMATION_TOKEN: ci-test-automation-token
+        run: |
+          pytest tests/ -v --cov=musicround --cov-report=xml --cov-report=term-missing
+
+      - name: Upload coverage report
+        uses: actions/upload-artifact@v4
+        with:
+          name: coverage-report
+          path: coverage.xml
+          retention-days: 14
+
+  build:
+    needs: [ci]
     runs-on: ubuntu-latest
     permissions:
       contents: read

Dockerfile

@@ -18,7 +18,7 @@ RUN pip install --no-cache-dir -r requirements.txt
 # The .dockerignore file ensures we only include what's specified
 COPY musicround/ ./musicround/
 COPY migrations/ ./migrations/
-COPY run_migration.py run.py docker-entrypoint.sh LICENSE favicon.ico .
+COPY run_migration.py run.py wsgi.py docker-entrypoint.sh LICENSE favicon.ico .
 
 # Make the entrypoint script executable
 RUN chmod +x docker-entrypoint.sh
@@ -31,4 +31,4 @@ ENV PYTHONPATH=/app
 ENV FLASK_APP=run.py
 
 # Use the entrypoint script
-CMD ["./docker-entrypoint.sh"]
+CMD ["./docker-entrypoint.sh"]

docker-entrypoint.sh

@@ -1,23 +1,40 @@
 #!/bin/bash
 set -e
 
-# Start the Flask application with better error reporting
+# Start the Flask application with better error reporting.
 echo "Starting Flask application..."
 export PYTHONUNBUFFERED=1
-export FLASK_DEBUG=1
+: "${FLASK_DEBUG:=0}"
 echo "Flask environment: $FLASK_ENV"
 echo "Database URI: $SQLALCHEMY_DATABASE_URI"
 echo "Database path: $DATABASE_PATH"
-echo "Available environment variables:"
-env | grep -v PASSWORD | grep -v SECRET
 
-# Use flask run with explicit reload for better hot reloading
+if [ "${LOG_ENV:-0}" = "1" ]; then
+  echo "Available non-sensitive environment variables:"
+  env | grep -Evi '(PASSWORD|PASS|SECRET|TOKEN|KEY|AUTH|CREDENTIAL|PRIVATE)'
+fi
+
 export PYTHONFAULTHANDLER=1
 export PYTHONDONTWRITEBYTECODE=1
 
-echo "Starting Flask development server with hot reload..."
-exec python -m flask run --host=0.0.0.0 --port=5000 --reload --debug || {
-  echo "Flask application failed to start. Error details:"
-  python -c "import traceback; traceback.print_exc()"
-  exit 1
-}
+case "${FLASK_DEBUG,,}" in
+  1|true|yes|on)
+    echo "Starting Flask development server with hot reload..."
+    exec python -m flask run --host=0.0.0.0 --port=5000 --reload --debug
+    ;;
+  *)
+    : "${GUNICORN_BIND:=0.0.0.0:5000}"
+    : "${GUNICORN_WORKERS:=2}"
+    : "${GUNICORN_THREADS:=4}"
+    : "${GUNICORN_TIMEOUT:=120}"
+    echo "Starting Gunicorn application server..."
+    exec gunicorn \
+      --bind "$GUNICORN_BIND" \
+      --workers "$GUNICORN_WORKERS" \
+      --threads "$GUNICORN_THREADS" \
+      --timeout "$GUNICORN_TIMEOUT" \
+      --access-logfile - \
+      --error-logfile - \
+      wsgi:app
+    ;;
+esac

docs/developer-guide/mcp.md

@@ -0,0 +1,63 @@
+# MCP Interface
+
+Quizzical Beats includes an MCP server for agentic round production workflows.
+It exposes the same catalog, round, export, email, and custom-audio capabilities
+used by the Flask application.
+
+## Run Locally
+
+Install dependencies and start the MCP server from the repository root:
+
+```bash
+pip install -r requirements.txt
+python -m musicround.mcp_server
+```
+
+The server uses the normal Quizzical Beats Flask configuration. Set the same
+environment variables you use for the web app, including `SECRET_KEY`,
+`AUTOMATION_TOKEN`, database configuration, mail settings, and any Spotify,
+Deezer, OpenAI, AWS Polly, or ElevenLabs credentials needed by the tools you
+plan to call.
+
+## Tools
+
+The MCP server exposes these tools:
+
+| Tool | Purpose |
+| --- | --- |
+| `find_songs` | Search the existing Quizzical Beats catalog before adding duplicates. |
+| `add_song` | Add or update a catalog song, including platform IDs and tags. |
+| `import_catalog_item` | Import a Spotify or Deezer track, album, or playlist. |
+| `compile_round` | Create a named round from explicit song IDs or selection criteria. |
+| `rename_round` | Set or clear a round name. |
+| `create_round_from_playlist` | Import a playlist and turn the imported songs into a round. |
+| `generate_round_assets` | Generate the round PDF and/or MP3. |
+| `inspect_round_mp3` | Check round MP3 duration, loudness, silence, and clipping indicators. |
+| `inspect_round_pdf` | Check round PDF existence and basic structural validity. |
+| `send_round_email` | Generate assets and email the finished round bundle. |
+| `generate_tts_snippet` | Generate and assign custom intro, replay, or outro TTS MP3s. |
+
+## Intended Workflow
+
+1. Search with `find_songs` to avoid duplicates.
+2. Add missing tracks with `add_song` or import platform content with
+   `import_catalog_item`.
+3. Create the round with `compile_round` or `create_round_from_playlist`.
+4. Generate PDF and MP3 files with `generate_round_assets`.
+5. Inspect the generated files with `inspect_round_pdf` and `inspect_round_mp3`.
+6. Send the completed bundle with `send_round_email`.
+
+For Spotify imports, pass a `user_id` for a user with connected Spotify tokens.
+For email, either pass an explicit recipient or use a selected user that has an
+email address.
+
+## Custom Audio
+
+Use `generate_tts_snippet` to update the reusable audio segments:
+
+- `intro`: lead-in before the first song.
+- `replay`: announcement before the repeat section.
+- `outro`: lead-out after the round.
+
+Supported TTS services follow the existing application helper: `openai`, `polly`,
+and `elevenlabs`.

migrations/add_user_is_admin.py

@@ -0,0 +1,52 @@
+"""
+Migration script to add the is_admin flag expected by the current User model.
+"""
+import logging
+
+from sqlalchemy import inspect, text
+
+logger = logging.getLogger(__name__)
+
+
+def run_migration():
+    """
+    Add user.is_admin and backfill it from the existing admin role assignment.
+    Returns:
+    - True: if changes were made successfully
+    - None: if no changes were needed
+    - False: if errors occurred
+    """
+    from musicround import db
+
+    try:
+        inspector = inspect(db.engine)
+        existing_columns = [column["name"] for column in inspector.get_columns("user")]
+
+        if "is_admin" in existing_columns:
+            logger.info("is_admin column already exists")
+            return None
+
+        with db.engine.connect() as conn:
+            logger.info("Adding is_admin column")
+            conn.execute(text('ALTER TABLE "user" ADD COLUMN is_admin BOOLEAN DEFAULT 0'))
+            conn.execute(
+                text(
+                    """
+                    UPDATE "user"
+                       SET is_admin = 1
+                     WHERE id IN (
+                         SELECT ur.user_id
+                           FROM user_roles ur
+                           JOIN role r ON r.id = ur.role_id
+                          WHERE lower(r.name) = 'admin'
+                     )
+                    """
+                )
+            )
+            conn.commit()
+
+        logger.info("Added is_admin column")
+        return True
+    except Exception as e:
+        logger.error(f"Migration add_user_is_admin failed: {str(e)}")
+        return False

mkdocs.yml

@@ -59,9 +59,10 @@ nav:
   - Developer Guide:
     - Architecture: developer-guide/architecture.md
     - API Reference: developer-guide/api-reference.md
+    - MCP Interface: developer-guide/mcp.md
     - Database Schema: developer-guide/database-schema.md
     - OAuth Integration: developer-guide/oauth-integration.md
     - Contributing: developer-guide/contributing.md
   - FAQ: faq.md
   - Changelog: changelog.md
-  - Brand Identity: brand-identity.md
+  - Brand Identity: brand-identity.md

musicround/mcp_server.py

@@ -0,0 +1,225 @@
+"""MCP server for agentic Quizzical Beats workflows.
+
+Run with:
+    python -m musicround.mcp_server
+"""
+
+from __future__ import annotations
+
+from functools import lru_cache
+from typing import Any
+
+from mcp.server.fastmcp import FastMCP
+
+from musicround import create_app
+from musicround.services import automation
+
+
+mcp = FastMCP("Quizzical Beats")
+
+
+@lru_cache(maxsize=1)
+def _app():
+    """Create the Flask app once for the MCP server process."""
+    return create_app()
+
+
+def _with_app_context(func, *args, **kwargs) -> dict[str, Any]:
+    """Run a service function inside the Quizzical Beats app context."""
+    app = _app()
+    with app.app_context():
+        return func(*args, **kwargs)
+
+
+@mcp.tool()
+def find_songs(
+    query: str | None = None,
+    title: str | None = None,
+    artist: str | None = None,
+    spotify_id: str | None = None,
+    deezer_id: str | None = None,
+    isrc: str | None = None,
+    limit: int = 20,
+) -> dict[str, Any]:
+    """Search the local Quizzical Beats catalog before adding a song."""
+    return _with_app_context(
+        automation.find_songs,
+        query=query,
+        title=title,
+        artist=artist,
+        spotify_id=spotify_id,
+        deezer_id=deezer_id,
+        isrc=isrc,
+        limit=limit,
+    )
+
+
+@mcp.tool()
+def add_song(
+    title: str,
+    artist: str,
+    album_name: str | None = None,
+    genre: str | None = None,
+    year: int | None = None,
+    preview_url: str | None = None,
+    cover_url: str | None = None,
+    spotify_id: str | None = None,
+    deezer_id: str | None = None,
+    isrc: str | None = None,
+    tags: list[str] | None = None,
+    source: str = "manual",
+) -> dict[str, Any]:
+    """Add a song to Quizzical Beats if it is not already present."""
+    return _with_app_context(
+        automation.add_song,
+        title=title,
+        artist=artist,
+        album_name=album_name,
+        genre=genre,
+        year=year,
+        preview_url=preview_url,
+        cover_url=cover_url,
+        spotify_id=spotify_id,
+        deezer_id=deezer_id,
+        isrc=isrc,
+        tags=tags,
+        source=source,
+    )
+
+
+@mcp.tool()
+def import_catalog_item(
+    service_name: str,
+    item_type: str,
+    item_id_or_url: str,
+    user_id: int | None = None,
+) -> dict[str, Any]:
+    """Import a Spotify or Deezer track, album, or playlist into the catalog."""
+    return _with_app_context(
+        automation.import_catalog_item,
+        service_name=service_name,
+        item_type=item_type,
+        item_id_or_url=item_id_or_url,
+        user_id=user_id,
+    )
+
+
+@mcp.tool()
+def compile_round(
+    name: str | None = None,
+    round_type: str = "random",
+    count: int = 8,
+    criteria: str | None = None,
+    song_ids: list[int] | None = None,
+) -> dict[str, Any]:
+    """Compile songs into a named quiz round."""
+    return _with_app_context(
+        automation.create_round,
+        name=name,
+        round_type=round_type,
+        count=count,
+        criteria=criteria,
+        song_ids=song_ids,
+    )
+
+
+@mcp.tool()
+def rename_round(round_id: int, name: str | None) -> dict[str, Any]:
+    """Set or clear the display name for a round."""
+    return _with_app_context(automation.rename_round, round_id=round_id, name=name)
+
+
+@mcp.tool()
+def create_round_from_playlist(
+    service_name: str,
+    playlist_id_or_url: str,
+    name: str | None = None,
+    count: int = 8,
+    user_id: int | None = None,
+) -> dict[str, Any]:
+    """Import a Spotify or Deezer playlist and turn it into a quiz round."""
+    return _with_app_context(
+        automation.create_round_from_playlist,
+        service_name=service_name,
+        playlist_id_or_url=playlist_id_or_url,
+        name=name,
+        count=count,
+        user_id=user_id,
+    )
+
+
+@mcp.tool()
+def generate_round_assets(
+    round_id: int,
+    user_id: int | None = None,
+    include_pdf: bool = True,
+    include_mp3: bool = True,
+) -> dict[str, Any]:
+    """Generate the PDF and/or MP3 files for a round."""
+    return _with_app_context(
+        automation.generate_round_assets,
+        round_id=round_id,
+        user_id=user_id,
+        include_pdf=include_pdf,
+        include_mp3=include_mp3,
+    )
+
+
+@mcp.tool()
+def inspect_round_mp3(path: str | None = None, round_id: int | None = None) -> dict[str, Any]:
+    """Check a round MP3 for duration, loudness, clipping, and silence issues."""
+    return _with_app_context(automation.inspect_mp3_quality, path=path, round_id=round_id)
+
+
+@mcp.tool()
+def inspect_round_pdf(path: str | None = None, round_id: int | None = None) -> dict[str, Any]:
+    """Check that a round PDF exists and has a valid basic PDF structure."""
+    return _with_app_context(automation.inspect_pdf_quality, path=path, round_id=round_id)
+
+
+@mcp.tool()
+def send_round_email(
+    round_id: int,
+    recipient: str | None = None,
+    user_id: int | None = None,
+    subject: str | None = None,
+    body_text: str | None = None,
+) -> dict[str, Any]:
+    """Generate assets and email the completed round bundle."""
+    return _with_app_context(
+        automation.email_round,
+        round_id=round_id,
+        recipient=recipient,
+        user_id=user_id,
+        subject=subject,
+        body_text=body_text,
+    )
+
+
+@mcp.tool()
+def generate_tts_snippet(
+    user_id: int,
+    mp3_type: str,
+    text: str,
+    service: str = "openai",
+    voice: str | None = None,
+    model: str | None = None,
+    stability: float | None = None,
+    similarity: float | None = None,
+) -> dict[str, Any]:
+    """Generate and assign a custom intro, replay, or outro TTS MP3."""
+    return _with_app_context(
+        automation.generate_tts_snippet,
+        user_id=user_id,
+        mp3_type=mp3_type,
+        text=text,
+        service=service,
+        voice=voice,
+        model=model,
+        stability=stability,
+        similarity=similarity,
+    )
+
+
+if __name__ == "__main__":
+    mcp.run()

musicround/services/__init__.py

@@ -0,0 +1 @@
+"""Service-layer helpers for Quizzical Beats."""

musicround/services/automation.py

@@ -0,0 +1,579 @@
+"""Automation services used by the MCP server and agent workflows."""
+
+from __future__ import annotations
+
+import os
+import re
+from datetime import datetime
+from typing import Any, Iterable
+
+from flask import current_app
+from flask_login import login_user, logout_user
+from pydub import AudioSegment
+from sqlalchemy import or_
+
+from musicround import db
+from musicround.helpers.email_helper import send_email
+from musicround.helpers.import_helper import ImportHelper
+from musicround.helpers.utils import generate_tts_mp3
+from musicround.models import Round, RoundExport, Song, Tag, User
+
+
+class AutomationError(ValueError):
+    """Raised when an automation request cannot be completed."""
+
+
+def _song_summary(song: Song) -> dict[str, Any]:
+    data = song.to_dict()
+    return {
+        "id": data["id"],
+        "title": data["title"],
+        "artist": data["artist"],
+        "genre": data["genre"],
+        "year": data["year"],
+        "source": song.source,
+        "preview_url": data["preview_url"],
+        "spotify_id": data["spotify_id"],
+        "deezer_id": data["deezer_id"],
+        "isrc": data["isrc"],
+        "tags": data["tags"],
+    }
+
+
+def _round_summary(round_obj: Round) -> dict[str, Any]:
+    ids = [int(song_id) for song_id in round_obj.songs.split(",") if song_id]
+    songs = Song.query.filter(Song.id.in_(ids)).all()
+    songs_by_id = {song.id: song for song in songs}
+    ordered = [songs_by_id[song_id] for song_id in ids if song_id in songs_by_id]
+    return {
+        "id": round_obj.id,
+        "name": round_obj.name,
+        "round_type": round_obj.round_type,
+        "criteria": round_obj.round_criteria_used,
+        "song_ids": ids,
+        "songs": [_song_summary(song) for song in ordered],
+        "mp3_generated": round_obj.mp3_generated,
+        "pdf_generated": round_obj.pdf_generated,
+        "last_generated_at": (
+            round_obj.last_generated_at.isoformat() if round_obj.last_generated_at else None
+        ),
+    }
+
+
+def _find_user(user_id: int | None = None) -> User:
+    if user_id is not None:
+        user = db.session.get(User, user_id)
+        if not user:
+            raise AutomationError(f"User {user_id} was not found.")
+        return user
+
+    users = User.query.order_by(User.id).limit(2).all()
+    if len(users) == 1:
+        return users[0]
+    if not users:
+        raise AutomationError(
+            "No users exist yet. Create a user before generating user-owned assets."
+        )
+    raise AutomationError(
+        "Multiple users exist. Pass user_id so the action uses the right account."
+    )
+
+
+def _parse_external_id(service_name: str, item_type: str, value: str) -> str:
+    service = service_name.lower()
+    item = item_type.lower()
+    stripped = value.strip()
+    if service == "spotify":
+        match = re.search(rf"spotify\.com/{item}/([A-Za-z0-9]+)", stripped)
+        if match:
+            return match.group(1)
+        match = re.search(rf"spotify:{item}:([A-Za-z0-9]+)", stripped)
+        if match:
+            return match.group(1)
+    if service == "deezer":
+        match = re.search(r"deezer\.page\.link/([A-Za-z0-9]+)", stripped)
+        if match:
+            return match.group(1)
+        match = re.search(rf"deezer\.com/(?:[a-z]{{2}}/)?{item}/(\d+)", stripped)
+        if match:
+            return match.group(1)
+    return stripped.split("?")[0].rstrip("/")
+
+
+def _attach_tags(song: Song, tag_names: Iterable[str] | None) -> None:
+    for raw_name in tag_names or []:
+        tag_name = raw_name.strip()
+        if not tag_name:
+            continue
+        tag = Tag.query.filter(Tag.name.ilike(tag_name)).first()
+        if not tag:
+            tag = Tag(name=tag_name)
+            db.session.add(tag)
+            db.session.flush()
+        if tag not in song.tags:
+            song.tags.append(tag)
+
+
+def add_song(
+    title: str,
+    artist: str,
+    album_name: str | None = None,
+    genre: str | None = None,
+    year: int | None = None,
+    preview_url: str | None = None,
+    cover_url: str | None = None,
+    spotify_id: str | None = None,
+    deezer_id: str | None = None,
+    isrc: str | None = None,
+    tags: list[str] | None = None,
+    source: str = "manual",
+) -> dict[str, Any]:
+    """Add or update a song in the local catalog."""
+    if not title or not artist:
+        raise AutomationError("Both title and artist are required.")
+
+    existing = None
+    if isrc:
+        existing = Song.query.filter_by(isrc=isrc).first()
+    if not existing and spotify_id:
+        existing = Song.query.filter_by(spotify_id=spotify_id).first()
+    if not existing and deezer_id:
+        existing = Song.query.filter_by(deezer_id=str(deezer_id)).first()
+
+    song = existing or Song(title=title.strip(), artist=artist.strip())
+    song.title = title.strip()
+    song.artist = artist.strip()
+    song.album_name = album_name or song.album_name
+    song.genre = genre or song.genre
+    song.year = year or song.year
+    song.preview_url = preview_url or song.preview_url
+    song.cover_url = cover_url or song.cover_url
+    song.spotify_id = spotify_id or song.spotify_id
+    song.deezer_id = str(deezer_id) if deezer_id else song.deezer_id
+    song.isrc = isrc or song.isrc
+    song.source = source or song.source or "manual"
+    _attach_tags(song, tags)
+
+    if not existing:
+        db.session.add(song)
+    db.session.commit()
+    return {"created": existing is None, "song": _song_summary(song)}
+
+
+def find_songs(
+    query: str | None = None,
+    title: str | None = None,
+    artist: str | None = None,
+    spotify_id: str | None = None,
+    deezer_id: str | None = None,
+    isrc: str | None = None,
+    limit: int = 20,
+) -> dict[str, Any]:
+    """Search the local catalog before adding or importing tracks."""
+    if limit < 1 or limit > 100:
+        raise AutomationError("limit must be between 1 and 100.")
+
+    filters = []
+    if query:
+        pattern = f"%{query.strip()}%"
+        filters.append(or_(Song.title.ilike(pattern), Song.artist.ilike(pattern)))
+    if title:
+        filters.append(Song.title.ilike(f"%{title.strip()}%"))
+    if artist:
+        filters.append(Song.artist.ilike(f"%{artist.strip()}%"))
+    if spotify_id:
+        filters.append(Song.spotify_id == spotify_id)
+    if deezer_id:
+        filters.append(Song.deezer_id == str(deezer_id))
+    if isrc:
+        filters.append(Song.isrc == isrc)
+
+    song_query = Song.query
+    for condition in filters:
+        song_query = song_query.filter(condition)
+    songs = song_query.order_by(Song.artist, Song.title).limit(limit).all()
+    return {"count": len(songs), "songs": [_song_summary(song) for song in songs]}
+
+
+def import_catalog_item(
+    service_name: str,
+    item_type: str,
+    item_id_or_url: str,
+    user_id: int | None = None,
+) -> dict[str, Any]:
+    """Import a track, album, or playlist from Spotify or Deezer."""
+    service = service_name.lower()
+    item = item_type.lower()
+    external_id = _parse_external_id(service, item, item_id_or_url)
+    if service == "spotify":
+        user = _find_user(user_id)
+        with current_app.test_request_context():
+            login_user(user)
+            try:
+                result = ImportHelper.import_item(service, item, external_id)
+            finally:
+                logout_user()
+    else:
+        result = ImportHelper.import_item(service, item, external_id)
+
+    if result.get("error_count", 0) > 0:
+        current_app.logger.warning("Import completed with errors: %s", result.get("errors", []))
+    return {"service_name": service, "item_type": item, "item_id": external_id, "result": result}
+
+
+def _songs_for_round(
+    round_type: str,
+    count: int,
+    criteria: str | None = None,
+    song_ids: list[int] | None = None,
+) -> tuple[str, str, list[Song]]:
+    from musicround.routes.generate import (
+        get_random_songs,
+        get_random_songs_from_decade,
+        get_random_songs_from_genre,
+        get_random_songs_from_least_used_decade,
+        get_random_songs_from_least_used_genre,
+        get_songs_by_tag,
+    )
+
+    normalized = round_type.lower().strip()
+    if song_ids:
+        songs_by_id = {song.id: song for song in Song.query.filter(Song.id.in_(song_ids)).all()}
+        songs = [songs_by_id[song_id] for song_id in song_ids if song_id in songs_by_id]
+        if len(songs) != len(song_ids):
+            missing = sorted(set(song_ids) - set(songs_by_id))
+            raise AutomationError(f"Unknown song IDs: {missing}")
+        return "Manual", "Explicit song selection", songs[:count]
+
+    if normalized == "random":
+        return "Random", "Random Selection", get_random_songs(count)
+    if normalized == "genre":
+        if criteria:
+            return "Genre", criteria, get_random_songs_from_genre(criteria, x=count)
+        songs, chosen = get_random_songs_from_least_used_genre(count)
+        return "Genre", chosen or "Least Used Genre", songs
+    if normalized == "decade":
+        if criteria:
+            return "Decade", criteria, get_random_songs_from_decade(criteria, x=count)
+        songs, chosen = get_random_songs_from_least_used_decade(count)
+        return "Decade", chosen or "Least Used Decade", songs
+    if normalized == "tag":
+        if not criteria:
+            raise AutomationError("Tag rounds require criteria with the tag name.")
+        return "Tag", criteria, get_songs_by_tag(criteria, count)
+    raise AutomationError("round_type must be one of random, genre, decade, tag, or manual.")
+
+
+def create_round(
+    name: str | None = None,
+    round_type: str = "random",
+    count: int = 8,
+    criteria: str | None = None,
+    song_ids: list[int] | None = None,
+) -> dict[str, Any]:
+    """Create and persist a quiz round."""
+    if count < 1:
+        raise AutomationError("count must be at least 1.")
+
+    resolved_type, resolved_criteria, songs = _songs_for_round(
+        round_type, count, criteria, song_ids
+    )
+    if not songs:
+        raise AutomationError("No songs matched the requested round criteria.")
+
+    round_obj = Round(
+        name=name,
+        round_type=resolved_type,
+        round_criteria_used=resolved_criteria,
+        songs=",".join(str(song.id) for song in songs),
+        created_at=datetime.utcnow(),
+    )
+    db.session.add(round_obj)
+    for song in songs:
+        song.used_count = (song.used_count or 0) + 1
+        song.last_used = datetime.utcnow()
+    db.session.commit()
+    return {"round": _round_summary(round_obj)}
+
+
+def rename_round(round_id: int, name: str | None) -> dict[str, Any]:
+    """Rename a persisted round."""
+    round_obj = db.session.get(Round, round_id)
+    if not round_obj:
+        raise AutomationError(f"Round {round_id} was not found.")
+    round_obj.name = name.strip() if name and name.strip() else None
+    db.session.commit()
+    return {"round": _round_summary(round_obj)}
+
+
+def _spotify_playlist_song_ids(playlist_id: str, limit: int, user_id: int | None) -> list[int]:
+    from musicround.routes.generate import get_songs_from_spotify_playlist
+
+    user = _find_user(user_id)
+    with current_app.test_request_context():
+        login_user(user)
+        try:
+            songs = get_songs_from_spotify_playlist(playlist_id)
+        finally:
+            logout_user()
+    return [song.id for song in songs[:limit]]
+
+
+def _deezer_playlist_song_ids(playlist_id: str, limit: int) -> list[int]:
+    deezer_client = current_app.config.get("deezer")
+    if not deezer_client:
+        raise AutomationError("Deezer client is not configured.")
+
+    tracks = deezer_client.get_playlist_tracks(playlist_id)
+    song_ids = []
+    lastfm_key = current_app.config.get("LASTFM_API_KEY")
+    for track in tracks[:limit]:
+        track_id = track.get("id")
+        if not track_id:
+            continue
+        song, _ = deezer_client.import_track(track_id, lastfm_api_key=lastfm_key)
+        if song:
+            song_ids.append(song.id)
+    db.session.commit()
+    return song_ids
+
+
+def create_round_from_playlist(
+    service_name: str,
+    playlist_id_or_url: str,
+    name: str | None = None,
+    count: int = 8,
+    user_id: int | None = None,
+) -> dict[str, Any]:
+    """Import a playlist and create a manual round from the imported songs."""
+    imported = import_catalog_item(service_name, "playlist", playlist_id_or_url, user_id=user_id)
+    playlist_id = imported["item_id"]
+    if service_name.lower() == "spotify":
+        song_ids = _spotify_playlist_song_ids(playlist_id, count, user_id)
+    else:
+        song_ids = imported.get("result", {}).get(
+            "imported_song_ids"
+        ) or _deezer_playlist_song_ids(playlist_id, count)
+    if not song_ids:
+        raise AutomationError("Playlist import did not return song IDs to build a round.")
+    round_result = create_round(
+        name=name, round_type="manual", count=count, song_ids=song_ids[:count]
+    )
+    return {"import": imported, "round": round_result["round"]}
+
+
+def generate_round_pdf(round_id: int) -> dict[str, Any]:
+    from musicround.routes.rounds import generate_pdf
+
+    round_obj = db.session.get(Round, round_id)
+    if not round_obj:
+        raise AutomationError(f"Round {round_id} was not found.")
+    pdf_data = generate_pdf(round_id)
+    if isinstance(pdf_data, str):
+        raise AutomationError(pdf_data)
+    round_obj.pdf_generated = True
+    round_obj.last_generated_at = datetime.utcnow()
+    db.session.commit()
+    path = os.path.join("/data/pdfs", f"round_{round_id}.pdf")
+    return {"round_id": round_id, "path": path, "bytes": len(pdf_data)}
+
+
+def generate_round_mp3(round_id: int, user_id: int | None = None) -> dict[str, Any]:
+    from musicround.routes.rounds import round_mp3
+
+    round_obj = db.session.get(Round, round_id)
+    if not round_obj:
+        raise AutomationError(f"Round {round_id} was not found.")
+    user = _find_user(user_id)
+    with current_app.test_request_context(headers={"X-Requested-With": "XMLHttpRequest"}):
+        login_user(user)
+        try:
+            response = round_mp3(round_id)
+        finally:
+            logout_user()
+
+    if hasattr(response, "get_json"):
+        payload = response.get_json(silent=True) or {}
+        if payload.get("success") is False or payload.get("error"):
+            raise AutomationError(payload.get("error", "MP3 generation failed."))
+
+    path = os.path.join("/data/rounds", f"round_{round_id}.mp3")
+    if not os.path.exists(path):
+        raise AutomationError(f"MP3 generation did not create {path}.")
+    return {"round_id": round_id, "path": path, "bytes": os.path.getsize(path)}
+
+
+def generate_round_assets(
+    round_id: int,
+    user_id: int | None = None,
+    include_pdf: bool = True,
+    include_mp3: bool = True,
+) -> dict[str, Any]:
+    """Generate requested round assets."""
+    assets: dict[str, Any] = {"round_id": round_id}
+    if include_pdf:
+        assets["pdf"] = generate_round_pdf(round_id)
+    if include_mp3:
+        assets["mp3"] = generate_round_mp3(round_id, user_id=user_id)
+    return assets
+
+
+def email_round(
+    round_id: int,
+    recipient: str | None = None,
+    user_id: int | None = None,
+    subject: str | None = None,
+    body_text: str | None = None,
+) -> dict[str, Any]:
+    """Generate assets and send a round as an email attachment bundle."""
+    user = _find_user(user_id)
+    target = recipient or user.email
+    if not target:
+        raise AutomationError("No recipient was provided and the selected user has no email.")
+
+    assets = generate_round_assets(round_id, user_id=user.id)
+    round_obj = db.session.get(Round, round_id)
+    title = round_obj.name if round_obj and round_obj.name else f"Quizzical Beats Round {round_id}"
+    email_subject = subject or title
+    email_body = body_text or "Attached are the MP3 and PDF files for your quiz round."
+
+    attachments = []
+    with open(assets["pdf"]["path"], "rb") as pdf_file:
+        attachments.append(
+            {
+                "data": pdf_file.read(),
+                "filename": f"round_{round_id}.pdf",
+                "mimetype": "application/pdf",
+            }
+        )
+    with open(assets["mp3"]["path"], "rb") as mp3_file:
+        attachments.append(
+            {
+                "data": mp3_file.read(),
+                "filename": f"round_{round_id}.mp3",
+                "mimetype": "audio/mpeg",
+            }
+        )
+
+    success, message = send_email(target, email_subject, email_body, attachments)
+    export = RoundExport(
+        round_id=round_id,
+        user_id=user.id,
+        export_type="email",
+        destination=target,
+        include_mp3s=True,
+        status="success" if success else "failed",
+        error_message=None if success else message,
+    )
+    db.session.add(export)
+    db.session.commit()
+    if not success:
+        raise AutomationError(message)
+    return {"success": True, "message": message, "recipient": target, "assets": assets}
+
+
+def inspect_mp3_quality(path: str | None = None, round_id: int | None = None) -> dict[str, Any]:
+    """Inspect basic MP3 quality and flag common generation issues."""
+    if not path:
+        if round_id is None:
+            raise AutomationError("Pass either path or round_id.")
+        path = os.path.join("/data/rounds", f"round_{round_id}.mp3")
+    if not os.path.exists(path):
+        raise AutomationError(f"MP3 file not found: {path}")
+
+    audio = AudioSegment.from_mp3(path)
+    warnings = []
+    if len(audio) < 1000:
+        warnings.append("Audio is shorter than one second.")
+    if audio.dBFS == float("-inf"):
+        warnings.append("Audio appears to be silent.")
+    elif audio.dBFS < -35:
+        warnings.append("Average loudness is very low.")
+    elif audio.dBFS > -8:
+        warnings.append("Average loudness is high; check for limiting or clipping.")
+
+    samples = audio.get_array_of_samples()
+    max_possible = float(1 << (8 * audio.sample_width - 1))
+    clipped = sum(1 for sample in samples if abs(sample) >= max_possible * 0.99)
+    clipping_ratio = clipped / len(samples) if samples else 0
+    if clipping_ratio > 0.001:
+        warnings.append("Potential clipping detected.")
+
+    return {
+        "path": path,
+        "duration_seconds": round(len(audio) / 1000, 3),
+        "channels": audio.channels,
+        "frame_rate": audio.frame_rate,
+        "sample_width_bytes": audio.sample_width,
+        "average_dbfs": None if audio.dBFS == float("-inf") else round(audio.dBFS, 2),
+        "peak_dbfs": round(audio.max_dBFS, 2),
+        "clipping_ratio": round(clipping_ratio, 6),
+        "warnings": warnings,
+        "ok": not warnings,
+    }
+
+
+def inspect_pdf_quality(path: str | None = None, round_id: int | None = None) -> dict[str, Any]:
+    """Inspect basic PDF integrity for generated round sheets."""
+    if not path:
+        if round_id is None:
+            raise AutomationError("Pass either path or round_id.")
+        path = os.path.join("/data/pdfs", f"round_{round_id}.pdf")
+    if not os.path.exists(path):
+        raise AutomationError(f"PDF file not found: {path}")
+
+    with open(path, "rb") as pdf_file:
+        data = pdf_file.read()
+    warnings = []
+    if not data.startswith(b"%PDF-"):
+        warnings.append("File does not start with a PDF header.")
+    if b"%%EOF" not in data[-2048:]:
+        warnings.append("PDF EOF marker was not found near the end of the file.")
+    if len(data) < 1024:
+        warnings.append("PDF file is unusually small.")
+    page_count = data.count(b"/Type /Page")
+    if page_count == 0:
+        warnings.append("No PDF pages were detected.")
+
+    return {
+        "path": path,
+        "bytes": len(data),
+        "page_count_estimate": page_count,
+        "warnings": warnings,
+        "ok": not warnings,
+    }
+
+
+def generate_tts_snippet(
+    user_id: int,
+    mp3_type: str,
+    text: str,
+    service: str = "openai",
+    voice: str | None = None,
+    model: str | None = None,
+    stability: float | None = None,
+    similarity: float | None = None,
+) -> dict[str, Any]:
+    """Generate and assign a custom intro, replay, or outro MP3 for a user."""
+    if mp3_type not in {"intro", "replay", "outro"}:
+        raise AutomationError("mp3_type must be intro, replay, or outro.")
+    if not text:
+        raise AutomationError("text is required for TTS generation.")
+
+    user = _find_user(user_id)
+    path = generate_tts_mp3(
+        text=text,
+        username=user.username,
+        mp3_type=mp3_type,
+        service=service,
+        voice=voice,
+        model=model,
+        stability=stability,
+        similarity=similarity,
+    )
+    if not path:
+        raise AutomationError("TTS generation failed.")
+
+    setattr(user, f"{mp3_type}_mp3", path)
+    db.session.commit()
+    return {"user_id": user.id, "mp3_type": mp3_type, "path": path}

requirements.txt

@@ -17,9 +17,11 @@ Flask-Mail
 Flask-Session
 authlib>=1.6.5
 Flask-Caching
+gunicorn
 psutil
+mcp[cli]
 
 # Testing dependencies
 pytest>=7.4.0
 pytest-cov>=4.1.0
-pytest-flask>=1.2.0
+pytest-flask>=1.2.0

tests/test_automation_service.py

@@ -0,0 +1,129 @@
+"""Tests for agent automation services."""
+
+import os
+import shutil
+import tempfile
+from unittest.mock import patch
+
+import pytest
+
+from musicround.models import Song, User, db
+from musicround.services import automation
+
+
+def _create_user(username="agentuser", email="agent@example.com"):
+    user = User(username=username, email=email)
+    user.password = "AgentPass123!"
+    db.session.add(user)
+    db.session.commit()
+    return user
+
+
+def _create_song(title="Song", artist="Artist", **kwargs):
+    song = Song(title=title, artist=artist, **kwargs)
+    db.session.add(song)
+    db.session.commit()
+    return song
+
+
+class TestSongAutomation:
+    """Tests for catalog lookup and mutation."""
+
+    def test_find_songs_by_query(self, app):
+        with app.app_context():
+            _create_song(title="Blue Monday", artist="New Order", genre="Synthpop")
+
+            result = automation.find_songs(query="blue")
+
+            assert result["count"] == 1
+            assert result["songs"][0]["title"] == "Blue Monday"
+
+    def test_add_song_reuses_existing_by_isrc_and_adds_tags(self, app):
+        with app.app_context():
+            _create_song(title="Old Title", artist="Old Artist", isrc="ABC123")
+
+            result = automation.add_song(
+                title="New Title",
+                artist="New Artist",
+                isrc="ABC123",
+                tags=["warmup", "classic"],
+            )
+
+            assert result["created"] is False
+            assert Song.query.count() == 1
+            song = Song.query.first()
+            assert song.title == "New Title"
+            assert sorted(tag.name for tag in song.tags) == ["classic", "warmup"]
+
+
+class TestRoundAutomation:
+    """Tests for round creation and naming."""
+
+    def test_create_and_rename_manual_round(self, app):
+        with app.app_context():
+            song_one = _create_song(title="One", artist="A")
+            song_two = _create_song(title="Two", artist="B")
+
+            created = automation.create_round(
+                name="Initial Name",
+                round_type="manual",
+                song_ids=[song_one.id, song_two.id],
+            )
+            renamed = automation.rename_round(created["round"]["id"], "Final Name")
+
+            assert created["round"]["song_ids"] == [song_one.id, song_two.id]
+            assert renamed["round"]["name"] == "Final Name"
+            assert Song.query.get(song_one.id).used_count == 1
+
+
+class TestAssetInspection:
+    """Tests for generated asset quality checks."""
+
+    def test_inspect_pdf_quality(self, app):
+        with app.app_context(), tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
+            tmp.write(b"%PDF-1.4\n")
+            tmp.write(b"1 0 obj << /Type /Page >> endobj\n" * 40)
+            tmp.write(b"%%EOF")
+            tmp_path = tmp.name
+
+        try:
+            with app.app_context():
+                result = automation.inspect_pdf_quality(path=tmp_path)
+
+            assert result["ok"] is True
+            assert result["page_count_estimate"] > 0
+        finally:
+            os.remove(tmp_path)
+
+    def test_inspect_mp3_quality_existing_fixture(self, app):
+        if not shutil.which("ffprobe"):
+            pytest.skip("ffprobe is required for MP3 inspection")
+
+        fixture_path = os.path.abspath("musicround/mp3/intro.mp3")
+
+        with app.app_context():
+            result = automation.inspect_mp3_quality(path=fixture_path)
+
+        assert result["duration_seconds"] > 0
+        assert result["channels"] >= 1
+        assert "ok" in result
+
+
+class TestTTSAutomation:
+    """Tests for TTS snippet assignment."""
+
+    def test_generate_tts_snippet_updates_user_audio_path(self, app):
+        with app.app_context():
+            user = _create_user()
+
+            with patch("musicround.services.automation.generate_tts_mp3") as mock_tts:
+                mock_tts.return_value = "custommp3/agentuser/intro.mp3"
+                result = automation.generate_tts_snippet(
+                    user_id=user.id,
+                    mp3_type="intro",
+                    text="Welcome to the quiz",
+                    service="openai",
+                )
+
+            assert result["path"] == "custommp3/agentuser/intro.mp3"
+            assert User.query.get(user.id).intro_mp3 == "custommp3/agentuser/intro.mp3"

tests/test_import_queue.py

@@ -2,7 +2,10 @@
 import pytest
 import threading
 import time
-from musicround.helpers.import_queue import ImportJob, ImportQueue
+from unittest.mock import patch
+
+from musicround.helpers.import_queue import ImportJob, ImportQueue, ImportWorker
+from musicround.models import User, db
 
 
 class TestImportJob:
@@ -154,3 +157,46 @@ class TestImportQueue:
 
         assert not errors
         assert len(results) == 5
+
+
+class TestImportWorker:
+    """Tests for ImportWorker job processing."""
+
+    def test_process_job_imports_as_user(self, app):
+        """Test that _process_job logs in the target user and imports the item."""
+        with app.app_context():
+            user = User(username='workeruser', email='worker@example.com')
+            user.password = 'WorkerPass123!'
+            db.session.add(user)
+            db.session.commit()
+
+            worker = ImportWorker(app, ImportQueue())
+            job = ImportJob(
+                priority=1,
+                service_name='deezer',
+                item_type='track',
+                item_id='123',
+                user_id=user.id,
+            )
+
+            with patch('musicround.helpers.import_queue.ImportHelper.import_item') as mock_import:
+                worker._process_job(job)
+
+            mock_import.assert_called_once_with('deezer', 'track', '123')
+
+    def test_process_job_unknown_user_does_not_import(self, app):
+        """Test that jobs for missing users are ignored."""
+        with app.app_context():
+            worker = ImportWorker(app, ImportQueue())
+            job = ImportJob(
+                priority=1,
+                service_name='deezer',
+                item_type='track',
+                item_id='123',
+                user_id=999,
+            )
+
+            with patch('musicround.helpers.import_queue.ImportHelper.import_item') as mock_import:
+                worker._process_job(job)
+
+            mock_import.assert_not_called()

wsgi.py

@@ -0,0 +1,3 @@
+from musicround import create_app
+
+app = create_app()