Add generic datastore CRUD automation

Add a bearer-protected streamable HTTP MCP ASGI entrypoint for production deployment.

Includes remote host/origin configuration, health checks, fallback token handling, documentation, and tests covering authentication behavior.

Verified with full local pytest suite and GitHub CI. Copilot review feedback addressed before merge.

.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,88 @@
+# 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
+```
+
+For a production streamable HTTP endpoint, run the authenticated ASGI entrypoint:
+
+```bash
+MCP_BEARER_TOKEN=... uvicorn musicround.mcp_http:app --host 0.0.0.0 --port 8000
+```
+
+If `MCP_BEARER_TOKEN` is not set, the HTTP entrypoint falls back to
+`AUTOMATION_TOKEN`. Set `MCP_ALLOWED_HOSTS` and `MCP_ALLOWED_ORIGINS` when the
+server is exposed behind a reverse proxy or ingress.
+
+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. |
+| `datastore_schema` | Describe all mapped datastore object types, columns, and primary keys. |
+| `list_datastore_objects` | List persisted objects with optional exact-match filters, ordering, limit, and offset. |
+| `get_datastore_object` | Fetch one persisted object by primary key. |
+| `create_datastore_object` | Create one persisted object from scalar column fields. |
+| `update_datastore_object` | Update scalar column fields on one persisted object. |
+| `delete_datastore_object` | Delete one persisted object by primary key. |
+| `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. |
+
+`find_songs` includes `used_count`, `usage_frequency`, and `last_used` for each
+result so agents can see how often songs have already appeared in rounds.
+
+The generic datastore CRUD tools operate on mapped SQLAlchemy models, including
+`song`, `round`, `tag`, `song_tag`, `user`, `role`, `user_preferences`,
+`round_export`, `system_setting`, and `import_job_record`. Read results redact
+fields whose names contain `password`, `token`, or `secret` unless
+`include_sensitive` is explicitly set.
+
+## 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_http.py

@@ -0,0 +1,92 @@
+"""Authenticated streamable HTTP entrypoint for the Quizzical Beats MCP server."""
+
+from __future__ import annotations
+
+import os
+from secrets import compare_digest
+
+from starlette.responses import JSONResponse
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+from musicround.mcp_server import mcp
+
+
+_BASE_ALLOWED_HOSTS = tuple(mcp.settings.transport_security.allowed_hosts)
+_BASE_ALLOWED_ORIGINS = tuple(mcp.settings.transport_security.allowed_origins)
+
+
+class BearerAuthMiddleware:
+    """Require a bearer token for MCP HTTP traffic."""
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        if scope.get("path") == "/healthz":
+            await JSONResponse({"ok": True})(scope, receive, send)
+            return
+
+        expected = os.getenv("MCP_BEARER_TOKEN") or os.getenv("AUTOMATION_TOKEN")
+        if not expected:
+            await JSONResponse(
+                {"error": "MCP bearer token is not configured."},
+                status_code=500,
+            )(scope, receive, send)
+            return
+
+        headers = dict(scope.get("headers", []))
+        authorization = headers.get(b"authorization", b"").decode("latin1")
+        scheme, _, token = authorization.partition(" ")
+        if scheme.lower() != "bearer" or not compare_digest(token.strip(), expected):
+            await JSONResponse(
+                {"error": "Unauthorized."},
+                headers={"WWW-Authenticate": "Bearer"},
+                status_code=401,
+            )(scope, receive, send)
+            return
+
+        await self.app(scope, receive, send)
+
+
+def _configure_server() -> None:
+    host = os.getenv("MCP_HOST", "0.0.0.0")
+    port = int(os.getenv("MCP_PORT", "8000"))
+    mcp.settings.host = host
+    mcp.settings.port = port
+
+    allowed_hosts = [
+        value.strip()
+        for value in os.getenv("MCP_ALLOWED_HOSTS", "qb.kaufdeinquiz.com").split(",")
+        if value.strip()
+    ]
+    allowed_origins = [
+        value.strip()
+        for value in os.getenv(
+            "MCP_ALLOWED_ORIGINS", "https://qb.kaufdeinquiz.com"
+        ).split(",")
+        if value.strip()
+    ]
+    security = mcp.settings.transport_security
+    security.allowed_hosts = list(dict.fromkeys([*_BASE_ALLOWED_HOSTS, *allowed_hosts]))
+    security.allowed_origins = list(
+        dict.fromkeys([*_BASE_ALLOWED_ORIGINS, *allowed_origins])
+    )
+
+
+def build_app() -> ASGIApp:
+    """Build the authenticated streamable HTTP MCP ASGI app."""
+    _configure_server()
+    return BearerAuthMiddleware(mcp.streamable_http_app())
+
+
+app = build_app()
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host=mcp.settings.host, port=mcp.settings.port)

