replica-omnisciente/.hermes/plans/2026-07-14_180000-muscriptor-mcp-weekly-midi.md

MuScriptor MCP Server + Weekly MIDI Pipeline

For Hermes: Use subagent-driven-development skill to implement this plan task-by-task.

Goal: Build a MuScriptor-powered MCP server that agents (Hermes, Aurelio) can call to transcribe audio → MIDI, and a weekly cron that auto-converts the user's top 3 tracks from their Navidrome/Maloja music stack.

Architecture: Python MCP server exposing transcribe_to_midi, get_top_tracks, search_and_transcribe tools. Backed by MuScriptor (Kyutai/Mirelo, 1B param music transcription transformer). Connects to existing Maloja (stats) + Navidrome (audio source) via their APIs. Deployed as a systemd service on Gigabyte (RTX 3070 GPUs) or Tomahawk MAX. Weekly Hermes cron job queries top tracks → downloads → transcribes → delivers MIDI.

Tech Stack: Python 3.12+, MuScriptor (pip install muscriptor), MCP Python SDK, Maloja API, Navidrome Subsonic API, Hermes cron.

---

Context

Existing Music Infrastructure

Service	URL	Port	Host	Purpose
Navidrome	music.portugalfuturista.org	4533	Tomahawk MAX (Gigabyte) Docker	Music library + Subsonic API
Maloja	maloja.portugalfuturista.org	42010	CT 216	Scrobble statistics
Multi-scrobbler	scrobbler.portugalfuturista.org	9078	CT 216	Aggregates from Jellyfin, Navidrome, YTMusic → Maloja + Last.fm
Jellyfin	—	8096	Gigabyte Docker	Media server (also scrobbles)

MuScriptor Model (Kyutai + Mirelo)

• Repo: github.com/muscriptor/muscriptor (MIT license)
• Models: muscriptor-small (103M), muscriptor-medium (307M, default), muscriptor-large (1.4B)
• License: CC BY-NC 4.0 (requires free HF account + token)
• Install: pip install muscriptor or uvx muscriptor
• Key API: model.transcribe_to_midi("audio.wav") → MIDI bytes
• GPU: medium runs fine on RTX 3070 (8GB); large needs ~12GB VRAM
• Supported instruments: piano, drums, guitar, bass, strings, winds, etc.

How It Connects to Aurelio / Hermes

• MCP server registered in .aurelio/mcp_config.json → available to all IDE agents
• Hermes cron job calls the MCP tools weekly
• User can also invoke directly: "Hermes, convert Sun by Caribou to MIDI"

---

Implementation Plan

Phase 1: MCP Server Core (muscriptIdor-mcp)

Task 1: Create MCP server skeleton

Objective: Scaffold the Python MCP server project structure.

Files:

• Create: scripts/muscriptor-mcp/pyproject.toml
• Create: scripts/muscriptor-mcp/src/muscriptor_mcp/__init__.py
• Create: scripts/muscriptor-mcp/src/muscriptor_mcp/server.py

Step 1: Create project with pyproject.toml:

[project]
name = "muscriptor-mcp"
version = "0.1.0"
description = "MCP server for MuScriptor music transcription (audio → MIDI)"
requires-python = ">=3.10"
dependencies = [
    "mcp[cli]>=1.0.0",
    "muscriptor>=0.1.0",
    "httpx>=0.27",
    "pydantic>=2.0",
]

[project.scripts]
muscriptor-mcp = "muscriptor_mcp.server:main"

Step 2: Create the MCP server skeleton with server.py:

"""MuScriptor MCP Server — audio → MIDI transcription for Aurelio agents."""
import asyncio
import logging
from pathlib import Path
from mcp.server import Server
from mcp.server.stdio import run_server
from mcp.types import Tool, TextContent

