WorkFlowy MCP Server

Instructions

Implementation Reference

• src/workflowy_mcp/server.py:1527-1564 (registration)

  MCP tool registration and thin handler delegating to WorkFlowyClient.nexus_anchor_jewels
  @mcp.tool( name="nexus_anchor_jewels", description=( "Anchor the PHANTOM JEWELS (S1) within the SHIMMERING TERRAIN (T1), " "transmuting the REVEALED GEMS into NEW JEWELS that are an exact " "impregnation of the PHANTOM JEWELS. The TERRAIN becomes ENCHANTED (SECOND " "IMBUE), with the PHANTOM GEM (S0) as witness to the ORIGINAL state. The " "ENCHANTED TERRAIN is now EMBODIED and REAL as JSON—Workflowy remains " "untouched until WEAVE." ), ) async def nexus_anchor_jewels( nexus_tag: str, ) -> dict: """ANCHOR PHANTOM JEWELS into SHIMMERING TERRAIN to create ENCHANTED TERRAIN. This performs the SECOND IMBUE: anchoring PHANTOM JEWELS (S1) within the SHIMMERING TERRAIN (T1), transmuting the REVEALED GEMS into NEW JEWELS that are an exact impregnation of the PHANTOM JEWELS. The TERRAIN becomes ENCHANTED, with the PHANTOM GEM (S0) as witness to the original state. The ENCHANTED TERRAIN is real as JSON, but the ETHER (Workflowy) is still untouched. """ client = get_client() if _rate_limiter: await _rate_limiter.acquire() try: result = await client.nexus_anchor_jewels(nexus_tag=nexus_tag) if _rate_limiter: _rate_limiter.on_success() return result except Exception as e: # noqa: BLE001 if _rate_limiter and hasattr(e, "__class__") and e.__class__.__name__ == "RateLimitError": _rate_limiter.on_rate_limit(getattr(e, "retry_after", None)) raise
• src/workflowy_mcp/client/api_client_nexus.py:1156-1235 (handler)

  Core handler: copies shimmering_terrain.json to enchanted_terrain.json, invokes nexus_json_tools.main(['enchanted_terrain.json', 'fuse-shard-3way', '--witness-shard', 'phantom_gem.json', '--morphed-shard', 'phantom_jewel.json', '--target-scry', 'enchanted_terrain.json']) to perform the 3-way fuse T1 + S0 + S1, then attaches preview_tree.
  async def nexus_anchor_jewels(self, nexus_tag: str) -> dict[str, Any]: """ANCHOR JEWELS → enchanted_terrain.json.""" import shutil import importlib run_dir = self._get_nexus_dir(nexus_tag) shimmering_path = os.path.join(run_dir, "shimmering_terrain.json") phantom_gem_path = os.path.join(run_dir, "phantom_gem.json") phantom_jewel_path = os.path.join(run_dir, "phantom_jewel.json") enchanted_path = os.path.join(run_dir, "enchanted_terrain.json") if not os.path.exists(shimmering_path): raise NetworkError("shimmering_terrain not found") if not os.path.exists(phantom_gem_path): raise NetworkError("phantom_gem not found") if not os.path.exists(phantom_jewel_path): raise NetworkError("phantom_jewel not found - run QUILLSTORM first") # Copy shimmering to enchanted shutil.copy2(shimmering_path, enchanted_path) # Import nexus_json_tools try: client_dir = os.path.dirname(os.path.abspath(__file__)) wf_mcp_dir = os.path.dirname(client_dir) mcp_servers_dir = os.path.dirname(wf_mcp_dir) project_root = os.path.dirname(mcp_servers_dir) if project_root not in sys.path: sys.path.insert(0, project_root) nexus_tools = importlib.import_module("nexus_json_tools") except Exception as e: raise NetworkError(f"Failed to import nexus_json_tools: {e}") from e # Call fuse-shard-3way try: argv = [ enchanted_path, "fuse-shard-3way", "--witness-shard", phantom_gem_path, "--morphed-shard", phantom_jewel_path, "--target-scry", enchanted_path, ] try: nexus_tools.main(argv) except SystemExit as se: if (se.code or 0) != 0: raise NetworkError(f"fuse-shard-3way exited {se.code}") from se except Exception as e: raise NetworkError(f"fuse-shard-3way failed: {e}") from e # Attach enchanted preview try: with open(enchanted_path, "r", encoding="utf-8") as f: enchanted_data = json.load(f) if isinstance(enchanted_data, dict) and isinstance(enchanted_data.get("nodes"), list): enchanted_preview = self._annotate_preview_ids_and_build_tree( enchanted_data.get("nodes") or [], "ET" ) enchanted_data = { "export_timestamp": enchanted_data.get("export_timestamp"), "export_root_children_status": enchanted_data.get("export_root_children_status"), "__preview_tree__": enchanted_preview, "export_root_id": enchanted_data.get("export_root_id"), "export_root_name": enchanted_data.get("export_root_name"), "nodes": enchanted_data.get("nodes"), "original_ids_seen": enchanted_data.get("original_ids_seen"), "explicitly_preserved_ids": enchanted_data.get("explicitly_preserved_ids"), } with open(enchanted_path, "w", encoding="utf-8") as f: json.dump(enchanted_data, f, indent=2, ensure_ascii=False) except Exception: pass return { "success": True, "nexus_tag": nexus_tag, "shimmering_terrain": shimmering_path, "enchanted_terrain": enchanted_path, }
