diff --git a/resources/scripts/napseer_mcp_server.py b/resources/scripts/napseer_mcp_server.py index 5456e33..327ec25 100755 --- a/resources/scripts/napseer_mcp_server.py +++ b/resources/scripts/napseer_mcp_server.py @@ -492,9 +492,11 @@ PUBLIC_TOOLS = { "nap_tee", "nap_patch", "nap_node_get", + "nap_node_by_id", + "nap_node_by_uri", + "nap_node_by_path", "nap_node_patch", "nap_rm", - "nap_cat", "nap_ln", "nap_backlinks", "nap_ls", @@ -3737,6 +3739,48 @@ def decrypt_nodes_for_return(nodes, project_id=None): return [decrypt_node_for_return(node, project_id=project_id) for node in (nodes or [])] +def nap_project_route_name(): + value = AUTH.get("project_slug") or AUTH.get("project_name") or pathlib.Path.cwd().name or "project" + slug = re.sub(r"[^a-zA-Z0-9._~-]+", "-", str(value).strip()).strip("-") + return slug or "project" + + +def nap_node_uri(project_id, node_id): + project_name = urllib.parse.quote(nap_project_route_name(), safe="._~-") + project = urllib.parse.quote(str(project_id or current_project_id() or ""), safe="") + node = urllib.parse.quote(str(node_id or ""), safe="") + if not project or not node: + return "" + return f"nap://{project_name}/{project}/{node}" + + +def node_identity_payload(node, project_id=None, lookup_kind=None, lookup_value=None): + resolved_project_id = project_id or node.get("project_id") or current_project_id() + return { + "node_id": node.get("id"), + "canonical_uri": nap_node_uri(resolved_project_id, node.get("id")), + "project_id": resolved_project_id, + "project_name": nap_project_route_name(), + "lookup_kind": lookup_kind, + "lookup_value": lookup_value, + "legacy_full_path": node.get("full_path"), + } + + +def parse_nap_node_uri(value): + parsed = urllib.parse.urlparse(str(value or "").strip()) + if parsed.scheme != "nap": + raise RuntimeError("uri must use nap://project-name/project-id/node-id") + segments = [urllib.parse.unquote(part) for part in parsed.path.split("/") if part] + if len(segments) != 2: + raise RuntimeError("uri must use nap://project-name/project-id/node-id") + return { + "project_name": urllib.parse.unquote(parsed.netloc), + "project_id": segments[0], + "node_id": segments[1], + } + + def decrypt_relay_frame(key, frame_text, context_hash, relay_lane, expected_direction, expected_seq): envelope = json.loads(frame_text) if ( @@ -5544,6 +5588,23 @@ def node_write_reference(node): return {key: value for key, value in item.items() if value not in (None, "")} +def node_route_reference(node): + item = { + "id": node.get("id"), + "project_id": node.get("project_id"), + "full_path": node.get("full_path"), + "type": node.get("type"), + "name": node.get("name"), + "folder_path": node.get("folder_path"), + "status": node_status(node), + "date": node_date(node), + "created_at": node.get("created_at"), + "updated_at": node.get("updated_at"), + "archived": node_archived(node), + } + return {key: value for key, value in item.items() if value not in (None, "")} + + def discovery_envelope(project_id, items, args, *, view=None, next_cursor=None, has_more=None, diagnostics=None, query_analysis=None, warnings=None): view = view or normalize_discovery_view(args)[0] limit = normalize_int(args.get("limit"), DISCOVERY_DEFAULT_LIMIT, 1, discovery_max_limit(view)) @@ -7504,11 +7565,79 @@ def get_node_by_path(args, allow_agent=False): return decrypt_node_for_return(node, project_id=project_id) +def get_node_by_id(args): + project_id = str(args.get("project_id") or resolve_project_id(args)).strip() + node_id = str(args.get("node_id") or args.get("id") or "").strip() + if not node_id: + raise RuntimeError("node_id is required") + view = args.get("view", "render") + if view not in {"render", "edit", "json"}: + raise RuntimeError("view must be render, edit, or json") + params = "" if view == "render" else f"?{rest_query_string({'view': view})}" + node = request_json("GET", f"/v1/projects/{project_id}/nodes/{urllib.parse.quote(node_id, safe='')}{params}") + if view == "render": + return decrypt_node_for_return(node, project_id=project_id) + if isinstance(node, dict) and isinstance(node.get("node"), dict): + node["node"] = decrypt_node_for_return(node["node"], project_id=project_id) + return node + + def node_get(args): view = args.get("view", "edit") if view not in {"render", "edit", "json"}: raise RuntimeError("view must be render, edit, or json") - return get_node_by_path({"path": args["path"], "view": view}) + if args.get("uri"): + return node_by_uri({"uri": args["uri"], "view": view})["node"] + if args.get("node_id") or args.get("id"): + return get_node_by_id({"node_id": args.get("node_id") or args.get("id"), "view": view}) + if args.get("path"): + raise RuntimeError("path is an index/route only; resolve it with nap_node_by_path, then read with node_id or uri") + raise RuntimeError("one of node_id or uri is required") + + +def node_by_id(args): + view = args.get("view", "render") + if view not in {"render", "edit", "json"}: + raise RuntimeError("view must be render, edit, or json") + node_id = str(args.get("node_id") or args.get("id") or "").strip() + node = get_node_by_id({"node_id": node_id, "view": view}) + return { + "ok": True, + "status": "resolved", + "identity": node_identity_payload(node, project_id=node.get("project_id"), lookup_kind="node_id", lookup_value=node_id), + "node": node, + } + + +def node_by_uri(args): + view = args.get("view", "render") + if view not in {"render", "edit", "json"}: + raise RuntimeError("view must be render, edit, or json") + parsed = parse_nap_node_uri(args["uri"]) + configured_project_id = current_project_id() + if configured_project_id and parsed["project_id"] != configured_project_id: + raise RuntimeError("nap:// URI project_id does not match the configured local project") + node = get_node_by_id({"project_id": parsed["project_id"], "node_id": parsed["node_id"], "view": view}) + return { + "ok": True, + "status": "resolved", + "identity": node_identity_payload(node, project_id=parsed["project_id"], lookup_kind="uri", lookup_value=args["uri"]), + "node": node, + } + + +def node_by_path(args): + path = normalize_node_path(args["path"]) + node = get_node_by_path({"path": path, "view": "render"}) + return { + "ok": True, + "status": "resolved", + "identity": node_identity_payload(node, project_id=node.get("project_id"), lookup_kind="path", lookup_value=path), + "route": node_route_reference(node), + "warnings": [ + "Path lookup is index/route resolution only. Read node content with identity.canonical_uri or identity.node_id." + ], + } def guarded_node_patch(args): @@ -7822,7 +7951,7 @@ def list_project_nodes(args): def list_active_plans(args): view, _warnings = normalize_discovery_view(args) if view == "full": - raise RuntimeError("nap_plan_list_active supports paths, summary, or metadata views; use nap_cat for full plan bodies") + raise RuntimeError("nap_plan_list_active supports paths, summary, or metadata views; resolve a path with nap_node_by_path and read by node_id or uri for full plan bodies") query = { **args, "folder_path": args.get("folder_path") or "/plans", @@ -8169,7 +8298,7 @@ def list_kanban_cards(args): project_id = resolve_project_id(args) view, view_warnings = normalize_discovery_view(args) if view == "full": - raise RuntimeError("nap_kanban_list supports paths, summary, or metadata views; use nap_cat for full card bodies") + raise RuntimeError("nap_kanban_list supports paths, summary, or metadata views; resolve a path with nap_node_by_path and read by node_id or uri for full card bodies") limit = normalize_int(args.get("limit"), DISCOVERY_DEFAULT_LIMIT, 1, DISCOVERY_COMPACT_MAX_LIMIT) preview_chars = normalize_int(args.get("preview_chars"), 240, 80, 1000) active_only = as_bool(args.get("active_only"), True) @@ -9187,7 +9316,7 @@ def discover_memory(args): result["awareness"] = { "default_sort": "search_relevance_then_updated_desc" if q else "updated_desc", "date_fields_in_rows": ["date", "created_at", "updated_at"], - "next_step": "Use nap_context or nap_cat on important paths when the preview is not enough.", + "next_step": "Use nap_node_by_path on important paths, then read by identity with nap_node_get when the preview is not enough.", } return result @@ -9227,7 +9356,8 @@ def classify_memory_query(args): {"tool": "nap_context", "when": "You already know one or more canonical paths and need surrounding links/backlinks."}, {"tool": "nap_discover", "when": "You need search plus folder/tag/type/status/date filters in one call."}, {"tool": "nap_recent", "when": "You need latest project changes by time window."}, - {"tool": "nap_cat", "when": "You selected a path and need the full body."}, + {"tool": "nap_node_by_path", "when": "You selected a path and need to resolve it to node_id / nap:// URI before reading."}, + {"tool": "nap_node_get", "when": "You have node_id or nap:// URI and need the full body."}, ], } @@ -10826,7 +10956,7 @@ def raw_tools(): }, { "name": "nap_discover", - "description": "Intent-shaped project memory discovery. Combine keywords, folder/tag/type/status filters, updated_at ranges, recent sorting, and compact date-aware rows before using nap_context or nap_cat.", + "description": "Intent-shaped project memory discovery. Combine keywords, folder/tag/type/status filters, updated_at ranges, recent sorting, and compact date-aware rows before resolving paths with nap_node_by_path or reading identity with nap_node_get.", "inputSchema": { "type": "object", "properties": { @@ -10855,7 +10985,7 @@ def raw_tools(): }, { "name": "nap_find", - "description": "Structured active memory search by folder, tag, path, or query. Returns path/date rows by default; use view='full' or nap_cat for body text.", + "description": "Structured active memory search by folder, tag, path, or query. Returns path/date rows by default; use view='full' for body text or resolve selected paths with nap_node_by_path before identity reads.", "inputSchema": { "type": "object", "properties": { @@ -11163,13 +11293,51 @@ def raw_tools(): }, { "name": "nap_node_get", - "description": "Read one node by path with optional edit/json view. Use view='edit' before nap_node_patch to get revision and read_fingerprint.", + "description": "Read one node by canonical node_id or nap:// URI. Paths are indexes/routes only; resolve a path with nap_node_by_path, then read by node_id or uri. Use view='edit' before nap_node_patch to get revision and read_fingerprint.", + "inputSchema": { + "type": "object", + "properties": { + "node_id": {"type": "string", "description": "Canonical node UUID."}, + "uri": {"type": "string", "description": "Canonical nap://project-name/project-id/node-id URI."}, + "view": {"type": "string", "enum": ["render", "edit", "json"], "default": "edit"}, + }, + "additionalProperties": False, + }, + }, + { + "name": "nap_node_by_id", + "description": "Read one node by canonical node id and return identity metadata including its nap:// URI.", + "inputSchema": { + "type": "object", + "required": ["node_id"], + "properties": { + "node_id": {"type": "string", "description": "Canonical node UUID."}, + "view": {"type": "string", "enum": ["render", "edit", "json"], "default": "render"}, + }, + "additionalProperties": False, + }, + }, + { + "name": "nap_node_by_uri", + "description": "Read one node by canonical nap://project-name/project-id/node-id URI.", + "inputSchema": { + "type": "object", + "required": ["uri"], + "properties": { + "uri": {"type": "string", "description": "Canonical nap://project-name/project-id/node-id URI."}, + "view": {"type": "string", "enum": ["render", "edit", "json"], "default": "render"}, + }, + "additionalProperties": False, + }, + }, + { + "name": "nap_node_by_path", + "description": "Resolve a path index/route to canonical node identity. This does not read full node content; use returned node_id or nap:// URI with nap_node_get, nap_node_by_id, or nap_node_by_uri.", "inputSchema": { "type": "object", "required": ["path"], "properties": { - "path": {"type": "string"}, - "view": {"type": "string", "enum": ["render", "edit", "json"], "default": "edit"}, + "path": {"type": "string", "description": "Index/route path such as /documentation/product/prd."}, }, "additionalProperties": False, }, @@ -11291,16 +11459,6 @@ def raw_tools(): "additionalProperties": False, }, }, - { - "name": "nap_cat", - "description": "Read one node by canonical full_path. Use this before linking if you know the path but need the node id or normalized path.", - "inputSchema": { - "type": "object", - "required": ["path"], - "properties": {"path": {"type": "string", "description": "Canonical node path such as /decisions/auth-model."}}, - "additionalProperties": False, - }, - }, { "name": "nap_ln", "description": "Create a first-class graph link from source_path to target_path. Use this instead of burying references in prose or metadata.", @@ -11590,7 +11748,9 @@ TOOL_CATEGORIES = { "nap_lint", "nap_doctor", "nap_node_get", - "nap_cat", + "nap_node_by_id", + "nap_node_by_uri", + "nap_node_by_path", "nap_backlinks", "nap_ls", "nap_grep_local", @@ -11700,8 +11860,10 @@ READ_ONLY_TOOLS = { "nap_doctor", "nap_feedback_admin_ls", "nap_ls_tags", - "nap_cat", "nap_node_get", + "nap_node_by_id", + "nap_node_by_uri", + "nap_node_by_path", "nap_backlinks", "nap_ls", "nap_agent_ls", @@ -12119,6 +12281,12 @@ def call_tool_impl(name, args): return update_node_by_path(args) if name == "nap_node_get": return node_get(args) + if name == "nap_node_by_id": + return node_by_id(args) + if name == "nap_node_by_uri": + return node_by_uri(args) + if name == "nap_node_by_path": + return node_by_path(args) if name == "nap_node_patch": return guarded_node_patch(args) if name == "nap_rm": @@ -12147,8 +12315,6 @@ def call_tool_impl(name, args): return update_kanban_card(args) if name == "nap_kanban_archive": return archive_node_by_path(args) - if name == "nap_cat": - return get_node_by_path(args) if name == "nap_ln": return link_nodes(args) if name == "nap_backlinks": diff --git a/tests/test_mcp_memory_ergonomics_helpers.py b/tests/test_mcp_memory_ergonomics_helpers.py index e620dde..4af78f4 100644 --- a/tests/test_mcp_memory_ergonomics_helpers.py +++ b/tests/test_mcp_memory_ergonomics_helpers.py @@ -5,6 +5,8 @@ import importlib.util import pathlib import sys +import pytest + def load_module(): script_path = pathlib.Path(__file__).resolve().parents[1] / "resources" / "scripts" / "napseer_mcp_server.py" @@ -19,6 +21,11 @@ def load_module(): return module +@pytest.fixture +def mod(): + return load_module() + + def node(path, status="planned", metadata=None, tags=None, content="body text", archived_at=None, node_type="plan", links=None): folder, name = path.rsplit("/", 1) merged_metadata = {"status": status} @@ -80,6 +87,143 @@ def test_node_discovery_allows_large_requested_limit(mod): assert len(calls) == 1 +def test_node_by_path_returns_canonical_identity_uri(mod): + mod.AUTH["project_slug"] = "napseer" + found = node("/documentation/product/prd", "active", node_type="product_spec") + found["id"] = "node-123" + + def fake_request(method, path, payload=None, **kwargs): + assert method == "GET" + assert "/nodes/by-path?" in path + assert "path=%2Fdocumentation%2Fproduct%2Fprd" in path + assert "view=render" in path + return found + + mod.request_json = fake_request + result = mod.node_by_path({"path": "documentation/product/prd"}) + assert result["ok"] is True + assert result["status"] == "resolved" + assert result["identity"] == { + "node_id": "node-123", + "canonical_uri": "nap://napseer/project-1/node-123", + "project_id": "project-1", + "project_name": "napseer", + "lookup_kind": "path", + "lookup_value": "/documentation/product/prd", + "legacy_full_path": "/documentation/product/prd", + } + assert result["route"]["full_path"] == "/documentation/product/prd" + assert "content_text" not in result["route"] + assert "preview" not in result["route"] + assert "read_fingerprint" not in result["route"] + assert "Path lookup is index/route resolution only" in result["warnings"][0] + + +def test_node_by_id_returns_canonical_identity_uri(mod): + mod.AUTH["project_slug"] = "napseer" + found = node("/documentation/product/prd", "active", node_type="product_spec") + found["id"] = "node-123" + + def fake_request(method, path, payload=None, **kwargs): + assert method == "GET" + assert path == "/v1/projects/project-1/nodes/node-123" + return found + + mod.request_json = fake_request + result = mod.node_by_id({"node_id": "node-123"}) + assert result["ok"] is True + assert result["identity"]["lookup_kind"] == "node_id" + assert result["identity"]["canonical_uri"] == "nap://napseer/project-1/node-123" + assert result["node"]["id"] == "node-123" + + +def test_node_by_uri_resolves_canonical_uri(mod): + mod.AUTH["project_slug"] = "napseer" + found = node("/documentation/product/prd", "active", node_type="product_spec") + found["id"] = "node-123" + + def fake_request(method, path, payload=None, **kwargs): + assert method == "GET" + assert path == "/v1/projects/project-1/nodes/node-123" + return found + + mod.request_json = fake_request + result = mod.node_by_uri({"uri": "nap://napseer/project-1/node-123"}) + assert result["ok"] is True + assert result["identity"]["lookup_kind"] == "uri" + assert result["identity"]["lookup_value"] == "nap://napseer/project-1/node-123" + assert result["identity"]["canonical_uri"] == "nap://napseer/project-1/node-123" + + +def test_node_get_accepts_node_id_without_path(mod): + found = node("/documentation/product/prd", "active", node_type="product_spec") + found["id"] = "node-123" + + def fake_request(method, path, payload=None, **kwargs): + assert method == "GET" + assert path == "/v1/projects/project-1/nodes/node-123" + return found + + mod.request_json = fake_request + result = mod.node_get({"node_id": "node-123", "view": "render"}) + assert result["id"] == "node-123" + assert result["full_path"] == "/documentation/product/prd" + + +def test_node_get_edit_view_uses_id_route_not_by_path(mod): + edit_payload = { + "node": node("/documentation/product/prd", "active", node_type="product_spec"), + "revision": "2026-05-06T12:00:00Z", + "read_fingerprint": "fingerprint-1", + "content_lines": ["body text"], + "line_count": 1, + "ends_with_newline": False, + } + edit_payload["node"]["id"] = "node-123" + + def fake_request(method, path, payload=None, **kwargs): + assert method == "GET" + assert path == "/v1/projects/project-1/nodes/node-123?view=edit" + return edit_payload + + mod.request_json = fake_request + result = mod.node_get({"node_id": "node-123", "view": "edit"}) + assert result["revision"] == "2026-05-06T12:00:00Z" + assert result["node"]["id"] == "node-123" + + +def test_node_get_rejects_path_reads(mod): + try: + mod.node_get({"path": "/documentation/product/prd", "view": "render"}) + except RuntimeError as exc: + assert "path is an index/route only" in str(exc) + else: + raise AssertionError("path reads should be rejected") + + +def test_tools_include_node_by_path_descriptor(mod): + get_tool = next(item for item in mod.tools() if item["name"] == "nap_node_get") + assert get_tool["inputSchema"].get("required") is None + assert "node_id" in get_tool["inputSchema"]["properties"] + assert "uri" in get_tool["inputSchema"]["properties"] + assert "path" not in get_tool["inputSchema"]["properties"] + + by_id = next(item for item in mod.tools() if item["name"] == "nap_node_by_id") + assert by_id["inputSchema"]["required"] == ["node_id"] + assert "nap:// URI" in by_id["description"] + + by_uri = next(item for item in mod.tools() if item["name"] == "nap_node_by_uri") + assert by_uri["inputSchema"]["required"] == ["uri"] + assert "nap://" in by_uri["description"] + + tool = next(item for item in mod.tools() if item["name"] == "nap_node_by_path") + assert tool["inputSchema"]["required"] == ["path"] + assert "preview_chars" not in tool["inputSchema"]["properties"] + assert "does not read full node content" in tool["description"] + + assert all(item["name"] != "nap_cat" for item in mod.tools()) + + def test_recent_memory_defaults_to_paths(mod): recent = node( "/notes/recent-title", @@ -629,6 +773,18 @@ def run(): mod = load_module() test_node_discovery_allows_large_requested_limit(mod) mod = load_module() + test_node_by_path_returns_canonical_identity_uri(mod) + mod = load_module() + test_node_by_id_returns_canonical_identity_uri(mod) + mod = load_module() + test_node_by_uri_resolves_canonical_uri(mod) + mod = load_module() + test_node_get_accepts_node_id_without_path(mod) + mod = load_module() + test_node_get_rejects_path_reads(mod) + mod = load_module() + test_tools_include_node_by_path_descriptor(mod) + mod = load_module() test_recent_memory_defaults_to_paths(mod) mod = load_module() test_find_defaults_to_paths_and_requests_compact_rest_view(mod)