logger = logging.getLogger(__name__)
app = Server("muscriptor-mcp")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="transcribe_to_midi",
            description="Transcribe an audio file (WAV/MP3/FLAC/OGG) to MIDI using MuScriptor. "
                        "Returns the path to the generated .mid file.",
            inputSchema={
                "type": "object",
                "properties": {
                    "audio_path": {"type": "string", "description": "Path to audio file"},
                    "output_dir": {"type": "string", "description": "Output directory for MIDI (default: same as input)"},
                    "model_size": {"type": "string", "enum": ["small", "medium", "large"], "default": "medium"},
                    "instruments": {"type": "array", "items": {"type": "string"}, "description": "Restrict to specific instruments"},
                },
                "required": ["audio_path"],
            },
        ),
        Tool(
            name="get_top_tracks",
            description="Get the user's most-played tracks from Maloja scrobble stats.",
            inputSchema={
                "type": "object",
                "properties": {
                    "count": {"type": "integer", "default": 3, "description": "Number of top tracks"},
                    "period": {"type": "string", "default": "week", "description": "Time period: day, week, month, year, overall"},
                },
            },
        ),
        Tool(
            name="search_and_download",
            description="Search Navidrome for a track and download the audio file locally.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query (e.g. 'Sun Caribou')"},
                    "download_dir": {"type": "string", "description": "Directory to save audio (default: /tmp/muscriptor)"},
                },
                "required": ["query"],
            },
        ),
        Tool(
            name="top_tracks_to_midi",
            description="Full pipeline: fetch top N tracks from Maloja, download from Navidrome, "
                        "transcribe each to MIDI. Returns paths to all generated .mid files.",
            inputSchema={
                "type": "object",
                "properties": {
                    "count": {"type": "integer", "default": 3},
                    "period": {"type": "string", "default": "week"},
                    "model_size": {"type": "string", "enum": ["small", "medium", "large"], "default": "medium"},
                },
            },
        ),
    ]

async def main():
    async with run_server(app) as server:
        await server.serve_forever()

if __name__ == "__main__":
    asyncio.run(main())

Verification: cd scripts/muscriptor-mcp && pip install -e . && python -c "from muscriptor_mcp.server import app; print('OK')"

---

Task 2: Implement Maloja API client (get_top_tracks)

Objective: Query Maloja for top tracks with play counts.

Files:

• Create: scripts/muscriptor-mcp/src/muscriptor_mcp/maloja_client.py

Step 1: Maloja exposes a REST API. Key endpoints:

GET /api/v1/top/tracks?period=week&max=3
Authorization: Token <api_key>

Response shape (from Maloja docs):

{
  "list": [
    {
      "track": {"title": "Sun", "artists": [{"name": "Caribou"}]},
      "amount": 12
    }
  ]
}

Step 2: Implement the client:

"""Maloja scrobble statistics client."""
import httpx
from dataclasses import dataclass

MALOJA_URL = "http://pf-maloja:42010"  # LAN — also accessible via maloja.portugalfuturista.org
MALOJA_API_KEY = ""  # TODO: load from env MALOJA_API_KEY

@dataclass
class TrackStat:
    title: str
    artist: str
    play_count: int

async def get_top_tracks(count: int = 3, period: str = "week") -> list[TrackStat]:
    """Fetch top tracks from Maloja API."""
    url = f"{MALOJA_URL}/api/v1/top/tracks"
    params = {"period": period, "max": count}
    headers = {"Authorization": f"Token {MALOJA_API_KEY}"}
    async with httpx.AsyncClient() as client:
        resp = await client.get(url, params=params, headers=headers, timeout=15)
        resp.raise_for_status()
        data = resp.json()
    return [
        TrackStat(
            title=item["track"]["title"],
            artist=item["track"]["artists"][0]["name"],
            play_count=item["amount"],
        )
        for item in data["list"]
    ]

Verification: Test with curl against live Maloja first to confirm API shape, then unit test with mocked response.

---

Task 3: Implement Navidrome Subsonic API client (search_and_download)

Objective: Search Navidrome and download audio files.

Files:

• Create: scripts/muscriptor-mcp/src/muscriptor_mcp/navidrome_client.py

Step 1: Navidrome exposes the Subsonic API. Key endpoints:

GET /rest/search3.view?query=Sun+Caribou&u=fabio&t=<token>&s=<salt>&v=1.16.1&c=muscriptor
GET /rest/download.view?id=<songId>&u=fabio&t=<token>&s=<salt>&v=1.16.1&c=muscriptor

Subsonic auth: token = md5(password + salt).

Step 2: Implement:

"""Navidrome (Subsonic API) client for audio download."""
import hashlib
import os
import secrets
from pathlib import Path
import httpx

