7 Commits

Author SHA1 Message Date
hermes-agent 8dc25624ff add: annas-mcp package (Go, fakeHash placeholders)
NixOS Configuration CI / Nix Flake Check (pull_request) Failing after 20s
2026-07-12 18:58:19 +00:00
hermes-agent 676d06f8ca audiobookshelf-mcp: fix vendorHash (from CI)
NixOS Configuration CI / Nix Flake Check (pull_request) Successful in 46s
2026-07-12 11:16:46 +00:00
hermes-agent 4856b98429 fix: remove committed pyc, add to gitignore
NixOS Configuration CI / Nix Flake Check (pull_request) Failing after 14s
2026-07-12 11:14:52 +00:00
hermes-agent d009717854 audiobookshelf-mcp: fix src hash (from CI), still need vendorHash
NixOS Configuration CI / Nix Flake Check (pull_request) Failing after 14s
2026-07-12 11:14:43 +00:00
hermes-agent 617bb9cab5 pinchflat-mcp: add pagination, downloaded filter, and delete_media tool
NixOS Configuration CI / Nix Flake Check (pull_request) Failing after 21s
- list_media: add offset and downloaded params for API pagination
- delete_media: new tool to delete downloaded files (with prevent_download option)
- Bump timeout on delete to 120s
2026-07-12 11:09:19 +00:00
hermes-agent 15906893c0 Add audiobookshelf-mcp package
Go binary from michaeldvinci/audiobookshelf-mcp (v0.0.5).
Covers libraries, audiobooks, podcasts, authors, collections,
playlists, and users. Uses buildGoModule with fakeHash placeholders
to be filled on first nix build.
2026-07-04 22:24:46 +00:00
hermes-agent dc23a30013 Add pinchflat-mcp package
NixOS Configuration CI / Nix Flake Check (pull_request) Successful in 26s
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
6 changed files with 344 additions and 8 deletions
+1
View File
@@ -27,3 +27,4 @@ result-*
# Local environment overrides
.envrc.local
/.dir-locals.el
__pycache__/
+15 -8
View File
@@ -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;
};
}
}
+27
View File
@@ -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;
};
}
+2
View File
@@ -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 { };
}
+42
View File
@@ -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;
};
}
+257
View File
@@ -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")