WorkFlowy MCP Server

Instructions

Implementation Reference

• src/workflowy_mcp/client/api_client_nexus.py:624-873 (handler)

  Core handler: orchestrates IGNITE SHARDS logic - indexes coarse terrain, performs targeted deep API exports on root_ids, constructs phantom_gem.json by splicing full subtrees into terrain skeleton while preserving truncation metadata

  async def nexus_ignite_shards(
      self,
      nexus_tag: str,
      root_ids: list[str],
      max_depth: int | None = None,
      child_limit: int | None = None,
      per_root_limits: dict[str, dict[str, int]] | None = None,
  ) -> dict[str, Any]:
      """IGNITE SHARDS → phantom_gem.json."""
      logger = _ClientLogger()
  
      run_dir = self._get_nexus_dir(nexus_tag)
      coarse_path = os.path.join(run_dir, "coarse_terrain.json")
      phantom_path = os.path.join(run_dir, "phantom_gem.json")
  
      if not os.path.exists(coarse_path):
          raise NetworkError("coarse_terrain.json not found; run nexus_scry first")
  
      if not root_ids:
          empty = {"nexus_tag": nexus_tag, "roots": [], "nodes": []}
          with open(phantom_path, "w", encoding="utf-8") as f:
              json.dump(empty, f, indent=2, ensure_ascii=False)
          return {
              "success": True,
              "nexus_tag": nexus_tag,
              "phantom_gem": phantom_path,
              "roots": [],
              "node_count": 0,
          }
  
      # Read coarse terrain
      with open(coarse_path, "r", encoding="utf-8") as f:
          coarse_data = json.load(f)
  
      terrain_nodes = coarse_data.get("nodes", [])
      export_root_id = coarse_data.get("export_root_id")
  
      # Build ledger
      original_ids_seen: set[str] = set()
      if isinstance(coarse_data.get("original_ids_seen"), list):
          original_ids_seen.update(str(nid) for nid in coarse_data.get("original_ids_seen", []) if nid)
      else:
          def _collect(nodes: list[dict[str, Any]]) -> None:
              for node in nodes or []:
                  if isinstance(node, dict):
                      nid = node.get("id")
                      if nid:
                          original_ids_seen.add(str(nid))
                      _collect(node.get("children") or [])
          _collect(terrain_nodes)
          if export_root_id:
              original_ids_seen.add(str(export_root_id))
  
      # Build indexes
      parent_by_id: dict[str, str | None] = {}
      node_by_id: dict[str, dict[str, Any]] = {}
  
      def _index(nodes: list[dict[str, Any]], parent_id: str | None) -> None:
          for node in nodes:
              nid = node.get("id")
              if nid:
                  parent_by_id[nid] = parent_id
                  node_by_id[nid] = node
                  children = node.get("children") or []
                  if children:
                      _index(children, nid)
  
      _index(terrain_nodes, export_root_id)
  
      # Normalize roots
      unique_root_ids = []
      for rid in root_ids:
          if rid not in unique_root_ids:
              unique_root_ids.append(rid)
      roots_set = set(unique_root_ids)
  
      # Check existence
      missing = [rid for rid in roots_set if rid not in parent_by_id]
      if missing:
          raise NetworkError(f"Roots not in coarse_terrain: {missing}")
  
      # Enforce disjointness
      for rid in roots_set:
          parent = parent_by_id.get(rid)
          while parent is not None:
              if parent in roots_set:
                  raise NetworkError(
                      f"Roots not disjoint: '{rid}' descends from '{parent}'"
                  )
              parent = parent_by_id.get(parent)
  
      # Deep SCRY for each root
      deep_subtrees: dict[str, dict[str, Any]] = {}
      roots_resolved = []
      total_nodes = 0
      per_root_limits = per_root_limits or {}
  
      for root_id in unique_root_ids:
          limits = per_root_limits.get(root_id, {})
          root_max_depth = limits.get("max_depth", max_depth)
          root_child_limit = limits.get("child_limit", child_limit)
  
          try:
              raw = await export_nodes_impl(self, node_id=root_id)
          except Exception as e:
              logger.error(f"Export failed for {root_id}: {e}")
              continue
  
          flat = raw.get("nodes", [])
          if not flat:
              continue
  
          total_nodes += raw.get("_total_fetched_from_api", len(flat))
  
          tree = self._build_hierarchy(flat, True)
          if not tree:
              continue
  
          root_subtree = next((c for c in tree if c.get("id") == root_id), tree[0])
  
          self._annotate_child_counts_and_truncate(
              [root_subtree],
              max_depth=root_max_depth,
              child_count_limit=root_child_limit,
              current_depth=1,
          )
  
          deep_subtrees[root_id] = root_subtree
          roots_resolved.append(root_id)
  
          # Extend ledger
          def _collect_subtree(node: dict[str, Any]) -> None:
              if isinstance(node, dict):
                  nid = node.get("id")
                  if nid:
                      original_ids_seen.add(str(nid))
                  for child in node.get("children") or []:
                      _collect_subtree(child)
          _collect_subtree(root_subtree)
  
      # Build GEM skeleton
      if not roots_resolved:
          phantom_payload = {
              "nexus_tag": nexus_tag,
              "roots": [],
              "export_root_id": export_root_id,
              "export_root_name": coarse_data.get("export_root_name", "Root"),
              "original_ids_seen": sorted(original_ids_seen),
              "nodes": [],
          }
      else:
          skeleton_by_id: dict[str, dict[str, Any]] = {}
          top_level_ids: list[str] = []
          top_level_nodes: list[dict[str, Any]] = []
  
          def ensure_skeleton(nid: str) -> dict[str, Any]:
              source = node_by_id.get(nid)
              if not source:
                  raise NetworkError(f"Node {nid} not in coarse_terrain")
              if nid in skeleton_by_id:
                  return skeleton_by_id[nid]
              skel = {k: v for k, v in source.items() if k != "children"}
              skel["children"] = []
              skeleton_by_id[nid] = skel
              return skel
  
          for root_id in roots_resolved:
              path: list[str] = []
              cur = root_id
              while cur is not None and cur != export_root_id:
                  path.append(cur)
                  cur = parent_by_id.get(cur)
              if cur != export_root_id:
                  raise NetworkError(f"Root {root_id} doesn't descend from export_root")
              path.reverse()
  
              parent_id_iter = export_root_id
              for nid in path:
                  node_skel = ensure_skeleton(nid)
                  if parent_id_iter == export_root_id:
                      if nid not in top_level_ids:
                          top_level_ids.append(nid)
                          top_level_nodes.append(node_skel)
                  else:
                      parent_skel = skeleton_by_id[parent_id_iter]
                      children_list = parent_skel.get("children") or []
                      if not any(isinstance(c, dict) and c.get("id") == nid for c in children_list):
                          children_list.append(node_skel)
                          parent_skel["children"] = children_list
                  parent_id_iter = nid
  
          # Replace skeletons with deep subtrees
          for root_id, subtree in deep_subtrees.items():
              parent_id = parent_by_id.get(root_id)
              if parent_id is None or parent_id == export_root_id:
                  for idx, nid in enumerate(top_level_ids):
                      if nid == root_id:
                          top_level_nodes[idx] = subtree
                          break
              else:
                  parent_skel = skeleton_by_id.get(parent_id)
                  if parent_skel:
                      children_list = parent_skel.get("children") or []
                      for idx, child in enumerate(children_list):
                          if isinstance(child, dict) and child.get("id") == root_id:
                              children_list[idx] = subtree
                              break
              skeleton_by_id[root_id] = subtree
  
          # Compute preview
          phantom_preview = None
          try:
              phantom_preview = self._annotate_preview_ids_and_build_tree(top_level_nodes, "PG")
          except Exception:
              pass
          
          phantom_payload = {
              "nexus_tag": nexus_tag,
              "__preview_tree__": phantom_preview,
              "export_root_id": export_root_id,
              "export_root_name": coarse_data.get("export_root_name", "Root"),
              "nodes": top_level_nodes,
              "roots": roots_resolved,
              "original_ids_seen": sorted(original_ids_seen),
          }
  
      # Update coarse terrain ledger
      try:
          coarse_data["original_ids_seen"] = sorted(original_ids_seen)
          with open(coarse_path, "w", encoding="utf-8") as f:
              json.dump(coarse_data, f, indent=2, ensure_ascii=False)
      except Exception:
          logger.warning("Failed to update coarse_terrain ledger")
  
      _log_to_file_helper(
          f"nexus_ignite_shards[{nexus_tag}]: roots={roots_resolved}, node_count={total_nodes}",
          "nexus",
      )
  
      with open(phantom_path, "w", encoding="utf-8") as f:
          json.dump(phantom_payload, f, indent=2, ensure_ascii=False)
  
      return {
          "success": True,
          "nexus_tag": nexus_tag,
          "phantom_gem": phantom_path,
          "roots": roots_resolved,
          "node_count": total_nodes,
      }