NAVIDROME_URL = "http://pf-navidrome:4533"  # LAN
NAVIDROME_USER = ""  # TODO: load from env
NAVIDROME_PASSWORD = ""  # TODO: load from env

def _subsonic_params() -> dict:
    salt = secrets.token_hex(8)
    token = hashlib.md5((NAVIDROME_PASSWORD + salt).encode()).hexdigest()
    return {"u": NAVIDROME_USER, "t": token, "s": salt, "v": "1.16.1", "c": "muscriptor-mcp"}

async def search_track(query: str) -> list[dict]:
    """Search Navidrome for tracks matching query. Returns list of {id, title, artist, duration}."""
    params = {**_subsonic_params(), "query": query, "songCount": 5}
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{NAVIDROME_URL}/rest/search3.view", params=params, timeout=15)
        resp.raise_for_status()
        data = resp.json()
    songs = data.get("subsonic-response", {}).get("searchResult3", {}).get("song", [])
    return [{"id": s["id"], "title": s["title"], "artist": s.get("artist", ""), "duration": s.get("duration", 0)} for s in songs]

async def download_track(song_id: str, output_dir: Path) -> Path:
    """Download a track by ID. Returns local file path."""
    output_dir.mkdir(parents=True, exist_ok=True)
    params = {**_subsonic_params(), "id": song_id}
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{NAVIDROME_URL}/rest/download.view", params=params, timeout=120)
        resp.raise_for_status()
        # Navidrome sends the file with its original extension
        content_type = resp.headers.get("content-type", "audio/mpeg")
        ext = _ext_from_content_type(content_type)
        out_path = output_dir / f"{song_id}{ext}"
        out_path.write_bytes(resp.content)
    return out_path

def _ext_from_content_type(ct: str) -> str:
    mapping = {
        "audio/mpeg": ".mp3", "audio/flac": ".flac", "audio/ogg": ".ogg",
        "audio/wav": ".wav", "audio/x-wav": ".wav", "audio/mp4": ".m4a",
    }
    return mapping.get(ct.split(";")[0].strip(), ".mp3")

Verification: Test search with curl against live Navidrome, confirm Subsonic auth works.

---

Task 4: Implement MuScriptor transcription engine

Objective: Wrap MuScriptor's transcribe_to_midi in an async wrapper.

Files:

• Create: scripts/muscriptor-mcp/src/muscriptor_mcp/transcriber.py

Step 1:

"""MuScriptor audio → MIDI transcription engine."""
import asyncio
from pathlib import Path
from functools import lru_cache

@lru_cache(maxsize=1)
def _load_model(size: str = "medium"):
    """Load and cache the MuScriptor model (singleton per size)."""
    from muscriptor import TranscriptionModel
    return TranscriptionModel.load_model(size)

async def transcribe(audio_path: str | Path, output_path: str | Path | None = None,
                     model_size: str = "medium", instruments: list[str] | None = None) -> Path:
    """Transcribe audio file to MIDI. Returns path to .mid file."""
    audio_path = Path(audio_path)
    if output_path is None:
        output_path = audio_path.with_suffix(".mid")
    else:
        output_path = Path(output_path)

    model = _load_model(model_size)

    # MuScriptor's transcribe is CPU/GPU-bound, run in thread pool
    loop = asyncio.get_event_loop()
    midi_bytes = await loop.run_in_executor(None, lambda: model.transcribe_to_midi(str(audio_path)))
    output_path.write_bytes(midi_bytes)
    return output_path

Verification: Run on a short WAV file to confirm model downloads + produces valid MIDI.

---

Task 5: Wire tools into MCP server

Objective: Connect all modules into the MCP server's tool handlers.

Files:

• Modify: scripts/muscriptor-mcp/src/muscriptor_mcp/server.py — add @app.call_tool() handler

Step 1: Add the tool dispatch to server.py:

from muscriptor_mcp.maloja_client import get_top_tracks
from muscriptor_mcp.navidrome_client import search_track, download_track
from muscriptor_mcp.transcriber import transcribe

