WorkFlowy MCP Server

Instructions

Implementation Reference

• src/workflowy_mcp/server.py:1395-1447 (registration)

  MCP tool registration for 'nexus_scry': decorator and thin wrapper handler that delegates to WorkFlowyClient.nexus_scry()

      name="nexus_scry",
      description=(
          "INITIATE a CORINTHIAN NEXUS on the ETHER: perform a COARSE SCRY of Workflowy "
          "under a root to reveal a limited TERRAIN (a new geography named by your "
          "NEXUS TAG). Choose max_depth and child_limit carefully—keep them minimal. "
          "Optionally set max_nodes to guard against accidental 1M-node SCRYs. "
          "Later, you will IGNITE the ETHER more deeply on selected SHARDS."
      ),
  )
  async def nexus_scry(
      nexus_tag: str,
      workflowy_root_id: str,
      max_depth: int,
      child_limit: int,
      reset_if_exists: bool = False,
      max_nodes: int | None = None,
  ) -> dict:
      """Tag-scoped SCRY the ETHER under a root to create a coarse TERRAIN bound to a NEXUS TAG.
  
      This reveals a limited TERRAIN—a new geography named by your NEXUS TAG.
      Keep the SCRY shallow: choose max_depth and child_limit carefully. Use
      max_nodes (when non-None) as a hard upper bound on SCRY size; if the tree
      would exceed this many nodes, the SCRY aborts with a clear error instead of
      exporting a massive JSON bundle.
  
      Unlike ``nexus_glimpse(mode="full")``, which may legitimately produce
      T0 = S0 = T1 in one step (because Dan’s UI expansion already encodes the
      GEM/SHIMMERING decision), this tag-scoped SCRY is explicitly **T0-only**:
      it writes only ``coarse_terrain.json``. S0 and T1 are introduced later via
      ``nexus_ignite_shards`` and ``nexus_anchor_gems``.
      """
      client = get_client()
  
      if _rate_limiter:
          await _rate_limiter.acquire()
  
      try:
          result = await client.nexus_scry(
              nexus_tag=nexus_tag,
              workflowy_root_id=workflowy_root_id,
              max_depth=max_depth,
              child_limit=child_limit,
              reset_if_exists=reset_if_exists,
              max_nodes=max_nodes,
          )
          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:557-623 (handler)

  Core implementation of nexus_scry: creates a timestamped nexus run directory, performs a controlled bulk export to coarse_terrain.json using bulk_export_to_file_impl with depth/child limits, and generates a nexus_manifest.json.

  async def nexus_scry(
      self,
      nexus_tag: str,
      workflowy_root_id: str,
      max_depth: int,
      child_limit: int,
      reset_if_exists: bool = False,
      max_nodes: int | None = None,
  ) -> dict[str, Any]:
      """Tag-scoped SCRY → coarse_terrain.json."""
      import shutil
  
      base_dir = Path(
          r"E:\__daniel347x\__Obsidian\__Inking into Mind\--TypingMind\Projects - All\Projects - Individual\TODO\temp\nexus_runs"
      )
      base_dir.mkdir(parents=True, exist_ok=True)
  
      if reset_if_exists:
          suffix = f"__{nexus_tag}"
          for child in base_dir.iterdir():
              if not child.is_dir():
                  continue
              name = child.name
              if name == nexus_tag or name.endswith(suffix):
                  shutil.rmtree(child, ignore_errors=True)
  
      timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
      run_dir = base_dir / f"{timestamp}__{nexus_tag}"
      run_dir.mkdir(parents=True, exist_ok=False)
      coarse_path = run_dir / "coarse_terrain.json"
  
      result = await bulk_export_to_file_impl(
          self,
          node_id=workflowy_root_id,
          output_file=str(coarse_path),
          include_metadata=True,
          use_efficient_traversal=False,
          max_depth=max_depth,
          child_count_limit=child_limit,
          max_nodes=max_nodes,
      )
  
      # Manifest
      manifest_path = run_dir / "nexus_manifest.json"
      try:
          manifest = {
              "nexus_tag": nexus_tag,
              "workflowy_root_id": workflowy_root_id,
              "max_depth": max_depth,
              "child_limit": child_limit,
              "max_nodes": max_nodes,
              "stage": "coarse_terrain",
              "timestamp": datetime.now().isoformat(timespec="seconds"),
          }
          with open(manifest_path, "w", encoding="utf-8") as mf:
              json.dump(manifest, mf, indent=2)
      except Exception:
          pass
  
      return {
          "success": True,
          "nexus_tag": nexus_tag,
          "coarse_terrain": str(coarse_path),
          "node_count": result.get("node_count"),
          "depth": result.get("depth"),
      }