From dc23a30013472451e25fe6fe858dc34665c963bb Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sat, 4 Jul 2026 22:05:12 +0000 Subject: [PATCH] Add pinchflat-mcp package MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- packages/default.nix | 1 + packages/pinchflat-mcp/package.nix | 42 ++++++ packages/pinchflat-mcp/server.py | 226 +++++++++++++++++++++++++++++ 3 files changed, 269 insertions(+) create mode 100644 packages/pinchflat-mcp/package.nix create mode 100644 packages/pinchflat-mcp/server.py diff --git a/packages/default.nix b/packages/default.nix index c3a6a2f..6f189ac 100644 --- a/packages/default.nix +++ b/packages/default.nix @@ -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 { }; } diff --git a/packages/pinchflat-mcp/package.nix b/packages/pinchflat-mcp/package.nix new file mode 100644 index 0000000..9ba5615 --- /dev/null +++ b/packages/pinchflat-mcp/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; + }; +} \ No newline at end of file diff --git a/packages/pinchflat-mcp/server.py b/packages/pinchflat-mcp/server.py new file mode 100644 index 0000000..439a55b --- /dev/null +++ b/packages/pinchflat-mcp/server.py @@ -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") \ No newline at end of file -- 2.54.0