DEFAULT_DOWNLOAD_DIR = Path("/tmp/muscriptor")

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "transcribe_to_midi":
        audio_path = arguments["audio_path"]
        output_dir = arguments.get("output_dir")
        model_size = arguments.get("model_size", "medium")
        instruments = arguments.get("instruments")
        out = Path(output_dir) / Path(audio_path).with_suffix(".mid").name if output_dir else None
        result = await transcribe(audio_path, out, model_size, instruments)
        return [TextContent(type="text", text=f"MIDI saved to: {result}")]

    elif name == "get_top_tracks":
        tracks = await get_top_tracks(arguments.get("count", 3), arguments.get("period", "week"))
        lines = [f"#{i+1}: {t.artist} — {t.title} ({t.play_count} plays)" for i, t in enumerate(tracks)]
        return [TextContent(type="text", text="\n".join(lines))]

    elif name == "search_and_download":
        results = await search_track(arguments["query"])
        if not results:
            return [TextContent(type="text", text=f"No results for '{arguments['query']}'")]
        track = results[0]
        dl_dir = Path(arguments.get("download_dir", str(DEFAULT_DOWNLOAD_DIR)))
        path = await download_track(track["id"], dl_dir)
        return [TextContent(type="text", text=f"Downloaded: {track['artist']} — {track['title']} → {path}")]

    elif name == "top_tracks_to_midi":
        count = arguments.get("count", 3)
        period = arguments.get("period", "week")
        model_size = arguments.get("model_size", "medium")
        tracks = await get_top_tracks(count, period)
        results = []
        dl_dir = DEFAULT_DOWNLOAD_DIR / "weekly"
        for t in tracks:
            query = f"{t.title} {t.artist}"
            search_results = await search_track(query)
            if not search_results:
                results.append(f"SKIP: {t.artist} — {t.title} (not found in Navidrome)")
                continue
            audio_path = await download_track(search_results[0]["id"], dl_dir)
            midi_path = await transcribe(audio_path, model_size=model_size)
            results.append(f"DONE: {t.artist} — {t.title} → {midi_path}")
        return [TextContent(type="text", text="\n".join(results))]

    else:
        return [TextContent(type="text", text=f"Unknown tool: {name}")]

Verification: cd scripts/muscriptor-mcp && python -m muscriptor_mcp.server starts without errors.

---

Phase 2: Configuration & Secrets

Task 6: Environment-based secrets

Objective: Load API keys from environment variables, never hardcode.

Files:

• Create: scripts/muscriptor-mcp/.env.example
• Modify: scripts/muscriptor-mcp/src/muscriptor_mcp/maloja_client.py — read from env
• Modify: scripts/muscriptor-mcp/src/muscriptor_mcp/navidrome_client.py — read from env

.env.example:

MALOJA_URL=http://pf-maloja:42010
MALOJA_API_KEY=your_maloja_api_key
NAVIDROME_URL=http://pf-navidrome:4533
NAVIDROME_USER=fabio
NAVIDROME_PASSWORD=your_navidrome_password
HF_TOKEN=your_huggingface_token

Verification: Start server without .env → graceful error. Start with .env → tools work.

---

Task 7: Register MCP server in Aurelio config

Objective: Make the server discoverable by Aurelio agents.

Files:

• Modify: .aurelio/mcp_config.json — add muscriptor-mcp entry

Add entry:

"muscriptor-mcp": {
  "command": "python",
  "args": ["-m", "muscriptor_mcp.server"],
  "env": {
    "MALOJA_URL": "http://pf-maloja:42010",
    "NAVIDROME_URL": "http://pf-navidrome:4533"
  },
  "_disabled": false,
  "disabledTools": []
}

Note: secrets (API keys, passwords) go in the env file, not in mcp_config.json. The MCP server reads them from its own environment.

---

Phase 3: Deployment on GPU Node

Task 8: Create systemd service for Gigabyte / Tomahawk MAX

Objective: Run the MCP server as a persistent service on the GPU node.

Files:

• Create: scripts/muscriptor-mcp/systemd/muscriptor-mcp.service

Systemd unit:

[Unit]
Description=MuScriptor MCP Server (audio → MIDI)
After=network.target docker.service

[Service]
Type=simple
User=fabio
WorkingDirectory=/home/fabio/muscriptor-mcp
EnvironmentFile=/home/fabio/muscriptor-mcp/.env
ExecStart=/home/fabio/muscriptor-mcp/.venv/bin/python -m muscriptor_mcp.server
Restart=on-failure
RestartSec=5
# GPU access
SupplementaryGroups=video render

