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"), }