• src/workflowy_mcp/nexus_json_tools.py:1299-1644 (helper)

  Core 3-way shard fuse logic invoked by nexus_anchor_jewels: merges morphed_shard (S1/phantom_jewel) into target_scry (T1/shimmering_terrain) using witness_shard (S0/phantom_gem) to detect explicit changes (creates/deletes/moves/reorders/updates) while preserving hidden Territory nodes.
  def cmd_fuse_shard_3way(args: argparse.Namespace) -> None: """3-way SHARD FUSE: merge S1 into Territory using S0 as witness. Inputs: - Territory T: args.target_scry (full SCRY JSON) - Witness S0: args.witness_shard (original shard after BURN THE LENS) - Morphed S1: args.morphed_shard (edited shard after REFRACTION) Semantics (per root): - Nodes never present in S0 are treated as *hidden* Territory nodes and are always preserved (not deleted or moved). - Nodes present in S0 but not in S1 are treated as explicit deletions: the corresponding subtrees in Territory are removed. - Nodes present in S1 but not in S0 are treated as creations: new subtrees are added under the parents indicated by S1. - Nodes present in both S0 and S1 are updated and possibly reparented to match S1's structure. Sibling order for these "visible" children follows S1, while hidden children keep their original relative order and remain grouped ahead of the visible block for that parent. This guarantees: - No unintended orphaning of hidden descendants. - Explicit delete/move operations are detected as differences between S0 and S1, not inferred from absence in a shallow shard alone. """ log_jewel( f"fuse-shard-3way: target={args.target_scry}, " f"witness={args.witness_shard}, morphed={args.morphed_shard}" ) territory = load_json(args.target_scry) witness = load_json(args.witness_shard) morphed = load_json(args.morphed_shard) witness_roots = witness.get("nodes") or [] morphed_roots = morphed.get("nodes") or [] if not isinstance(witness_roots, list) or not isinstance(morphed_roots, list): die("Both witness and morphed shards must have a 'nodes' list") # Index S0, S1, and Territory s0_by_id, s0_parent, s0_children_ids = _index_shard_roots(witness_roots) s1_by_id, s1_parent, s1_children_ids = _index_shard_roots(morphed_roots) t_by_id, t_parent, t_children_lists = _index_territory(territory) # --- Invariant checks: enforce correct NEXUS pipeline before fusing --- ids_T1: Set[str] = set(t_by_id.keys()) ids_S0: Set[str] = set(s0_by_id.keys()) ids_S1: Set[str] = set(s1_by_id.keys()) # (A) GEM must be fully embedded into TERRAIN: every GEM id should be # present in shimmering_terrain.json. If not, it likely means IGNITE SHARDS # was run after ATTACH GEMS (or ATTACH GEMS was skipped entirely). missing_in_T1 = ids_S0 - ids_T1 if missing_in_T1: msg = ( "nexus_anchor_jewels / fuse-shard-3way invariant violation: " "phantom_gem (S0) contains Workflowy ids that are not present in " "shimmering_terrain (T1). This usually means you ran IGNITE SHARDS " "after ATTACH GEMS or never called ATTACH GEMS for this tag." ) log_jewel(msg + f" Offending ids (sample): {sorted(list(missing_in_T1))[:5]}") die(msg) # (B) JEWEL must not introduce new Workflowy ids that were never in the GEM. # New JEWEL nodes should be id-less; WEAVE will create real ids in ETHER. extra_S1_vs_S0 = ids_S1 - ids_S0 if extra_S1_vs_S0: msg = ( "nexus_anchor_jewels / fuse-shard-3way invariant violation: " "phantom_jewel (S1) contains Workflowy ids that do not appear in " "phantom_gem (S0). JEWELSTORM must not introduce new Workflowy ids; " "only id-less nodes are allowed as new children. Did you run a new " "SCRY/IGNITE or copy ids from another tree after capturing the GEM?" ) log_jewel(msg + f" Offending ids (sample): {sorted(list(extra_S1_vs_S0))[:5]}") die(msg) original_status: Dict[str, Optional[str]] = {} original_truncated: Dict[str, bool] = {} for nid, node in t_by_id.items(): if not nid: continue status = node.get("children_status") # Truncation semantics belong to the TERRAIN layer and are encoded via # children_status. We no longer infer truncation from count fields; the # *_human_readable_only counters are purely informational. original_status[nid] = status original_truncated[nid] = bool(status and status != "complete") affected_ids: Set[str] = set() # Determine which roots to process: intersection of root_ids present in both S0 and S1 s0_root_ids = [n.get("id") for n in witness_roots if isinstance(n, dict) and n.get("id")] s1_root_ids = {n.get("id") for n in morphed_roots if isinstance(n, dict) and n.get("id")} root_ids = [rid for rid in s0_root_ids if rid in s1_root_ids] if not root_ids: die("No overlapping shard roots between witness and morphed shards") updates = 0 creates = 0 deletes = 0 moves = 0 reorders = 0 for root_id in root_ids: if root_id not in t_by_id: print(f"[nexus_json_tools] Warning: root {root_id} not found in Territory; skipping") continue # Collect subtree ids for this root in S0 and S1 ids_s0 = set(_collect_subtree_ids(root_id, s0_children_ids)) if root_id in s0_by_id else set() ids_s1 = set(_collect_subtree_ids(root_id, s1_children_ids)) if root_id in s1_by_id else set() # Partition ids for this shard root delete_ids = list(ids_s0 - ids_s1) new_ids = ids_s1 - ids_s0 common_ids = ids_s0 & ids_s1 # Compute S1 depths for this shard root so that, if S1 ever introduces # new nodes *with* real Workflowy ids, we create parents before # children. In the canonical JEWELSTORM flow, new nodes are id-less and # handled separately below; this depth map is a forward-compatible # safeguard. depth_s1: Dict[str, int] = {} stack: List[Tuple[str, int]] = [(root_id, 0)] if root_id in ids_s1 else [] while stack: cur_id, d = stack.pop() if cur_id in depth_s1: continue depth_s1[cur_id] = d for cid in s1_children_ids.get(cur_id, []): if cid in ids_s1: stack.append((cid, d + 1)) # --- CREATE: nodes present in S1 but not in S0 (id-based) --- # Process in top-down order so that any new parent ids are created # before their new children. ordered_new_ids = sorted(new_ids, key=lambda nid: depth_s1.get(nid, 0)) for nid in ordered_new_ids: s1_node = s1_by_id.get(nid) if s1_node is None: continue parent_id = s1_parent.get(nid) if parent_id is None: parent_children = territory.get("nodes") if not isinstance(parent_children, list): parent_children = [] territory["nodes"] = parent_children else: parent_node = t_by_id.get(parent_id) if parent_node is None: print( f"[nexus_json_tools] Warning: new node {nid} has parent {parent_id} " "which is not present in Territory; skipping creation." ) continue parent_children = parent_node.get("children") if not isinstance(parent_children, list): parent_children = [] parent_node["children"] = parent_children t_children_lists[parent_id] = parent_children new_node = copy.deepcopy(s1_node) parent_children.append(new_node) _register_subtree_in_indexes(new_node, parent_id, t_by_id, t_parent, t_children_lists) affected_ids.update(filter(None, [nid, parent_id])) creates += 1 # --- CREATE: id-less direct children from S1 under existing parents --- # These are new subtrees introduced by JEWELSTORM that intentionally # have no Workflowy id yet. We append them to the corresponding Territory # parents so the reconciliation algorithm can CREATE them. for p, s1_parent_node in s1_by_id.items(): if p not in ids_s1: continue t_parent_node = t_by_id.get(p) if t_parent_node is None: continue t_children = t_parent_node.get("children") if not isinstance(t_children, list): t_children = [] t_parent_node["children"] = t_children t_children_lists[p] = t_children else: t_children_lists.setdefault(p, t_children) # Build a simple signature for existing id-less children to avoid # obvious duplicates when re-running fuse on the same Territory. existing_signatures: set[tuple[Any, Any]] = set() for child in t_children: if not isinstance(child, dict) or child.get("id") is not None: continue existing_signatures.add((child.get("name"), child.get("note"))) for child in s1_parent_node.get("children") or []: if not isinstance(child, dict) or child.get("id") is not None: continue sig = (child.get("name"), child.get("note")) if sig in existing_signatures: continue new_child = copy.deepcopy(child) t_children.append(new_child) existing_signatures.add(sig) affected_ids.add(p) creates += 1 # --- MOVE: nodes in both S0 and S1 whose parent changed --- for nid in common_ids: s1_p = s1_parent.get(nid) t_p = t_parent.get(nid) if s1_p == t_p: continue if t_p is None: old_siblings = territory.get("nodes") or [] else: old_siblings = t_children_lists.get(t_p) or [] old_siblings[:] = [n for n in old_siblings if not (isinstance(n, dict) and n.get("id") == nid)] if t_p is not None: t_children_lists[t_p] = old_siblings if s1_p is None: new_siblings = territory.get("nodes") if not isinstance(new_siblings, list): new_siblings = [] territory["nodes"] = new_siblings else: new_parent_node = t_by_id.get(s1_p) if new_parent_node is None: print( f"[nexus_json_tools] Warning: cannot reparent {nid} to {s1_p} " "(parent not found in Territory); leaving in place." ) continue new_siblings = new_parent_node.get("children") if not isinstance(new_siblings, list): new_siblings = [] new_parent_node["children"] = new_siblings t_children_lists[s1_p] = new_siblings node_obj = t_by_id.get(nid) if node_obj is None: continue new_siblings.append(node_obj) t_parent[nid] = s1_p affected_ids.update(filter(None, [nid, t_p, s1_p])) moves += 1 # --- DELETE: nodes present in S0 but not in S1 (after moves) --- delete_roots = _compute_delete_roots(delete_ids, s0_parent) for nid in delete_roots: t_node = t_by_id.get(nid) if t_node is None: continue parent_id = t_parent.get(nid) if parent_id is None: siblings = territory.get("nodes") or [] else: siblings = t_children_lists.get(parent_id) or [] siblings[:] = [n for n in siblings if not (isinstance(n, dict) and n.get("id") == nid)] _remove_subtree_from_indexes(t_node, t_by_id, t_parent, t_children_lists) if parent_id: affected_ids.add(parent_id) deletes += 1 # --- REORDER: align order of visible children with S1, keep hidden + id-less children --- parents_to_consider = {s1_parent[nid] for nid in ids_s1} | {root_id} parents_to_consider = {p for p in parents_to_consider if p is not None and p in t_by_id} for p in parents_to_consider: desired_ids = s1_children_ids.get(p, []) if not desired_ids: continue children_list = t_children_lists.get(p) if children_list is None: parent_node = t_by_id[p] new_children_list: List[JsonDict] = [] for cid in desired_ids: child_node = t_by_id.get(cid) if child_node is not None: new_children_list.append(child_node) parent_node["children"] = new_children_list t_children_lists[p] = new_children_list if new_children_list: reorders += 1 affected_ids.add(p) continue hidden_children: List[JsonDict] = [] for child in children_list: if not isinstance(child, dict): continue cid = child.get("id") # Preserve id-less children (new JEWELSTORM subtrees and hidden Territory # nodes that never had an id in the shard) as part of the hidden block. if cid is None: hidden_children.append(child) continue # Also preserve Territory nodes that are outside the S0/S1 shard entirely. if cid not in ids_s0 and cid not in ids_s1: hidden_children.append(child) new_children_list: List[JsonDict] = [] new_children_list.extend(hidden_children) for cid in desired_ids: child_node = t_by_id.get(cid) if child_node is not None and child_node not in new_children_list: new_children_list.append(child_node) if new_children_list != children_list: parent_node = t_by_id[p] parent_node["children"] = new_children_list t_children_lists[p] = new_children_list reorders += 1 affected_ids.add(p) # --- UPDATE content for common ids --- for nid in common_ids: t_node = t_by_id.get(nid) s1_node = s1_by_id.get(nid) if t_node is None or s1_node is None: continue t_node["name"] = s1_node.get("name") t_node["note"] = s1_node.get("note") t_node["data"] = s1_node.get("data") t_node["completed"] = bool(s1_node.get("completed", False)) affected_ids.add(nid) updates += 1 # Expand affected set to include ancestors so ancestor counts stay correct affected_ids = {nid for nid in affected_ids if nid} _expand_with_ancestors(affected_ids, t_parent) _recalc_counts_preserving_truncation(territory, affected_ids, original_status, original_truncated) save_json(args.target_scry, territory) summary_msg = ( "[nexus_json_tools] 3-way fuse complete: " f"updates={updates}, creates={creates}, deletes={deletes}, moves={moves}, reorders={reorders}." ) print(summary_msg) log_jewel(summary_msg)