"""CLI and stdio MCP tools for the authenticated Nore HTTPS API.""" import argparse import base64 import json import os import re import sys from pathlib import Path from urllib.error import HTTPError from urllib.parse import urlsplit from urllib.request import Request, urlopen from uuid import uuid4 FIELDS = { "tenant": "Tenant ID", "project": "Project ID", "key": "Document key", "hash": "Asset SHA-256", "frame": "Frame ID from the project manifest", "cue": "Sound cue ID from nore_sounds", "after": "Last received event sequence", "title": "Proposal title", "text": "Proposed direction", "base_revision": "Revision you read", "value": "JSON document value", } DEFS = { "nore_projects": ("List accessible projects and your contributor identity.", []), "nore_frames": ( "Read the baseline and candidates for a project.", ["tenant", "project"], ), "nore_image": ( "Read one authorized project preview as image pixels.", ["tenant", "project", "hash"], ), "nore_documents": ( "List project contributions and revisions.", ["tenant", "project"], ), "nore_read": ("Read a versioned project document.", ["tenant", "project", "key"]), "nore_directions": ( "Read submitted directions and your current choice for a frame.", ["tenant", "project", "frame"], ), "nore_suggest": ( "Publish an immutable direction. value needs frame_sha256, title (64 chars), change (180; concrete on-screen or audible direction), why (300; for the reviewer), listen_for (180; reviewer check) and cue from nore_sounds.", ["tenant", "project", "frame", "value"], ), "nore_motion_prompt": ( "Read the planned or stored motion prompt, separate reviewer check, and frame hash for a suggestion. Does not request a render.", ["tenant", "project", "key"], ), "nore_prompt_bench": ( "Read the verified source still prompt, changes since the previous displayed frame, model, seed, and saved variants. Missing text is null.", ["tenant", "project", "frame"], ), "nore_prompt_variant": ( "Save a single-phrase still prompt variant without generating. Supply frame_sha256, prompt_sha256 from source, exact unique before, after (empty to delete), and a visible check for the reviewer.", ["tenant", "project", "frame", "value"], ), "nore_sounds": ( "List short audio sketches, concepts and listening exercises.", ["tenant", "project"], ), "nore_listen": ("Fetch a sound sketch as WAV audio.", ["tenant", "project", "cue"]), "nore_evals": ( "Read human choices by contributor, with blind pairs, named pairs and solo trials counted separately.", ["tenant", "project"], ), "nore_events": ( "Read project changes after a sequence number.", ["tenant", "project", "after"], ), "nore_propose": ( "Create a separate creative proposal.", ["tenant", "project", "title", "text"], ), "nore_write": ( "Save an existing or new document at an explicit base revision.", ["tenant", "project", "key", "base_revision", "value"], ), } def tools_list(): return [ { "name": k, "description": desc, "inputSchema": { "type": "object", "properties": { f: { "type": "integer" if f in ("after", "base_revision") else "object" if f == "value" else "string", "description": FIELDS[f], } for f in fields }, "required": fields, "additionalProperties": False, }, } for k, (desc, fields) in DEFS.items() ] def call(tool, args): if tool not in DEFS or set(args) != set(DEFS[tool][1]): raise ValueError("Unknown tool or invalid arguments.") base = os.environ["NORE_URL"].rstrip("/") u = urlsplit(base) if u.scheme != "https" and not ( u.scheme == "http" and u.hostname in ("127.0.0.1", "localhost") ): raise ValueError("HTTPS required.") token = Path(os.environ["NORE_TOKEN_FILE"]).read_text().strip() for k in ("tenant", "project", "key", "hash", "cue", "frame"): if k in args and not re.fullmatch(r"[A-Za-z0-9._-]{1,150}", args[k]): raise ValueError("Invalid identifier.") path = "/api/me" body = None if tool != "nore_projects": scope = "/" + args["tenant"] + "/" + args["project"] if tool in ("nore_directions", "nore_suggest"): path = "/api" + scope + "/directions/" + args["frame"] if tool == "nore_suggest": body = args["value"] elif tool == "nore_motion_prompt": path = "/api" + scope + "/motion/" + args["key"] + "/prompt" elif tool in ("nore_prompt_bench", "nore_prompt_variant"): path = "/api" + scope + "/prompt-bench/" + args["frame"] if tool == "nore_prompt_variant": body = args["value"] elif tool in ("nore_sounds", "nore_listen"): path = "/api" + scope + "/sounds" if tool == "nore_listen": path += "/" + args["cue"] elif tool == "nore_evals": path = "/api" + scope + "/direction-evals" elif tool == "nore_image": path = "/p" + scope + "/media/" + args["hash"] elif tool == "nore_frames": path = "/api" + scope + "/manifest" elif tool == "nore_events": path = "/api" + scope + "/events?after=" + str(int(args["after"])) else: path = "/api" + scope + "/docs" if tool in ("nore_read", "nore_write"): path += "/" + args["key"] if tool == "nore_propose": path += "/proposal." + uuid4().hex body = { "base_revision": 0, "value": { "title": args["title"], "text": args["text"], "status": "proposal", }, } if tool == "nore_write": body = {"base_revision": args["base_revision"], "value": args["value"]} req = Request( base + path, method="POST" if tool in ("nore_suggest", "nore_prompt_variant") else "GET" if body is None else "PUT", data=None if body is None else json.dumps(body).encode(), headers={ "Authorization": "Bearer " + token, "Content-Type": "application/json", }, ) try: with urlopen(req, timeout=20) as res: if tool in ("nore_image", "nore_listen"): return { "type": "audio" if tool == "nore_listen" else "image", "mimeType": res.headers.get_content_type(), "data": base64.b64encode(res.read()).decode(), } return { "type": "text", "text": json.dumps(json.load(res), ensure_ascii=False), } except HTTPError as e: raise ValueError( "HTTP " + str(e.code) + ": " + e.read(512).decode(errors="replace") ) from None def mcp(): for line in sys.stdin: ident = None try: req = json.loads(line) ident = req.get("id") method = req.get("method") params = req.get("params") or {} if ident is None: continue if method == "initialize": supported = ("2025-11-25", "2025-06-18", "2025-03-26", "2024-11-05") version = params.get("protocolVersion") result = { "protocolVersion": version if version in supported else supported[0], "capabilities": {"tools": {}}, "serverInfo": {"name": "nore-desk", "version": "1.0"}, } elif method == "tools/list": result = {"tools": tools_list()} elif method == "tools/call": try: result = { "content": [call(params["name"], params.get("arguments", {}))] } except (ValueError, KeyError, OSError) as e: result = { "isError": True, "content": [{"type": "text", "text": str(e)}], } elif method == "ping": result = {} else: print( json.dumps( { "jsonrpc": "2.0", "id": ident, "error": {"code": -32601, "message": "Unknown method"}, } ), flush=True, ) continue response = {"jsonrpc": "2.0", "id": ident, "result": result} except (ValueError, AttributeError, TypeError, KeyError): response = { "jsonrpc": "2.0", "id": ident, "error": {"code": -32600, "message": "Invalid request"}, } print(json.dumps(response), flush=True) if __name__ == "__main__": p = argparse.ArgumentParser() p.add_argument("tool", choices=["mcp", *DEFS]) p.add_argument( "--args", default="{}", help="JSON arguments; credentials come from NORE_TOKEN_FILE", ) a = p.parse_args() if a.tool == "mcp": mcp() else: print(json.dumps(call(a.tool, json.loads(a.args)), ensure_ascii=False))