musicround/mcp_server.py

@@ -0,0 +1,309 @@
+"""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 datastore_schema() -> dict[str, Any]:
+    """Describe every datastore object type available to generic CRUD tools."""
+    return _with_app_context(automation.datastore_schema)
+
+
+@mcp.tool()
+def list_datastore_objects(
+    object_type: str,
+    filters: dict[str, Any] | None = None,
+    limit: int = 50,
+    offset: int = 0,
+    order_by: str | None = None,
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """List datastore objects such as songs, rounds, users, tags, exports, and settings."""
+    return _with_app_context(
+        automation.list_datastore_objects,
+        object_type=object_type,
+        filters=filters,
+        limit=limit,
+        offset=offset,
+        order_by=order_by,
+        include_sensitive=include_sensitive,
+    )
+
+
+@mcp.tool()
+def get_datastore_object(
+    object_type: str,
+    object_id: Any,
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """Fetch one datastore object by primary key."""
+    return _with_app_context(
+        automation.get_datastore_object,
+        object_type=object_type,
+        object_id=object_id,
+        include_sensitive=include_sensitive,
+    )
+
+
+@mcp.tool()
+def create_datastore_object(
+    object_type: str,
+    fields: dict[str, Any],
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """Create one datastore object from scalar column fields."""
+    return _with_app_context(
+        automation.create_datastore_object,
+        object_type=object_type,
+        fields=fields,
+        include_sensitive=include_sensitive,
+    )
+
+
+@mcp.tool()
+def update_datastore_object(
+    object_type: str,
+    object_id: Any,
+    fields: dict[str, Any],
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """Update scalar column fields on one datastore object."""
+    return _with_app_context(
+        automation.update_datastore_object,
+        object_type=object_type,
+        object_id=object_id,
+        fields=fields,
+        include_sensitive=include_sensitive,
+    )
+
+
+@mcp.tool()
+def delete_datastore_object(object_type: str, object_id: Any) -> dict[str, Any]:
+    """Delete one datastore object by primary key."""
+    return _with_app_context(
+        automation.delete_datastore_object,
+        object_type=object_type,
+        object_id=object_id,
+    )
+
+
+@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,875 @@
+"""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 inspect as sa_inspect
+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 import models as datastore_models
+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"],
+        "used_count": data["used_count"] or 0,
+        "usage_frequency": data["used_count"] or 0,
+        "last_used": data["last_used"],
+        "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 _snake_case(value: str) -> str:
+    value = re.sub(r"(.)([A-Z][a-z]+)", r"\1_\2", value)
+    value = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", value)
+    return value.lower()
+
+
+def _model_registry() -> dict[str, type[db.Model]]:
+    registry: dict[str, type[db.Model]] = {}
+    for value in vars(datastore_models).values():
+        if not isinstance(value, type):
+            continue
+        if value is db.Model or not issubclass(value, db.Model):
+            continue
+        mapper = sa_inspect(value, raiseerr=False)
+        if mapper is None or getattr(value, "__table__", None) is None:
+            continue
+        canonical = _snake_case(value.__name__)
+        registry[canonical] = value
+        registry[value.__name__] = value
+        registry[value.__name__.lower()] = value
+        registry[value.__tablename__] = value
+    return registry
+
+
+def _canonical_model_key(model: type[db.Model]) -> str:
+    return _snake_case(model.__name__)
+
+
+def _get_model(object_type: str) -> type[db.Model]:
+    if not object_type:
+        raise AutomationError("object_type is required.")
+    model = _model_registry().get(object_type)
+    if not model:
+        allowed = sorted({_canonical_model_key(model) for model in _model_registry().values()})
+        raise AutomationError(f"Unknown object_type '{object_type}'. Allowed values: {allowed}")
+    return model
+
+
+def _column_map(model: type[db.Model]) -> dict[str, Any]:
+    return {column.key: column for column in sa_inspect(model).columns}
+
+
+def _primary_key_columns(model: type[db.Model]) -> list[Any]:
+    return list(sa_inspect(model).primary_key)
+
+
+def _is_sensitive_field(field_name: str) -> bool:
+    lowered = field_name.lower()
+    return any(marker in lowered for marker in ("password", "token", "secret"))
+
+
+def _json_value(value: Any, *, sensitive: bool = False, include_sensitive: bool = False) -> Any:
+    if sensitive and value is not None and not include_sensitive:
+        return "[redacted]"
+    if isinstance(value, datetime):
+        return value.isoformat()
+    return value
+
+
+def _serialize_model(instance: db.Model, *, include_sensitive: bool = False) -> dict[str, Any]:
+    data = {}
+    for column in sa_inspect(instance.__class__).columns:
+        value = getattr(instance, column.key)
+        data[column.key] = _json_value(
+            value,
+            sensitive=_is_sensitive_field(column.key),
+            include_sensitive=include_sensitive,
+        )
+    return data
+
+
+def _coerce_column_value(column: Any, value: Any) -> Any:
+    if value is None:
+        return None
+    try:
+        python_type = column.type.python_type
+    except NotImplementedError:
+        return value
+
+    if python_type is datetime:
+        if isinstance(value, datetime):
+            return value
+        if isinstance(value, str):
+            normalized = value.replace("Z", "+00:00")
+            return datetime.fromisoformat(normalized)
+        raise AutomationError(f"{column.key} must be an ISO datetime string.")
+    if python_type is bool and isinstance(value, str):
+        lowered = value.lower()
+        if lowered in {"true", "1", "yes", "on"}:
+            return True
+        if lowered in {"false", "0", "no", "off"}:
+            return False
+    if python_type in {int, float, str, bool} and not isinstance(value, python_type):
+        return python_type(value)
+    return value
+
+
+def _identity_for_object(model: type[db.Model], object_id: Any) -> Any:
+    primary_key = _primary_key_columns(model)
+    if not primary_key:
+        raise AutomationError(f"{_canonical_model_key(model)} does not have a primary key.")
+
+    if isinstance(object_id, dict):
+        missing = [column.key for column in primary_key if column.key not in object_id]
+        if missing:
+            raise AutomationError(f"Missing primary key field(s): {missing}")
+        values = [_coerce_column_value(column, object_id[column.key]) for column in primary_key]
+    elif len(primary_key) == 1:
+        values = [_coerce_column_value(primary_key[0], object_id)]
+    elif isinstance(object_id, list):
+        if len(object_id) != len(primary_key):
+            raise AutomationError(
+                f"Composite primary key requires {len(primary_key)} values in order."
+            )
+        values = [
+            _coerce_column_value(column, object_id[index])
+            for index, column in enumerate(primary_key)
+        ]
+    else:
+        names = [column.key for column in primary_key]
+        raise AutomationError(f"Composite primary key requires an object with keys {names}.")
+
+    return values[0] if len(values) == 1 else tuple(values)
+
+
+def _get_datastore_instance(model: type[db.Model], object_id: Any) -> db.Model:
+    instance = db.session.get(model, _identity_for_object(model, object_id))
+    if not instance:
+        raise AutomationError(f"{_canonical_model_key(model)} {object_id} was not found.")
+    return instance
+
+
+def _apply_datastore_filters(query: Any, model: type[db.Model], filters: dict[str, Any] | None) -> Any:
+    columns = _column_map(model)
+    for field_name, raw_value in (filters or {}).items():
+        column = columns.get(field_name)
+        if column is None:
+            raise AutomationError(f"Unknown filter field '{field_name}'.")
+        query = query.filter(getattr(model, field_name) == _coerce_column_value(column, raw_value))
+    return query
+
+
+def _assign_datastore_fields(instance: db.Model, fields: dict[str, Any], *, creating: bool) -> None:
+    if not fields:
+        raise AutomationError("fields must not be empty.")
+
+    model = instance.__class__
+    columns = _column_map(model)
+    primary_keys = {column.key for column in _primary_key_columns(model)}
+    for field_name, raw_value in fields.items():
+        column = columns.get(field_name)
+        if column is None:
+            raise AutomationError(f"Unknown field '{field_name}'.")
+        if not creating and field_name in primary_keys:
+            raise AutomationError("Primary key fields cannot be updated.")
+        setattr(instance, field_name, _coerce_column_value(column, raw_value))
+
+
+def datastore_schema() -> dict[str, Any]:
+    """Describe datastore objects available through generic MCP CRUD tools."""
+    models_by_key = {
+        _canonical_model_key(model): model for model in _model_registry().values()
+    }
+    objects = []
+    for object_type, model in sorted(models_by_key.items()):
+        mapper = sa_inspect(model)
+        objects.append(
+            {
+                "object_type": object_type,
+                "table": model.__tablename__,
+                "primary_key": [column.key for column in mapper.primary_key],
+                "columns": [
+                    {
+                        "name": column.key,
+                        "type": str(column.type),
+                        "nullable": column.nullable,
+                        "primary_key": column.primary_key,
+                        "sensitive": _is_sensitive_field(column.key),
+                    }
+                    for column in mapper.columns
+                ],
+            }
+        )
+    return {"object_types": [item["object_type"] for item in objects], "objects": objects}
+
+
+def list_datastore_objects(
+    object_type: str,
+    filters: dict[str, Any] | None = None,
+    limit: int = 50,
+    offset: int = 0,
+    order_by: str | None = None,
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """List persisted rows for a mapped datastore object."""
+    if limit < 1 or limit > 500:
+        raise AutomationError("limit must be between 1 and 500.")
+    if offset < 0:
+        raise AutomationError("offset must not be negative.")
+
+    model = _get_model(object_type)
+    query = _apply_datastore_filters(model.query, model, filters)
+    total = query.count()
+
+    if order_by:
+        descending = order_by.startswith("-")
+        field_name = order_by[1:] if descending else order_by
+        if field_name not in _column_map(model):
+            raise AutomationError(f"Unknown order_by field '{field_name}'.")
+        column = getattr(model, field_name)
+        query = query.order_by(column.desc() if descending else column.asc())
+    else:
+        primary_key = _primary_key_columns(model)
+        if primary_key:
+            query = query.order_by(*[getattr(model, column.key).asc() for column in primary_key])
+
+    rows = query.offset(offset).limit(limit).all()
+    return {
+        "object_type": _canonical_model_key(model),
+        "count": len(rows),
+        "total": total,
+        "limit": limit,
+        "offset": offset,
+        "objects": [_serialize_model(row, include_sensitive=include_sensitive) for row in rows],
+    }
+
+
+def get_datastore_object(
+    object_type: str,
+    object_id: Any,
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """Fetch a single persisted datastore object by primary key."""
+    model = _get_model(object_type)
+    instance = _get_datastore_instance(model, object_id)
+    return {
+        "object_type": _canonical_model_key(model),
+        "object": _serialize_model(instance, include_sensitive=include_sensitive),
+    }
+
+
+def create_datastore_object(
+    object_type: str,
+    fields: dict[str, Any],
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """Create a persisted datastore object from scalar column fields."""
+    model = _get_model(object_type)
+    instance = model()
+    _assign_datastore_fields(instance, fields, creating=True)
+    db.session.add(instance)
+    db.session.commit()
+    return {
+        "created": True,
+        "object_type": _canonical_model_key(model),
+        "object": _serialize_model(instance, include_sensitive=include_sensitive),
+    }
+
+
+def update_datastore_object(
+    object_type: str,
+    object_id: Any,
+    fields: dict[str, Any],
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """Update scalar column fields for a persisted datastore object."""
+    model = _get_model(object_type)
+    instance = _get_datastore_instance(model, object_id)
+    _assign_datastore_fields(instance, fields, creating=False)
+    db.session.commit()
+    return {
+        "updated": True,
+        "object_type": _canonical_model_key(model),
+        "object": _serialize_model(instance, include_sensitive=include_sensitive),
+    }
+
+
+def delete_datastore_object(object_type: str, object_id: Any) -> dict[str, Any]:
+    """Delete a persisted datastore object by primary key."""
+    model = _get_model(object_type)
+    instance = _get_datastore_instance(model, object_id)
+    serialized = _serialize_model(instance)
+    db.session.delete(instance)
+    db.session.commit()
+    return {
+        "deleted": True,
+        "object_type": _canonical_model_key(model),
+        "object": serialized,
+    }
+
+
+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,211 @@
+"""Tests for agent automation services."""
+
+import os
+import shutil
+import tempfile
+from unittest.mock import patch
+
+import pytest
+
+os.environ.setdefault("SECRET_KEY", "test-secret-key-for-testing-only")
+os.environ.setdefault("AUTOMATION_TOKEN", "test-automation-token-for-testing")
+
+from musicround.models import Song, SongTag, Tag, 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",
+                used_count=3,
+            )
+
+            result = automation.find_songs(query="blue")
+
+            assert result["count"] == 1
+            assert result["songs"][0]["title"] == "Blue Monday"
+            assert result["songs"][0]["used_count"] == 3
+            assert result["songs"][0]["usage_frequency"] == 3
+            assert "last_used" in result["songs"][0]
+
+    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"
+
+
+class TestDatastoreCrudAutomation:
+    """Tests for generic datastore CRUD operations exposed through MCP."""
+
+    def test_datastore_schema_lists_mapped_models(self, app):
+        with app.app_context():
+            schema = automation.datastore_schema()
+
+            assert "song" in schema["object_types"]
+            assert "round" in schema["object_types"]
+            assert "user" in schema["object_types"]
+            song_schema = next(
+                item for item in schema["objects"] if item["object_type"] == "song"
+            )
+            assert song_schema["primary_key"] == ["id"]
+            assert any(column["name"] == "title" for column in song_schema["columns"])
+
+    def test_crud_lifecycle_for_single_primary_key_object(self, app):
+        with app.app_context():
+            created = automation.create_datastore_object("tag", {"name": "warmup"})
+            tag_id = created["object"]["id"]
+
+            listed = automation.list_datastore_objects(
+                "tag", filters={"name": "warmup"}, order_by="id"
+            )
+            fetched = automation.get_datastore_object("tag", tag_id)
+            updated = automation.update_datastore_object(
+                "tag", tag_id, {"name": "opener"}
+            )
+            deleted = automation.delete_datastore_object("tag", tag_id)
+
+            assert created["object_type"] == "tag"
+            assert listed["total"] == 1
+            assert fetched["object"]["name"] == "warmup"
+            assert updated["object"]["name"] == "opener"
+            assert deleted["deleted"] is True
+            assert Tag.query.get(tag_id) is None
+
+    def test_crud_supports_composite_primary_keys(self, app):
+        with app.app_context():
+            song = _create_song(title="Composite", artist="Key")
+            tag = Tag(name="linked")
+            db.session.add(tag)
+            db.session.commit()
+
+            created = automation.create_datastore_object(
+                "song_tag", {"song_id": song.id, "tag_id": tag.id}
+            )
+            fetched = automation.get_datastore_object(
+                "song_tag", {"song_id": song.id, "tag_id": tag.id}
+            )
+            deleted = automation.delete_datastore_object(
+                "song_tag", {"song_id": song.id, "tag_id": tag.id}
+            )
+
+            assert created["object"]["song_id"] == song.id
+            assert fetched["object"]["tag_id"] == tag.id
+            assert deleted["deleted"] is True
+            assert SongTag.query.count() == 0
+
+    def test_user_sensitive_fields_are_redacted_by_default(self, app):
+        with app.app_context():
+            user = _create_user()
+            user.spotify_token = "secret-token"
+            db.session.commit()
+
+            result = automation.get_datastore_object("user", user.id)
+
+            assert result["object"]["spotify_token"] == "[redacted]"
+            assert result["object"]["password_hash"] == "[redacted]"

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()

tests/test_mcp_http.py

@@ -0,0 +1,69 @@
+"""Tests for the authenticated MCP HTTP entrypoint."""
+
+import os
+
+from starlette.responses import Response
+from starlette.testclient import TestClient
+
+os.environ.setdefault("SECRET_KEY", "test-secret-key-for-testing-only")
+os.environ.setdefault("AUTOMATION_TOKEN", "test-automation-token-for-testing")
+
+from musicround.mcp_http import BearerAuthMiddleware, build_app, mcp  # noqa: E402
+
+
+async def _dummy_app(scope, receive, send):
+    await Response(status_code=204)(scope, receive, send)
+
+
+def test_healthz_does_not_require_auth(monkeypatch):
+    monkeypatch.setenv("MCP_BEARER_TOKEN", "test-mcp-token")
+    client = TestClient(build_app())
+
+    response = client.get("/healthz")
+
+    assert response.status_code == 200
+    assert response.json() == {"ok": True}
+
+
+def test_mcp_requires_bearer_token(monkeypatch):
+    monkeypatch.setenv("MCP_BEARER_TOKEN", "test-mcp-token")
+    client = TestClient(BearerAuthMiddleware(_dummy_app))
+
+    missing = client.get("/mcp")
+    wrong = client.get("/mcp", headers={"Authorization": "Bearer wrong"})
+    accepted = client.get("/mcp", headers={"Authorization": "Bearer test-mcp-token"})
+
+    assert missing.status_code == 401
+    assert missing.headers["WWW-Authenticate"] == "Bearer"
+    assert wrong.status_code == 401
+    assert accepted.status_code != 401
+
+
+def test_mcp_falls_back_to_automation_token(monkeypatch):
+    monkeypatch.delenv("MCP_BEARER_TOKEN", raising=False)
+    monkeypatch.setenv("AUTOMATION_TOKEN", "automation-token")
+    client = TestClient(BearerAuthMiddleware(_dummy_app))
+
+    accepted = client.get("/mcp", headers={"Authorization": "Bearer automation-token"})
+
+    assert accepted.status_code == 204
+
+
+def test_mcp_reports_missing_bearer_configuration(monkeypatch):
+    monkeypatch.delenv("MCP_BEARER_TOKEN", raising=False)
+    monkeypatch.delenv("AUTOMATION_TOKEN", raising=False)
+    client = TestClient(BearerAuthMiddleware(_dummy_app))
+
+    response = client.get("/mcp", headers={"Authorization": "Bearer anything"})
+
+    assert response.status_code == 500
+
+
+def test_allowed_hosts_are_replaced_between_builds(monkeypatch):
+    monkeypatch.setenv("MCP_ALLOWED_HOSTS", "first.example")
+    build_app()
+    monkeypatch.setenv("MCP_ALLOWED_HOSTS", "second.example")
+    build_app()
+
+    assert "first.example" not in mcp.settings.transport_security.allowed_hosts
+    assert "second.example" in mcp.settings.transport_security.allowed_hosts

wsgi.py

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