cubelinux-2/cube-notes-mcp.py

#!/usr/bin/env python3
"""cube-notes-mcp — MCP (stdio) server exposing the CUBELinux-2 notes CLI as
model-facing tools `mcp__cubenotes__note_*` / `mcp__cubenotes__project_*`.

Wraps `cube note ...` / `cube project ...` (the durable WordFlags
message-save-path with categories + project threads). The harness spawns this
server via the `mcp-client` plugin; each tool call runs the `cube` binary,
which persists to `$CUBE_NOTES_DIR` (default ~/.cubelinux-notes).
"""
import sys, json, subprocess

CUBE = "/home/CUBELinux/CUBELinux-2/target/release/cube"
SERVER_NAME = "cube-notes-mcp"
SERVER_VERSION = "0.2.0"
PROTOCOL_VERSION = "2024-11-05"


def tool_schemas():
    return [
        {
            "name": "note_write",
            "description": "Add a note to the durable WordFlags message-save-path. "
                           "Optional: project P associates it to a project thread, "
                           "category C categorises it (doc_type note:C). Returns the coord.",
            "inputSchema": {"type": "object", "properties": {
                "text": {"type": "string"},
                "project": {"type": "string", "description": "Project name (thread)."},
                "category": {"type": "string", "description": "e.g. design, impl, cube."},
            }, "required": ["text"]},
        },
        {
            "name": "note_list",
            "description": "List notes (newest first), optionally filtered by session, project, category, since/until (YYYY-MM-DD).",
            "inputSchema": {"type": "object", "properties": {
                "session": {"type": "string"},
                "project": {"type": "string"},
                "category": {"type": "string"},
                "since": {"type": "string"},
                "until": {"type": "string"},
            }},
        },
        {
            "name": "note_search",
            "description": "Substring-search notes, optionally scoped to a project/category.",
            "inputSchema": {"type": "object", "properties": {
                "query": {"type": "string"},
                "project": {"type": "string"},
                "category": {"type": "string"},
            }, "required": ["query"]},
        },
        {
            "name": "note_show",
            "description": "Show one note by C.Z.Y.X.",
            "inputSchema": {"type": "object", "properties": {
                "coord": {"type": "string"},
            }, "required": ["coord"]},
        },
        {
            "name": "project_list",
            "description": "List all project threads (doc_type=project).",
            "inputSchema": {"type": "object", "properties": {}},
        },
        {
            "name": "project_show",
            "description": "Walk a project's associated notes (its thread/timeline) by C.Z.Y.X.",
            "inputSchema": {"type": "object", "properties": {
                "coord": {"type": "string"},
            }, "required": ["coord"]},
        },
    ]


def run_cube(args):
    try:
        r = subprocess.run([CUBE] + args, capture_output=True, text=True, timeout=30)
        out = (r.stdout or "").strip()
        if r.returncode != 0:
            return f"error: {(r.stderr or out).strip()}"
        return out if out else "(no result)"
    except Exception as e:  # noqa: BLE001
        return f"error: {e}"


def opt(flag, val):
    return [flag, str(val)] if val else []


def dispatch(name, args):
    args = args or {}
    if name == "note_write":
        text = (args.get("text") or "").strip()
        if not text:
            return "error: missing 'text'"
        return run_cube(["note", "add"] + opt("--project", args.get("project"))
                        + opt("--cat", args.get("category")) + [text])
    if name == "note_list":
        cmd = ["note", "list"]
        cmd += opt("--project", args.get("project"))
        cmd += opt("--cat", args.get("category"))
        cmd += opt("--since", args.get("since"))
        cmd += opt("--until", args.get("until"))
        if args.get("session"):
            cmd.append(args["session"])
        return run_cube(cmd)
    if name == "note_search":
        q = (args.get("query") or "").strip()
        if not q:
            return "error: missing 'query'"
        return run_cube(["note", "search", q] + opt("--project", args.get("project"))
                        + opt("--cat", args.get("category")))
    if name == "note_show":
        c = (args.get("coord") or "").strip()
        return run_cube(["note", "show", c]) if c else "error: missing 'coord'"
    if name == "project_list":
        return run_cube(["project", "list"])
    if name == "project_show":
        c = (args.get("coord") or "").strip()
        return run_cube(["project", "show", c]) if c else "error: missing 'coord'"
    return f"error: unknown tool {name}"


def handle(msg):
    if not isinstance(msg, dict):
        return None
    mid, method = msg.get("id"), msg.get("method")
    if method == "initialize":
        return {"jsonrpc": "2.0", "id": mid, "result": {
            "protocolVersion": PROTOCOL_VERSION,
            "capabilities": {"tools": {}},
            "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION},
        }}
    if method == "ping":
        return {"jsonrpc": "2.0", "id": mid, "result": {}}
    if method in ("notifications/initialized", "initialized"):
        return None
    if method == "tools/list":
        return {"jsonrpc": "2.0", "id": mid, "result": {"tools": tool_schemas()}}
    if method == "tools/call":
        name = msg.get("params", {}).get("name", "")
        args = msg.get("params", {}).get("arguments", {})
        return {"jsonrpc": "2.0", "id": mid, "result": {
            "content": [{"type": "text", "text": dispatch(name, args)}], "isError": False}}
    if mid is not None:
        return {"jsonrpc": "2.0", "id": mid, "result": {}}
    return None


for line in sys.stdin:
    line = line.strip()
    if not line:
        continue
    try:
        msg = json.loads(line)
    except Exception:  # noqa: BLE001
        continue
    resp = handle(msg)
    if resp is not None:
        sys.stdout.write(json.dumps(resp) + "\n")
        sys.stdout.flush()