napseer/nap-cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Require identity for context graph tools

Browse Source
This commit is contained in:
HumbleTurtle
2026-05-22 23:34:59 -03:00
parent fa754bb5bd
commit 1cb8075324
2 changed files with 103 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
resources/scripts/napseer_mcp_server.py
+37 -24
View File
@@ -7935,10 +7935,10 @@ def link_nodes(args):
def get_backlinks_by_path(args):
if args.get("path"):
if args.get("path") and not args.get("_internal_path_ref"):
raise RuntimeError("path is an index/route only; resolve it with nap_node_by_path, then use node_id or uri")
project_id = resolve_project_id(args)
target = resolve_existing_node_ref(args)
target = get_node_by_path({"path": args["path"]}) if args.get("path") else resolve_existing_node_ref(args)
backlinks = request_json("GET", f"/v1/projects/{project_id}/nodes/{target['id']}/backlinks")
return {"target": {"id": target["id"], "full_path": target["full_path"]}, **backlinks}
@@ -9455,28 +9455,38 @@ def classify_memory_query(args):
def get_memory_context(args):
raw_paths = args.get("paths") or ([args["path"]] if args.get("path") else [])
if not raw_paths:
raise RuntimeError("path or paths is required")
if args.get("path") or args.get("paths"):
raise RuntimeError("paths are index/routes only; resolve them with nap_node_by_path, then use node_id/uri inputs")
raw_node_ids = args.get("node_ids") or ([args["node_id"]] if args.get("node_id") else [])
raw_uris = args.get("uris") or ([args["uri"]] if args.get("uri") else [])
if not raw_node_ids and not raw_uris:
raise RuntimeError("node_id, node_ids, uri, or uris is required")
max_nodes = max(1, min(int(args.get("max_nodes", 20)), 100))
depth = max(0, min(int(args.get("depth", 1)), 2))
default_view = "summary"
view, _warnings = normalize_discovery_view(args, default=default_view)
queue_paths = [normalize_node_path(path) for path in raw_paths]
queue = []
for node_id in raw_node_ids:
queue.append(("node", get_node_by_id({"node_id": node_id})))
for uri in raw_uris:
queue.append(("node", node_by_uri({"uri": uri})["node"]))
nodes = []
edges = []
seen_paths = set()
current_depth = 0
while queue_paths and len(nodes) < max_nodes and current_depth <= depth:
next_paths = []
for path in queue_paths:
if path in seen_paths or len(nodes) >= max_nodes:
while queue and len(nodes) < max_nodes and current_depth <= depth:
next_queue = []
for kind, value in queue:
if len(nodes) >= max_nodes:
continue
seen_paths.add(path)
try:
node = read_node_by_path({"path": path}, allow_agent=is_agent_namespace_path(path))
node = value if kind == "node" else read_node_by_path({"path": value}, allow_agent=is_agent_namespace_path(value))
except Exception:
continue
path = node.get("full_path")
if not path or path in seen_paths:
continue
seen_paths.add(path)
nodes.append(summarize_node(node, view=view, preview_chars=normalize_int(args.get("preview_chars"), 240, 80, 1000)))
links = node.get("links") if isinstance(node.get("links"), list) else []
for link in links:
@@ -9484,30 +9494,31 @@ def get_memory_context(args):
if target_path:
edges.append({"source": node["full_path"], "target": target_path, "relation": link.get("relation", "references")})
if current_depth < depth:
next_paths.append(target_path)
next_queue.append(("path", target_path))
if current_depth < depth:
try:
backlinks = get_backlinks_by_path({"path": node["full_path"]}).get("items", [])
backlinks = get_backlinks_by_path({"node_id": node["id"]}).get("items", [])
for backlink in backlinks:
source_path = backlink.get("full_path")
if source_path:
edges.append({"source": source_path, "target": node["full_path"], "relation": "backlink"})
next_paths.append(source_path)
next_queue.append(("path", source_path))
except Exception:
pass
queue_paths = next_paths
queue = next_queue
current_depth += 1
return {"items": unique_nodes(nodes), "edges": edges[: max_nodes * 4], "view": view}
def find_related_nodes(args):
path = args["path"]
if args.get("path"):
raise RuntimeError("path is an index/route only; resolve it with nap_node_by_path, then use node_id or uri")
view, _warnings = normalize_discovery_view(args, default="paths")
limit = max(1, min(int(args.get("limit", 25)), 100))
target = read_node_by_path({"path": path}, allow_agent=is_agent_namespace_path(path))
target = resolve_node_ref_allow_agent(args)
related = []
try:
related.extend(get_backlinks_by_path({"path": target["full_path"]}).get("items", []))
related.extend(get_backlinks_by_path({"node_id": target["id"]}).get("items", []))
except Exception:
pass
links = target.get("links") if isinstance(target.get("links"), list) else []
@@ -9529,7 +9540,7 @@ def find_related_nodes(args):
def inbound_reference_plan(path):
try:
backlinks = get_backlinks_by_path({"path": path}).get("items", [])
backlinks = get_backlinks_by_path({"path": path, "_internal_path_ref": True}).get("items", [])
except Exception:
backlinks = []
return [
@@ -11205,8 +11216,10 @@ def raw_tools():
"inputSchema": {
"type": "object",
"properties": {
"path": {"type": "string"},
"paths": {"type": "array", "items": {"type": "string"}},
"node_id": {"type": "string"},
"node_ids": {"type": "array", "items": {"type": "string"}},
"uri": {"type": "string"},
"uris": {"type": "array", "items": {"type": "string"}},
"depth": {"type": "integer", "minimum": 0, "maximum": 2},
"max_nodes": {"type": "integer", "minimum": 1, "maximum": 100},
"view": {"type": "string", "enum": ["paths", "summary", "metadata"], "default": "summary"},
@@ -11231,9 +11244,9 @@ def raw_tools():
"description": "Find related memory by outgoing links, backlinks, shared tags, and folder.",
"inputSchema": {
"type": "object",
"required": ["path"],
"properties": {
"path": {"type": "string"},
"node_id": {"type": "string"},
"uri": {"type": "string"},
"limit": {"type": "integer", "minimum": 1, "maximum": 100},
"view": {"type": "string", "enum": ["titles", "paths", "summary", "metadata"], "default": "paths"},
"include_content": {"type": "boolean", "default": False, "description": "Legacy no-op; read full bodies with nap_node_get by node_id or uri."},
tests/test_mcp_memory_ergonomics_helpers.py
+66
View File
@@ -323,6 +323,19 @@ def test_tools_include_node_by_path_descriptor(mod):
assert "uri" in backlinks_schema["properties"]
assert "path" not in backlinks_schema["properties"]
context_schema = next(item for item in mod.tools() if item["name"] == "nap_context")["inputSchema"]
assert "node_id" in context_schema["properties"]
assert "node_ids" in context_schema["properties"]
assert "uri" in context_schema["properties"]
assert "uris" in context_schema["properties"]
assert "path" not in context_schema["properties"]
assert "paths" not in context_schema["properties"]
related_schema = next(item for item in mod.tools() if item["name"] == "nap_find_related")["inputSchema"]
assert "node_id" in related_schema["properties"]
assert "uri" in related_schema["properties"]
assert "path" not in related_schema["properties"]
complete_schema = next(item for item in mod.tools() if item["name"] == "nap_plan_complete")["inputSchema"]
assert "plan_node_id" in complete_schema["properties"]
assert "plan_uri" in complete_schema["properties"]
@@ -566,6 +579,59 @@ def test_agent_tools_create_by_route_and_update_by_identity(mod):
mod.link_agent_node({"agent_slug": "alice", "source_path": "/memory/current", "target_path": "/notes/target"})
def test_context_and_related_use_identity_inputs(mod):
root = node("/notes/root", "active", node_type="note", links=[{"path": "/notes/linked", "relation": "references"}], tags=["topic"])
root["id"] = "root-123"
linked = node("/notes/linked", "active", node_type="note", tags=["topic"])
linked["id"] = "linked-123"
backlink = node("/notes/backlink", "active", node_type="note")
backlink["id"] = "backlink-123"
reads = []
def fake_get_node_by_id(args):
reads.append(args)
node_id = args.get("node_id")
if node_id == "root-123":
return root
if node_id == "linked-123":
return linked
raise AssertionError(f"unexpected node_id {node_id}")
def fake_read_by_path(args, allow_agent=False):
if args["path"] == "/notes/linked":
return linked
if args["path"] == "/notes/backlink":
return backlink
raise RuntimeError("HTTP 404")
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
if path == "/v1/projects/project-1/nodes/root-123/backlinks":
return {"items": [backlink]}
if path == "/v1/projects/project-1/nodes/linked-123/backlinks":
return {"items": []}
raise AssertionError(path)
mod.get_node_by_id = fake_get_node_by_id
mod.read_node_by_path = fake_read_by_path
mod.request_json = fake_request
mod.list_project_nodes = lambda args: {"items": [root, linked]}
context = mod.get_memory_context({"node_id": "root-123", "depth": 1, "view": "paths"})
assert [item["full_path"] for item in context["items"]] == ["/notes/root", "/notes/linked", "/notes/backlink"]
assert {"source": "/notes/root", "target": "/notes/linked", "relation": "references"} in context["edges"]
assert {"source": "/notes/backlink", "target": "/notes/root", "relation": "backlink"} in context["edges"]
related = mod.find_related_nodes({"node_id": "root-123", "view": "paths"})
assert related["target"]["full_path"] == "/notes/root"
assert {item["full_path"] for item in related["items"]} >= {"/notes/backlink", "/notes/linked"}
with pytest.raises(RuntimeError, match="paths are index/routes only"):
mod.get_memory_context({"path": "/notes/root"})
with pytest.raises(RuntimeError, match="path is an index/route only"):
mod.find_related_nodes({"path": "/notes/root"})
def test_discovery_rejects_full_content_view(mod):
with pytest.raises(RuntimeError, match="full node bodies require nap_node_get"):
mod.list_project_nodes({"view": "full"})