Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 8dc25624ff | |||
| 676d06f8ca | |||
| 4856b98429 | |||
| d009717854 | |||
| 617bb9cab5 | |||
| 15906893c0 | |||
| dc23a30013 |
@@ -27,3 +27,4 @@ result-*
|
||||
# Local environment overrides
|
||||
.envrc.local
|
||||
/.dir-locals.el
|
||||
__pycache__/
|
||||
|
||||
@@ -2,28 +2,35 @@
|
||||
lib,
|
||||
buildGoModule,
|
||||
fetchFromGitHub,
|
||||
makeWrapper,
|
||||
}:
|
||||
|
||||
buildGoModule rec {
|
||||
pname = "annas-mcp";
|
||||
version = "0.0.5";
|
||||
version = "0.1.0-unstable-2026-06-26";
|
||||
|
||||
src = fetchFromGitHub {
|
||||
owner = "iosifache";
|
||||
repo = "annas-mcp";
|
||||
rev = "v${version}";
|
||||
hash = "sha256-XicM7tU5jD8B8n7JJDQ/84koBiLb8XF4+WBQ4LCUoRU=";
|
||||
rev = "c095963ad5d10d33da3c614495246b6ec7f680d4";
|
||||
hash = lib.fakeHash;
|
||||
};
|
||||
|
||||
vendorHash = "sha256-2NdG5p2XfrhVgi388dRDBUSGwg6ybnzfn9495TWNGsA=";
|
||||
vendorHash = lib.fakeHash;
|
||||
|
||||
subPackages = [ "cmd/annas-mcp" ];
|
||||
nativeBuildInputs = [ makeWrapper ];
|
||||
|
||||
# The MCP server mode is invoked with "mcp" subcommand
|
||||
postInstall = ''
|
||||
wrapProgram $out/bin/annas-mcp \
|
||||
--add-flags "mcp"
|
||||
'';
|
||||
|
||||
meta = with lib; {
|
||||
description = "MCP server and CLI for searching and downloading documents from Anna's Archive";
|
||||
description = "MCP server and CLI tool for searching and downloading documents from Anna's Archive";
|
||||
homepage = "https://github.com/iosifache/annas-mcp";
|
||||
license = licenses.mit;
|
||||
mainProgram = "annas-mcp";
|
||||
platforms = platforms.all;
|
||||
platforms = platforms.linux;
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
{
|
||||
lib,
|
||||
buildGoModule,
|
||||
fetchFromGitHub,
|
||||
}:
|
||||
|
||||
buildGoModule rec {
|
||||
pname = "audiobookshelf-mcp";
|
||||
version = "unstable-2026-06-27";
|
||||
|
||||
src = fetchFromGitHub {
|
||||
owner = "michaeldvinci";
|
||||
repo = "audiobookshelf-mcp";
|
||||
rev = "e790d9c75085772180c95f2ec1376acc98e9282c";
|
||||
hash = "sha256-vQR+Ej3Qtg8ehIUWnXkSayNYpqnOZPWV0TpTVPYh+io=";
|
||||
};
|
||||
|
||||
vendorHash = "sha256-W8hlCGf4QdFbKc3QFc9pa4MWBhnp5A5GvWFNzg0BEhw=";
|
||||
|
||||
meta = with lib; {
|
||||
description = "MCP server for Audiobookshelf — libraries, audiobooks, podcasts, authors, collections, playlists";
|
||||
homepage = "https://github.com/michaeldvinci/audiobookshelf-mcp";
|
||||
license = licenses.gpl3Only;
|
||||
mainProgram = "abs-mcp";
|
||||
platforms = platforms.all;
|
||||
};
|
||||
}
|
||||
@@ -19,5 +19,7 @@
|
||||
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 { };
|
||||
audiobookshelf-mcp = pkgs.callPackage ./audiobookshelf-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,257 @@
|
||||
#!/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")
|
||||
Reference in New Issue
Block a user