mcp-servers/packages/pinchflat-mcp/server.py

#!/usr/bin/env python3
"""Pinchflat MCP server — exposes the Pinchflat REST API as MCP tools.

Tool tiers
----------
Two tiers of tools exist and only one is exposed by default:

* ``subscriptions`` (default) — sources, media items, media profiles, tasks.
  This is what an assistant needs to manage subscriptions and their files.
* ``admin`` — global application state: ``get_settings``, ``get_app_info``.
  These describe the installation rather than a subscription and are NOT
  exposed unless explicitly enabled.

Select the tier with ``PINCHFLAT_MCP_TOOLSET``:

    PINCHFLAT_MCP_TOOLSET=subscriptions   # default
    PINCHFLAT_MCP_TOOLSET=admin
    PINCHFLAT_MCP_TOOLSET=all

Responses are scrubbed of secrets on the way out (see ``_scrub``), because the
Pinchflat settings record embeds the application's own ``api_token``. Exposing
it to a tool caller hands out a credential that outlives the tool session, so
secret-bearing keys are redacted at the transport boundary rather than left to
individual tools to remember.
"""

import os
import sys
from typing import Any, Optional

import httpx
from fastmcp import FastMCP

PINCHFLAT_URL = os.environ.get("PINCHFLAT_URL", "http://localhost:8945")
PINCHFLAT_TOKEN = os.environ.get("PINCHFLAT_API_TOKEN", "")

if not PINCHFLAT_TOKEN:
    print("PINCHFLAT_API_TOKEN environment variable is required", file=sys.stderr)
    sys.exit(1)

BASE = f"{PINCHFLAT_URL.rstrip('/')}/api/v1"
HEADERS = {"Authorization": f"Bearer {PINCHFLAT_TOKEN}", "Content-Type": "application/json"}

mcp = FastMCP("pinchflat")

# --- Tool tiers -------------------------------------------------------------

_TIERS = {
    "subscriptions": {"subscriptions"},
    "admin": {"subscriptions", "admin"},
    "all": {"subscriptions", "admin"},
}
_TOOLSET = os.environ.get("PINCHFLAT_MCP_TOOLSET", "subscriptions").strip().lower()
ENABLED_TIERS = _TIERS.get(_TOOLSET, _TIERS["subscriptions"])


def tool(tier: str = "subscriptions"):
    """Register a tool only when its tier is enabled."""

    def decorator(fn):
        if tier in ENABLED_TIERS:
            return mcp.tool()(fn)
        return fn

    return decorator


# --- Response scrubbing -----------------------------------------------------

# Keys whose values are credentials and must never cross the tool boundary.
SECRET_KEYS = frozenset(
    {
        "api_token",
        "api_key",
        "token",
        "secret",
        "password",
        "password_hash",
        "youtube_api_key",
        "apprise_server",
    }
)
REDACTED = "[redacted]"


def _scrub(obj: Any) -> Any:
    """Recursively redact credential-bearing values from an API response."""
    if isinstance(obj, dict):
        out = {}
        for k, v in obj.items():
            if isinstance(k, str) and k.strip().lower() in SECRET_KEYS and v:
                out[k] = REDACTED
            else:
                out[k] = _scrub(v)
        return out
    if isinstance(obj, list):
        return [_scrub(v) for v in obj]
    return obj


# --- HTTP helpers -----------------------------------------------------------


def _unwrap(r: httpx.Response) -> Any:
    """Decode a response and scrub secrets before it reaches the caller."""
    body = r.json()
    return _scrub(body)


def _get(path: str, params: Optional[dict] = None) -> Any:
    """Make a GET request to the Pinchflat API."""
    with httpx.Client(timeout=30) as client:
        r = client.get(f"{BASE}{path}", headers=HEADERS, params=params)
        r.raise_for_status()
        return _unwrap(r)


def _post(path: str, json: Optional[dict] = None) -> Any:
    """Make a POST request to the Pinchflat API."""
    with httpx.Client(timeout=30) as client:
        r = client.post(f"{BASE}{path}", headers=HEADERS, json=json)
        r.raise_for_status()
        return _unwrap(r)


def _put(path: str, json: dict) -> Any:
    """Make a PUT request to the Pinchflat API."""
    with httpx.Client(timeout=30) as client:
        r = client.put(f"{BASE}{path}", headers=HEADERS, json=json)
        r.raise_for_status()
        return _unwrap(r)


def _delete(path: str, params: Optional[dict] = None, timeout: int = 30) -> Any:
    """Make a DELETE request to the Pinchflat API."""
    with httpx.Client(timeout=timeout) as client:
        r = client.delete(f"{BASE}{path}", headers=HEADERS, params=params)
        r.raise_for_status()
        return _unwrap(r)


# --- Sources ---


@tool()
def list_sources() -> dict:
    """List all configured media sources (YouTube channels, playlists, etc.)."""
    return _get("/sources")


@tool()
def get_source(source_id: int) -> dict:
    """Get details of a specific source by ID, including its pending tasks."""
    return _get(f"/sources/{source_id}")