[Install]
WantedBy=multi-user.target

Deployment steps:

1. Clone repo to Gigabyte node (192.168.0.104)
2. cd scripts/muscriptor-mcp && python -m venv .venv && . .venv/bin/activate && pip install -e .
3. Create .env with real secrets
4. Copy service file, enable, start
5. First run will download ~1.2GB model weights (cached after)

---

Task 9: Expose via Cloudflare tunnel (optional)

Objective: Make MCP server accessible from Hermes gateway (CT 217) and laptop.

Option A: SSH tunnel from CT 217 to Gigabyte (preferred, no public exposure). Option B: Add Cloudflare tunnel route for muscriptor-mcp.portugalfuturista.org.

---

Phase 4: Weekly Cron Job

Task 10: Hermes cron job for weekly top-3 MIDI conversion

Objective: Every Monday morning, auto-convert top 3 weekly tracks to MIDI.

Using Hermes cron:

Schedule: 0 9 * * 1  (every Monday at 09:00)
Prompt: "Run the muscriptor-mcp top_tracks_to_midi tool with count=3, period=week, model_size=medium. 
         Report which tracks were converted and where the MIDI files are saved."
Deliver: telegram (or wherever the user wants)

Or via the MCP tool directly from any agent:

"Hermes, convert my top 3 tracks this week to MIDI"

---

Phase 5: Interactive Usage

Task 11: Document agent usage patterns

Objective: Show how users interact with the transcription tools.

Usage patterns:

1. Ad-hoc transcription:

   "Hermes, here's Sun by Caribou. Convert it to MIDI for Ableton." → Agent calls search_and_download then transcribe_to_midi

2. Weekly auto-conversion: → Cron job runs top_tracks_to_midi automatically

3. From a file path:

   "Transcribe /tmp/my_recording.wav to MIDI" → Agent calls transcribe_to_midi directly

4. With instrument restriction:

   "Transcribe just the piano from this track" → Agent calls transcribe_to_midi with instruments=["acoustic_piano"]

5. Batch conversion:

   "Convert all tracks from this album to MIDI" → Agent searches Navidrome, loops transcribe_to_midi

---

Files Summary

Action	Path
Create	scripts/muscriptor-mcp/pyproject.toml
Create	scripts/muscriptor-mcp/src/muscriptor_mcp/__init__.py
Create	scripts/muscriptor-mcp/src/muscriptor_mcp/server.py
Create	scripts/muscriptor-mcp/src/muscriptor_mcp/maloja_client.py
Create	scripts/muscriptor-mcp/src/muscriptor_mcp/navidrome_client.py
Create	scripts/muscriptor-mcp/src/muscriptor_mcp/transcriber.py
Create	scripts/muscriptor-mcp/.env.example
Create	scripts/muscriptor-mcp/systemd/muscriptor-mcp.service
Modify	.aurelio/mcp_config.json (add muscriptor-mcp entry)

Risks & Open Questions

1. GPU VRAM: MuScriptor medium (307M) should fit in RTX 3070 8GB. Large (1.4B) needs ~12GB — may OOM on 8GB cards. Use medium by default.
2. Audio format compatibility: MuScriptor supports WAV natively. For MP3/FLAC/OGG, may need ffmpeg or soundfile for pre-conversion. Navidrome sends original format.
3. Maloja API shape: Need to verify the exact API response structure — Maloja docs are sparse. Test with live instance first.
4. Navidrome auth: Subsonic API uses token+salt auth. The multi-scrobbler config has credentials but they should go in .env, not committed.
5. Network path: MCP server on Gigabyte needs to reach Maloja on CT 216 (192.168.0.126:42010) and Navidrome on Gigabyte Docker (pf-navidrome:4533). Verify LAN connectivity.
6. Model download: First run downloads ~1.2GB from HuggingFace. Requires HF_TOKEN env var (free HF account).
7. Hermes cron limitation: Cron jobs in TUI are local-only. For the weekly job to work reliably, either run it from the Hermes gateway on CT 217, or use a systemd timer on the GPU node.

Dependencies to Install

pip install muscriptor mcp[cli] httpx pydantic

MuScriptor pulls in: torch, torchaudio, transformers, safetensors, soundfile.