Add pinchflat-mcp package

Python MCP server (fastmcp + httpx) that exposes the Pinchflat REST API as MCP tools. Covers sources, media items, media profiles, tasks, and settings endpoints. Tested against live Pinchflat instance — 28 sources, search, app info all working via stdio transport.
2026-07-04 22:05:12 +00:00
parent ac60ea8712
commit dc23a30013
3 changed files with 269 additions and 0 deletions
@@ -19,5 +19,6 @@
  annas-mcp = pkgs.callPackage ./annas-mcp/package.nix { };
  mcp-score = pkgs.callPackage ./mcp-score/package.nix { };
  applemusic-mcp = pkgs.callPackage ./applemusic-mcp/package.nix { };
+  pinchflat-mcp = pkgs.callPackage ./pinchflat-mcp/package.nix { };
  # example-mcp-server = pkgs.callPackage ./example-mcp-server/package.nix { };
 }
@@ -0,0 +1,42 @@
+{
+  lib,
+  stdenv,
+  makeWrapper,
+  python3,
+}:
+
+let
+  python = python3.withPackages (
+    ps: with ps; [
+      fastmcp
+      httpx
+    ]
+  );
+in
+stdenv.mkDerivation {
+  pname = "pinchflat-mcp";
+  version = "0.1.0";
+
+  src = ./server.py;
+
+  nativeBuildInputs = [ makeWrapper ];
+
+  dontUnpack = true;
+
+  installPhase = ''
+    mkdir -p $out/lib/pinchflat-mcp $out/bin
+    cp $src $out/lib/pinchflat-mcp/server.py
+
+    makeWrapper ${python}/bin/python $out/bin/pinchflat-mcp \
+      --add-flags "$out/lib/pinchflat-mcp/server.py" \
+      --set-default PINCHFLAT_MCP_TRANSPORT stdio
+  '';
+
+  meta = with lib; {
+    description = "MCP server for Pinchflat media downloader";
+    homepage = "https://gitea.bueso.eu/hermes-agent/pinchflat";
+    license = licenses.mit;
+    mainProgram = "pinchflat-mcp";
+    platforms = platforms.all;
+  };
+}
@@ -0,0 +1,226 @@
+#!/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) -> dict:
+    """List media items. Optionally filter by source_id."""
+    params = {}
+    if source_id:
+        params["source_id"] = source_id
+    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")
+
+
+# --- 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")