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

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

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


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 r.json()


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 r.json()


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 r.json()


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


# --- Sources ---

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


@mcp.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}")


@mcp.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,
) -> 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)
    """
    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
    return _post("/sources", json={"source": source})


@mcp.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,
) -> dict:
    """Update an existing source. Only provided fields will be changed."""
    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
    return _put(f"/sources/{source_id}", json={"source": updates})


@mcp.tool()
def delete_source(source_id: int) -> dict:
    """Delete a source (marks it for deletion)."""
    return _delete(f"/sources/{source_id}")


@mcp.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")


@mcp.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")


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


@mcp.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 ---

@mcp.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)


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


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


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


@mcp.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"
    with httpx.Client(timeout=120) as client:
        r = client.delete(f"{BASE}/media/{media_id}", headers=HEADERS, params=params)
        r.raise_for_status()
        return r.json()


# --- Media Profiles ---

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


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


# --- Tasks ---

@mcp.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)


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


# --- Settings & Info ---

@mcp.tool()
def get_settings() -> dict:
    """Get current Pinchflat settings."""
    return _get("/settings")


@mcp.tool()
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")