From 80d0352d1e3bb646cd4196d13146de884d3645c7 Mon Sep 17 00:00:00 2001 From: hermes-agent Date: Sun, 18 Oct 2026 15:18:18 +0200 Subject: [PATCH] pinchflat-mcp: redact credentials, gate admin tools, expose retention --- packages/pinchflat-mcp/server.py | 189 +++++++++++++++++++++++++------ 1 file changed, 152 insertions(+), 37 deletions(-) diff --git a/packages/pinchflat-mcp/server.py b/packages/pinchflat-mcp/server.py index 4d5f669..a310f00 100644 --- a/packages/pinchflat-mcp/server.py +++ b/packages/pinchflat-mcp/server.py @@ -1,5 +1,28 @@ #!/usr/bin/env python3 -"""Pinchflat MCP server — exposes the Pinchflat REST API as MCP tools.""" +"""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 @@ -20,13 +43,76 @@ HEADERS = {"Authorization": f"Bearer {PINCHFLAT_TOKEN}", "Content-Type": "applic 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 r.json() + return _unwrap(r) def _post(path: str, json: Optional[dict] = None) -> Any: @@ -34,7 +120,7 @@ def _post(path: str, json: Optional[dict] = None) -> Any: with httpx.Client(timeout=30) as client: r = client.post(f"{BASE}{path}", headers=HEADERS, json=json) r.raise_for_status() - return r.json() + return _unwrap(r) def _put(path: str, json: dict) -> Any: @@ -42,38 +128,40 @@ def _put(path: str, json: dict) -> Any: with httpx.Client(timeout=30) as client: r = client.put(f"{BASE}{path}", headers=HEADERS, json=json) r.raise_for_status() - return r.json() + return _unwrap(r) -def _delete(path: str) -> Any: +def _delete(path: str, params: Optional[dict] = None, timeout: int = 30) -> Any: """Make a DELETE request to the Pinchflat API.""" - with httpx.Client(timeout=30) as client: - r = client.delete(f"{BASE}{path}", headers=HEADERS) + with httpx.Client(timeout=timeout) as client: + r = client.delete(f"{BASE}{path}", headers=HEADERS, params=params) r.raise_for_status() - return r.json() + return _unwrap(r) # --- Sources --- -@mcp.tool() + +@tool() def list_sources() -> dict: """List all configured media sources (YouTube channels, playlists, etc.).""" return _get("/sources") -@mcp.tool() +@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() +@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). @@ -83,6 +171,8 @@ def create_source( 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, @@ -93,18 +183,30 @@ def create_source( 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}) -@mcp.tool() +@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.""" + """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 @@ -114,34 +216,46 @@ def update_source( 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}) -@mcp.tool() +@tool() def delete_source(source_id: int) -> dict: - """Delete a source (marks it for deletion).""" + """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}") -@mcp.tool() +@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() +@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() +@tool() def force_index(source_id: int) -> dict: """Force re-index of a source.""" return _post(f"/sources/{source_id}/force_index") -@mcp.tool() +@tool() def force_metadata_refresh(source_id: int) -> dict: """Force metadata refresh for a source.""" return _post(f"/sources/{source_id}/force_metadata_refresh") @@ -149,7 +263,8 @@ def force_metadata_refresh(source_id: int) -> dict: # --- Media Items --- -@mcp.tool() + +@tool() def list_media( source_id: Optional[int] = None, limit: int = 50, @@ -172,25 +287,25 @@ def list_media( return _get("/media", params=params) -@mcp.tool() +@tool() def search_media(query: str) -> dict: """Search media items by title or description.""" return _get("/media/search", params={"q": query}) -@mcp.tool() +@tool() def get_media(media_id: int) -> dict: """Get details of a specific media item.""" return _get(f"/media/{media_id}") -@mcp.tool() +@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() +@tool() def delete_media(media_id: int, prevent_download: bool = False) -> dict: """Delete a media item's downloaded files from disk. @@ -201,21 +316,19 @@ def delete_media(media_id: int, prevent_download: bool = False) -> dict: 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() + return _delete(f"/media/{media_id}", params=params, timeout=120) # --- Media Profiles --- -@mcp.tool() + +@tool() def list_media_profiles() -> dict: """List all media profiles (download quality/container settings).""" return _get("/media_profiles") -@mcp.tool() +@tool() def get_media_profile(profile_id: int) -> dict: """Get details of a specific media profile.""" return _get(f"/media_profiles/{profile_id}") @@ -223,7 +336,8 @@ def get_media_profile(profile_id: int) -> dict: # --- Tasks --- -@mcp.tool() + +@tool() def list_tasks(source_id: Optional[int] = None) -> dict: """List all tasks. Optionally filter by source_id.""" params = {} @@ -232,21 +346,22 @@ def list_tasks(source_id: Optional[int] = None) -> dict: return _get("/tasks", params=params) -@mcp.tool() +@tool() def get_task(task_id: int) -> dict: """Get details of a specific task.""" return _get(f"/tasks/{task_id}") -# --- Settings & Info --- +# --- Settings & Info (admin tier — not exposed by default) --- -@mcp.tool() + +@tool("admin") def get_settings() -> dict: - """Get current Pinchflat settings.""" + """Get current Pinchflat settings (credential values are redacted).""" return _get("/settings") -@mcp.tool() +@tool("admin") def get_app_info() -> dict: """Get Pinchflat app info (version, environment, etc.).""" return _get("/app_info") @@ -254,4 +369,4 @@ def get_app_info() -> dict: 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 + mcp.run(transport=transport if transport in ("stdio", "http") else "stdio") -- 2.55.0