@tool()
def create_source(
    original_url: str,
    media_profile_id: int,
    collection_type: str = "channel",
    custom_name: Optional[str] = None,
    download_cutoff_date: Optional[str] = None,
    retention_period_days: Optional[int] = None,
) -> dict:
    """Create a new media source (e.g. add a YouTube channel to monitor).

    Args:
        original_url: URL of the channel/playlist to monitor
        media_profile_id: ID of the media profile to use for downloads
        collection_type: 'channel', 'playlist', or 'video'
        custom_name: Optional custom name for the source
        download_cutoff_date: Only download media after this date (YYYY-MM-DD)
        retention_period_days: Delete downloaded media this many days after
            publication; omit to keep everything
    """
    source = {
        "media_profile_id": media_profile_id,
        "collection_type": collection_type,
        "original_url": original_url,
    }
    if custom_name:
        source["custom_name"] = custom_name
    if download_cutoff_date:
        source["download_cutoff_date"] = download_cutoff_date
    if retention_period_days is not None:
        source["retention_period_days"] = retention_period_days
    return _post("/sources", json={"source": source})


@tool()
def update_source(
    source_id: int,
    custom_name: Optional[str] = None,
    download_cutoff_date: Optional[str] = None,
    enabled: Optional[bool] = None,
    index_frequency_minutes: Optional[int] = None,
    retention_period_days: Optional[int] = None,
    prevent_culling: Optional[bool] = None,
    download_media: Optional[bool] = None,
) -> dict:
    """Update an existing source. Only provided fields will be changed.

    Args:
        retention_period_days: Delete downloaded media this many days after
            publication (0 keeps everything)
        prevent_culling: If True, never delete this source's downloads
        download_media: If False, index the source but download nothing
    """
    updates = {}
    if custom_name is not None:
        updates["custom_name"] = custom_name
    if download_cutoff_date is not None:
        updates["download_cutoff_date"] = download_cutoff_date
    if enabled is not None:
        updates["enabled"] = enabled
    if index_frequency_minutes is not None:
        updates["index_frequency_minutes"] = index_frequency_minutes
    if retention_period_days is not None:
        updates["retention_period_days"] = retention_period_days
    if prevent_culling is not None:
        updates["prevent_culling"] = prevent_culling
    if download_media is not None:
        updates["download_media"] = download_media
    if not updates:
        return {"data": {"message": "No fields to update."}}
    return _put(f"/sources/{source_id}", json={"source": updates})


@tool()
def delete_source(source_id: int) -> dict:
    """Delete a source (marks it for deletion).

    Note: this removes the source and its media records, but does NOT delete
    the files from disk. Delete the media items first if space is the goal.
    """
    return _delete(f"/sources/{source_id}")


@tool()
def force_download_pending(source_id: int) -> dict:
    """Force download of all pending media items for a source."""
    return _post(f"/sources/{source_id}/force_download_pending")


@tool()
def force_redownload(source_id: int) -> dict:
    """Force re-download of all existing media for a source."""
    return _post(f"/sources/{source_id}/force_redownload")


@tool()
def force_index(source_id: int) -> dict:
    """Force re-index of a source."""
    return _post(f"/sources/{source_id}/force_index")


@tool()
def force_metadata_refresh(source_id: int) -> dict:
    """Force metadata refresh for a source."""
    return _post(f"/sources/{source_id}/force_metadata_refresh")


# --- Media Items ---


@tool()
def list_media(
    source_id: Optional[int] = None,
    limit: int = 50,
    offset: int = 0,
    downloaded: Optional[bool] = None,
) -> dict:
    """List media items with pagination. Optionally filter by source_id or download status.

    Args:
        source_id: Filter to a specific source
        limit: Max items to return (default 50)
        offset: Skip first N items for pagination (default 0)
        downloaded: If True, only return downloaded media; if False, only undownloaded
    """
    params = {"limit": limit, "offset": offset}
    if source_id:
        params["source_id"] = source_id
    if downloaded is not None:
        params["downloaded"] = "true" if downloaded else "false"
    return _get("/media", params=params)


@tool()
def search_media(query: str) -> dict:
    """Search media items by title or description."""
    return _get("/media/search", params={"q": query})


@tool()
def get_media(media_id: int) -> dict:
    """Get details of a specific media item."""
    return _get(f"/media/{media_id}")


@tool()
def force_download_media(media_id: int) -> dict:
    """Force download of a specific media item."""
    return _post(f"/media/{media_id}/force_download")


@tool()
def delete_media(media_id: int, prevent_download: bool = False) -> dict:
    """Delete a media item's downloaded files from disk.

    Args:
        media_id: The media item ID to delete
        prevent_download: If True, prevent the item from being re-downloaded in the future
    """
    params = {}
    if prevent_download:
        params["prevent_download"] = "true"
    return _delete(f"/media/{media_id}", params=params, timeout=120)


# --- Media Profiles ---


@tool()
def list_media_profiles() -> dict:
    """List all media profiles (download quality/container settings)."""
    return _get("/media_profiles")


@tool()
def get_media_profile(profile_id: int) -> dict:
    """Get details of a specific media profile."""
    return _get(f"/media_profiles/{profile_id}")


# --- Tasks ---


@tool()
def list_tasks(source_id: Optional[int] = None) -> dict:
    """List all tasks. Optionally filter by source_id."""
    params = {}
    if source_id:
        params["source_id"] = source_id
    return _get("/tasks", params=params)


@tool()
def get_task(task_id: int) -> dict:
    """Get details of a specific task."""
    return _get(f"/tasks/{task_id}")


# --- Settings & Info (admin tier — not exposed by default) ---


@tool("admin")
def get_settings() -> dict:
    """Get current Pinchflat settings (credential values are redacted)."""
    return _get("/settings")


@tool("admin")
def get_app_info() -> dict:
    """Get Pinchflat app info (version, environment, etc.)."""
    return _get("/app_info")


if __name__ == "__main__":
    transport = os.environ.get("PINCHFLAT_MCP_TRANSPORT", "stdio")
    mcp.run(transport=transport if transport in ("stdio", "http") else "stdio")