• src/workflowy_mcp/server.py:1450-1490 (registration)

  MCP tool registration: @mcp.tool decorator binds 'nexus_ignite_shards' name to handler that delegates to WorkFlowyClient.nexus_ignite_shards

      name="nexus_ignite_shards",
      description=(
          "IGNITE selected SHARDS so the ETHER glows more deeply around them, revealing "
          "deeper layers (but not necessarily to FULL depth). The deeper revelation is "
          "captured as a PHANTOM GEM (S0), an unrefracted witness of those subtrees."
      ),
  )
  async def nexus_ignite_shards(
      nexus_tag: str,
      root_ids: list[str],
      max_depth: int | None = None,
      child_limit: int | None = None,
      per_root_limits: dict[str, dict[str, int]] | None = None,
  ) -> dict:
      """IGNITE SHARDS in the TERRAIN so the ETHER glows more deeply around them.
  
      From an existing TERRAIN, mark specific nodes as SHARDS and IGNITE them. The
      ETHER glows around these SHARDS, revealing deeper layers (but not necessarily
      to full depth). The revealed structure is condensed into a PHANTOM GEM (S0).
      """
      client = get_client()
  
      if _rate_limiter:
          await _rate_limiter.acquire()
  
      try:
          result = await client.nexus_ignite_shards(
              nexus_tag=nexus_tag,
              root_ids=root_ids,
              max_depth=max_depth,
              child_limit=child_limit,
              per_root_limits=per_root_limits,
          )
          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