cube-notes-mcp: stdio MCP server exposing the cubesys::notes CLI (cube note add|list|search|show) as model-facing mcp__cubenotes__note_* tools for the DeepSeek Harness <-> CUBE session message path.

cube-notes-mcp.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""cube-notes-mcp — MCP (stdio) server exposing the CUBELinux-2 notes CLI as
+model-facing tools `mcp__cubenotes__note_*`.
+
+Wraps `cube note ...` (the durable WordFlags message-save-path) so the agent
+can write / list / search / show session notes from the harness. The harness
+spawns this server via the `mcp-client` plugin (see the profile's
+cordis.patch.yml); 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.1.0"
+PROTOCOL_VERSION = "2024-11-05"
+
+
+def tool_schemas():
+    return [
+        {
+            "name": "note_write",
+            "description": "Add a session note to the durable WordFlags CUBE message-save-path "
+                           "(per-session/per-day, searchable). Returns the note's C.Z.Y.X coord.",
+            "inputSchema": {"type": "object", "properties": {
+                "text": {"type": "string", "description": "Note text (subject + body)."}
+            }, "required": ["text"]},
+        },
+        {
+            "name": "note_list",
+            "description": "List notes (newest first). Optionally filter by session label "
+                           "(default: all), or by a date.",
+            "inputSchema": {"type": "object", "properties": {
+                "session": {"type": "string", "description": "Session label to filter by."},
+            }},
+        },
+        {
+            "name": "note_search",
+            "description": "Substring-search notes across sessions (case-insensitive).",
+            "inputSchema": {"type": "object", "properties": {
+                "query": {"type": "string", "description": "Substring to find."},
+            }, "required": ["query"]},
+        },
+        {
+            "name": "note_show",
+            "description": "Show one note's subject + body by its C.Z.Y.X coordinate.",
+            "inputSchema": {"type": "object", "properties": {
+                "coord": {"type": "string", "description": "Note coordinate (C.Z.Y.X)."},
+            }, "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 dispatch(name, args):
+    args = args or {}
+    if name == "note_write":
+        text = (args.get("text") or "").strip()
+        return run_cube(["note", "add", text]) if text else "error: missing 'text'"
+    if name == "note_list":
+        session = args.get("session")
+        return run_cube(["note", "list", session] if session else ["note", "list"])
+    if name == "note_search":
+        q = (args.get("query") or "").strip()
+        return run_cube(["note", "search", q]) if q else "error: missing 'query'"
+    if name == "note_show":
+        c = (args.get("coord") or "").strip()
+        return run_cube(["note", "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()