Enforce ID-based node reads in MCP wrapper

This commit is contained in:
HumbleTurtle
2026-05-22 22:09:11 -03:00
parent 9bce54ac0a
commit 27a37a4938
2 changed files with 347 additions and 25 deletions
+191 -25
View File
@@ -492,9 +492,11 @@ PUBLIC_TOOLS = {
"nap_tee", "nap_tee",
"nap_patch", "nap_patch",
"nap_node_get", "nap_node_get",
"nap_node_by_id",
"nap_node_by_uri",
"nap_node_by_path",
"nap_node_patch", "nap_node_patch",
"nap_rm", "nap_rm",
"nap_cat",
"nap_ln", "nap_ln",
"nap_backlinks", "nap_backlinks",
"nap_ls", "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 [])] 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): def decrypt_relay_frame(key, frame_text, context_hash, relay_lane, expected_direction, expected_seq):
envelope = json.loads(frame_text) envelope = json.loads(frame_text)
if ( if (
@@ -5544,6 +5588,23 @@ def node_write_reference(node):
return {key: value for key, value in item.items() if value not in (None, "")} 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): 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] view = view or normalize_discovery_view(args)[0]
limit = normalize_int(args.get("limit"), DISCOVERY_DEFAULT_LIMIT, 1, discovery_max_limit(view)) 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) 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): def node_get(args):
view = args.get("view", "edit") view = args.get("view", "edit")
if view not in {"render", "edit", "json"}: if view not in {"render", "edit", "json"}:
raise RuntimeError("view must be render, edit, or 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): def guarded_node_patch(args):
@@ -7822,7 +7951,7 @@ def list_project_nodes(args):
def list_active_plans(args): def list_active_plans(args):
view, _warnings = normalize_discovery_view(args) view, _warnings = normalize_discovery_view(args)
if view == "full": 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 = { query = {
**args, **args,
"folder_path": args.get("folder_path") or "/plans", "folder_path": args.get("folder_path") or "/plans",
@@ -8169,7 +8298,7 @@ def list_kanban_cards(args):
project_id = resolve_project_id(args) project_id = resolve_project_id(args)
view, view_warnings = normalize_discovery_view(args) view, view_warnings = normalize_discovery_view(args)
if view == "full": 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) 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) preview_chars = normalize_int(args.get("preview_chars"), 240, 80, 1000)
active_only = as_bool(args.get("active_only"), True) active_only = as_bool(args.get("active_only"), True)
@@ -9187,7 +9316,7 @@ def discover_memory(args):
result["awareness"] = { result["awareness"] = {
"default_sort": "search_relevance_then_updated_desc" if q else "updated_desc", "default_sort": "search_relevance_then_updated_desc" if q else "updated_desc",
"date_fields_in_rows": ["date", "created_at", "updated_at"], "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 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_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_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_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", "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": { "inputSchema": {
"type": "object", "type": "object",
"properties": { "properties": {
@@ -10855,7 +10985,7 @@ def raw_tools():
}, },
{ {
"name": "nap_find", "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": { "inputSchema": {
"type": "object", "type": "object",
"properties": { "properties": {
@@ -11163,13 +11293,51 @@ def raw_tools():
}, },
{ {
"name": "nap_node_get", "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": { "inputSchema": {
"type": "object", "type": "object",
"required": ["path"], "required": ["path"],
"properties": { "properties": {
"path": {"type": "string"}, "path": {"type": "string", "description": "Index/route path such as /documentation/product/prd."},
"view": {"type": "string", "enum": ["render", "edit", "json"], "default": "edit"},
}, },
"additionalProperties": False, "additionalProperties": False,
}, },
@@ -11291,16 +11459,6 @@ def raw_tools():
"additionalProperties": False, "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", "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.", "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_lint",
"nap_doctor", "nap_doctor",
"nap_node_get", "nap_node_get",
"nap_cat", "nap_node_by_id",
"nap_node_by_uri",
"nap_node_by_path",
"nap_backlinks", "nap_backlinks",
"nap_ls", "nap_ls",
"nap_grep_local", "nap_grep_local",
@@ -11700,8 +11860,10 @@ READ_ONLY_TOOLS = {
"nap_doctor", "nap_doctor",
"nap_feedback_admin_ls", "nap_feedback_admin_ls",
"nap_ls_tags", "nap_ls_tags",
"nap_cat",
"nap_node_get", "nap_node_get",
"nap_node_by_id",
"nap_node_by_uri",
"nap_node_by_path",
"nap_backlinks", "nap_backlinks",
"nap_ls", "nap_ls",
"nap_agent_ls", "nap_agent_ls",
@@ -12119,6 +12281,12 @@ def call_tool_impl(name, args):
return update_node_by_path(args) return update_node_by_path(args)
if name == "nap_node_get": if name == "nap_node_get":
return node_get(args) 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": if name == "nap_node_patch":
return guarded_node_patch(args) return guarded_node_patch(args)
if name == "nap_rm": if name == "nap_rm":
@@ -12147,8 +12315,6 @@ def call_tool_impl(name, args):
return update_kanban_card(args) return update_kanban_card(args)
if name == "nap_kanban_archive": if name == "nap_kanban_archive":
return archive_node_by_path(args) return archive_node_by_path(args)
if name == "nap_cat":
return get_node_by_path(args)
if name == "nap_ln": if name == "nap_ln":
return link_nodes(args) return link_nodes(args)
if name == "nap_backlinks": if name == "nap_backlinks":
+156
View File
@@ -5,6 +5,8 @@ import importlib.util
import pathlib import pathlib
import sys import sys
import pytest
def load_module(): def load_module():
script_path = pathlib.Path(__file__).resolve().parents[1] / "resources" / "scripts" / "napseer_mcp_server.py" script_path = pathlib.Path(__file__).resolve().parents[1] / "resources" / "scripts" / "napseer_mcp_server.py"
@@ -19,6 +21,11 @@ def load_module():
return 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): 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) folder, name = path.rsplit("/", 1)
merged_metadata = {"status": status} merged_metadata = {"status": status}
@@ -80,6 +87,143 @@ def test_node_discovery_allows_large_requested_limit(mod):
assert len(calls) == 1 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): def test_recent_memory_defaults_to_paths(mod):
recent = node( recent = node(
"/notes/recent-title", "/notes/recent-title",
@@ -629,6 +773,18 @@ def run():
mod = load_module() mod = load_module()
test_node_discovery_allows_large_requested_limit(mod) test_node_discovery_allows_large_requested_limit(mod)
mod = load_module() 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) test_recent_memory_defaults_to_paths(mod)
mod = load_module() mod = load_module()
test_find_defaults_to_paths_and_requests_compact_rest_view(mod) test_find_defaults_to_paths_and_requests_compact_rest_view(mod)