Expose the tool layer over MCP

README.md

@@ -360,6 +360,39 @@ SessionIQ works offline with a deterministic answer engine. To upgrade:
 Run `.\.venv\Scripts\python.exe scripts\check_local_ai.py` to see what's active and get setup hints.
 Models and vector indexes download to your machine's cache — they are **never committed**.
 
+### Optional: query your library from your own agent (MCP)
+
+SessionIQ speaks the [Model Context Protocol](https://modelcontextprotocol.io), so an MCP client —
+Claude Desktop, Cursor, or an agent of your own — can query your library directly.
+
+```powershell
+.\.venv\Scripts\python.exe -m pip install -e ".[mcp]"
+```
+
+Point the client at the server. For a client that takes a JSON config:
+
+```json
+{
+  "mcpServers": {
+    "sessioniq": {
+      "command": "/path/to/sessioniq/.venv/bin/python",
+      "args": ["/path/to/sessioniq/scripts/run_mcp.py"]
+    }
+  }
+}
+```
+
+On Windows the interpreter is `.venv\Scripts\python.exe`.
+
+It exposes the same seven tools the in-app assistant calls — `list_projects`, `search_library`,
+`filter_assets`, `compute_stat`, `asset_details`, `similar_tracks` and `next_up` — and delegates to
+the same code, so a question answered here and the same question asked in the app are computed
+identically. That logic is what `evals/` measures.
+
+Two things worth knowing. The server is **read-only**: it answers questions and changes nothing.
+And it reads the library at startup, so it sees the library as of when the client launched it —
+restart the client to pick up files added since.
+
 ## 📁 Project structure
 
 ```
@@ -378,6 +411,7 @@ sessioniq/
 │   ├── advisor.py          # reference A/B, finish-next ranking, weekly digest
 │   ├── transcription.py    # optional local Whisper voice-memo transcription
 │   ├── jobs.py             # in-process background jobs (progress, cancel)
+│   ├── mcp_server.py       # the tool layer exposed over MCP (optional extra)
 │   ├── plugins.py          # analyzer registry
 │   └── project_workspace.py# projects, smart collections, health, reports, file ops
 ├── web/src/                # React + TypeScript dashboard

pyproject.toml

@@ -26,9 +26,13 @@ dependencies = [
 
 [project.optional-dependencies]
 dev = [
+    "mcp>=2.2.0",
     "pytest>=8.2.0",
     "ruff>=0.5.0",
 ]
+mcp = [
+    "mcp>=2.2.0",
+]
 vector = [
     "chromadb>=0.5.0",
 ]

scripts/run_mcp.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+from sessioniq.mcp_server import LibraryUnavailable, run  # noqa: E402
+
+if __name__ == "__main__":
+    # stdout carries the protocol, so anything a human needs to read goes to
+    # stderr or it corrupts the stream.
+    try:
+        run()
+    except LibraryUnavailable as exc:
+        print(f"sessioniq mcp: {exc}", file=sys.stderr)
+        raise SystemExit(1) from exc
+    except ModuleNotFoundError as exc:
+        print(
+            f"sessioniq mcp: {exc}\n"
+            'Install the server with:  pip install -e ".[mcp]"',
+            file=sys.stderr,
+        )
+        raise SystemExit(1) from exc

src/sessioniq/mcp_server.py

@@ -0,0 +1,236 @@
+"""SessionIQ as a Model Context Protocol server.
+
+Exposes the same tool layer the in-app assistant calls, so any MCP client —
+Claude Desktop, Cursor, an agent of your own — can query a local SessionIQ
+library.
+
+    python scripts/run_mcp.py        # speaks MCP over stdio
+
+Every tool delegates to :class:`~sessioniq.tools.LibraryToolbox`. A question
+answered here and the same question asked inside the app are computed by
+identical code, and the evaluation in ``evals/`` covers that logic.
+
+Retrieval runs in memory rather than through ChromaDB. The MCP server is a
+read-only query surface, so it should start instantly and must not need an
+embedding model; the lexical and metadata blend it uses scored 0.733 hit@1
+against 0.767 with vectors, which is not worth a model download here.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+from sessioniq.models import ProjectAsset
+from sessioniq.project_workspace import UPLOAD_ROOT, summarize_projects
+from sessioniq.retrieval import InMemoryRetriever
+from sessioniq.tools import LibraryToolbox
+
+if TYPE_CHECKING:  # The SDK is an optional install; see the `mcp` extra.
+    from mcp.server.mcpserver import MCPServer
+
+LIBRARY_INDEX_NAME = "library-index.json"
+
+KINDS = ("audio", "midi", "note", "image", "reference", "export")
+STATUSES = ("Idea", "In Progress", "Needs Work", "Ready", "Reference", "Archived")
+STAT_FIELDS = ("bpm", "rms_db", "duration", "brightness", "note_count")
+STAT_OPS = ("min", "max", "avg", "count")
+
+
+class LibraryUnavailable(RuntimeError):
+    """Raised when there is no library for the server to answer from."""
+
+
+def library_index_path() -> Path:
+    return UPLOAD_ROOT.parent / LIBRARY_INDEX_NAME
+
+
+def load_library(index_path: Path | None = None) -> tuple[list[ProjectAsset], dict[str, str]]:
+    """Read the analyzed library from disk.
+
+    The API keeps this in memory; the MCP server is a separate process that
+    reads the same file at startup. It sees the library as of the moment it
+    launched, which is the right trade for a query surface that should never
+    block on a running app.
+    """
+    path = index_path or library_index_path()
+    if not path.exists():
+        raise LibraryUnavailable(
+            f"No library found at {path}. Run the app once, or "
+            "`python scripts/seed_demo.py`, to create one."
+        )
+    try:
+        raw = json.loads(path.read_bytes())
+    except (OSError, json.JSONDecodeError) as exc:
+        raise LibraryUnavailable(f"Could not read {path}: {exc}") from exc
+
+    assets: list[ProjectAsset] = []
+    for item in raw.get("assets", []):
+        try:
+            assets.append(ProjectAsset.model_validate(item))
+        except Exception:
+            # One unreadable entry must not take down the whole server; the
+            # app skips these on load too.
+            continue
+
+    statuses = {
+        str(key): str(value) for key, value in (raw.get("task_statuses") or {}).items()
+    }
+    return assets, statuses
+
+
+def build_toolbox(assets: list[ProjectAsset], statuses: dict[str, str]) -> LibraryToolbox:
+    """Assemble the toolbox exactly as the API does, minus the vector layer."""
+    summaries = summarize_projects(assets, statuses)
+    tasks = [task for project in summaries for task in project.tasks]
+
+    retriever = InMemoryRetriever()
+    retriever.add_assets(assets)
+
+    return LibraryToolbox(
+        assets,
+        tasks=tasks,
+        search=retriever.search,
+        summaries=summaries,
+    )
+
+
+def build_server(index_path: Path | None = None) -> MCPServer:
+    """Create the MCP server, loading the library up front."""
+    from mcp.server.mcpserver import MCPServer  # imported here: optional install
+
+    assets, statuses = load_library(index_path)
+    toolbox = build_toolbox(assets, statuses)
+
+    def call(tool: str, arguments: dict[str, Any]) -> str:
+        """Run a toolbox tool and return its JSON payload."""
+        return toolbox.execute(tool, arguments)
+
+    def arguments(**values: Any) -> dict[str, Any]:
+        """Drop unset options so the toolbox sees only what was asked for."""
+        return {key: value for key, value in values.items() if value is not None}
+
+    server = MCPServer(
+        "sessioniq",
+        instructions=(
+            "Query a local SessionIQ music library: analyzed audio, MIDI and session "
+            "notes organized into projects. Start with list_projects to learn the "
+            "project names, then filter_assets or compute_stat for exact questions "
+            "and search_library for open-ended ones."
+        ),
+    )
+
+    @server.tool()
+    def list_projects() -> str:
+        """List every project with its file counts and open task count.
+
+        Call this first: other tools accept a project_name, and this is how you
+        find the valid ones.
+        """
+        return call("list_projects", {})
+
+    @server.tool()
+    def search_library(query: str, project_name: str | None = None) -> str:
+        """Find files by meaning across the library.
+
+        Best for open-ended questions like "tracks that still need mastering".
+        For exact conditions use filter_assets, and for numbers use compute_stat.
+        """
+        return call("search_library", arguments(query=query, project_name=project_name))
+
+    @server.tool()
+    def filter_assets(
+        status: str | None = None,
+        kind: str | None = None,
+        tag: str | None = None,
+        key: str | None = None,
+        bpm_min: float | None = None,
+        bpm_max: float | None = None,
+        project_name: str | None = None,
+    ) -> str:
+        """List assets matching exact conditions, with the total match count.
+
+        Arguments are combined with AND. status is one of Idea, In Progress,
+        Needs Work, Ready, Reference, Archived. kind is one of audio, midi,
+        note, image. tag matches any part of a tag label. key is a pitch class
+        such as C or F#. The returned list may be truncated; total_matches
+        always reflects every match.
+        """
+        return call(
+            "filter_assets",
+            arguments(
+                status=status,
+                kind=kind,
+                tag=tag,
+                key=key,
+                bpm_min=bpm_min,
+                bpm_max=bpm_max,
+                project_name=project_name,
+            ),
+        )
+
+    @server.tool()
+    def compute_stat(
+        field: str,
+        op: str,
+        status: str | None = None,
+        kind: str | None = None,
+        tag: str | None = None,
+        key: str | None = None,
+        bpm_min: float | None = None,
+        bpm_max: float | None = None,
+        project_name: str | None = None,
+    ) -> str:
+        """Compute an exact aggregate over the library.
+
+        Use this for any count, average, or superlative — "how many tracks are
+        in C", "which track is the loudest". min and max also return the
+        winning asset. field is one of bpm, rms_db, duration, brightness,
+        note_count; op is one of min, max, avg, count. Filters behave as in
+        filter_assets.
+        """
+        return call(
+            "compute_stat",
+            arguments(
+                field=field,
+                op=op,
+                status=status,
+                kind=kind,
+                tag=tag,
+                key=key,
+                bpm_min=bpm_min,
+                bpm_max=bpm_max,
+                project_name=project_name,
+            ),
+        )
+
+    @server.tool()
+    def asset_details(asset_id: str) -> str:
+        """Full extracted metadata for one asset, including note text and tasks.
+
+        Use an asset_id returned by another tool.
+        """
+        return call("asset_details", {"asset_id": asset_id})
+
+    @server.tool()
+    def similar_tracks(asset_id: str, limit: int = 5) -> str:
+        """Tracks most similar to one asset, across tempo, key, tone and loudness."""
+        return call("similar_tracks", {"asset_id": asset_id, "limit": limit})
+
+    @server.tool()
+    def next_up() -> str:
+        """Rank projects by what is closest to finishable right now.
+
+        Blends readiness, task progress and freshness, and flags
+        near-finished-but-stale work as stalled. Use for "what should I
+        finish next?".
+        """
+        return call("next_up", {})
+
+    return server
+
+
+def run(index_path: Path | None = None) -> None:
+    """Serve over stdio until the client disconnects."""
+    build_server(index_path).run(transport="stdio")

src/sessioniq/tools.py

@@ -212,6 +212,39 @@ def _brief(asset: ProjectAsset) -> dict:
     }
 
 
+# Payload keys holding lists of results, in the order worth trimming.
+_TRUNCIBLE_KEYS = ("assets", "matches", "ranking", "projects")
+
+
+def _shrink_payload(result: object, limit: int) -> object:
+    """Trim a tool payload until it serialises under ``limit``.
+
+    Chopping the serialised string produced invalid JSON, which the model had
+    to cope with and an MCP client would simply fail to parse. Dropping whole
+    list items keeps the payload valid and still bounds its size.
+    """
+    if not isinstance(result, dict):
+        return result
+
+    candidate = result
+    while len(json.dumps(candidate, default=str)) > limit:
+        for key in _TRUNCIBLE_KEYS:
+            items = candidate.get(key)
+            if isinstance(items, list) and len(items) > 1:
+                candidate = {**candidate, key: items[: len(items) // 2], "truncated": True}
+                break
+        else:
+            shortened = {
+                key: (value[:400] if isinstance(value, str) and len(value) > 400 else value)
+                for key, value in candidate.items()
+            }
+            if shortened != candidate:
+                candidate = {**shortened, "truncated": True}
+                continue
+            return {"truncated": True, "error": "Result too large to return in full."}
+    return candidate
+
+
 class LibraryToolbox:
     """Executes tool calls against a snapshot of the library."""
 
@@ -254,10 +287,7 @@ def execute(self, name: str, arguments: dict) -> str:
             result = handler(arguments)
         except Exception as exc:  # One bad argument must not kill the answer.
             result = {"error": f"{type(exc).__name__}: {exc}"}
-        text = json.dumps(result, default=str)
-        if len(text) > MAX_TOOL_RESULT_CHARS:
-            text = text[:MAX_TOOL_RESULT_CHARS] + '…"}'
-        return text
+        return json.dumps(_shrink_payload(result, MAX_TOOL_RESULT_CHARS), default=str)
 
     # --- Filter shared by filter_assets and compute_stat -----------------