WorkFlowy MCP Server

Instructions

Implementation Reference

• src/workflowy_mcp/server.py:1983-2044 (registration)

  MCP tool registration for 'workflowy_glimpse' with handler wrapper that delegates to WorkFlowyClient.workflowy_glimpse, handling WebSocket connection and rate limiting.
  name="workflowy_glimpse", description="Load entire node tree into context (no file intermediary). GLIMPSE command for direct context loading. Optional output_file writes TERRAIN export (WebSocket+API merge with full NEXUS semantics)." ) async def glimpse( node_id: str, output_file: str | None = None, ) -> dict: """Load entire node tree into agent context. GLIMPSE command - efficient context loading for agent analysis. Tries WebSocket DOM extraction first (if Chrome extension connected). Falls back to API fetch if WebSocket unavailable. Args: node_id: Root node UUID to read from output_file: Optional absolute path; if provided, write a TERRAIN-style export package using shared nexus_helper functions (WebSocket GLIMPSE + API SCRY merged for has_hidden_children / children_status annotations, original_ids_seen ledger, full NEXUS TERRAIN format) Returns: When output_file is None: Minimal in-memory preview with only preview_tree + basic stats (no full JSON node tree). When output_file is provided: Compact summary with terrain_file, markdown_file, stats. """ client = get_client() ws_conn, ws_queue = get_ws_connection() # Check if extension is connected if _rate_limiter: await _rate_limiter.acquire() try: # Pass WebSocket connection, queue, AND output_file to client method result = await client.workflowy_glimpse( node_id, output_file=output_file, _ws_connection=ws_conn, _ws_queue=ws_queue, ) if _rate_limiter: _rate_limiter.on_success() # For in-memory GLIMPSE results, return only the MINITREE to the agent. # SCRY-to-disk paths (output_file is not None) continue to carry the # full JSON tree on disk only (coarse_terrain/phantom_gem/etc.). if output_file is None and isinstance(result, dict) and "preview_tree" in result: return { "success": result.get("success", True), "_source": result.get("_source"), "node_count": result.get("node_count"), "depth": result.get("depth"), "preview_tree": result.get("preview_tree"), } return result except Exception as e: 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:177-242 (handler)

  Core handler logic for workflowy_glimpse: prefers WebSocket DOM extraction via Chrome extension, falls back with error. Builds preview_tree, logs to file, and handles TERRAIN export via _write_glimpse_as_terrain if output_file provided.
  async def workflowy_glimpse( self, node_id: str, use_efficient_traversal: bool = False, output_file: str | None = None, _ws_connection=None, _ws_queue=None ) -> dict[str, Any]: """Load node tree via WebSocket (GLIMPSE command).""" import asyncio import json as json_module logger = _ClientLogger() # Try WebSocket first if _ws_connection and _ws_queue: try: logger.info(f"🔌 WebSocket extraction for {node_id[:8]}...") request = {"action": "extract_dom", "node_id": node_id} await _ws_connection.send(json_module.dumps(request)) response = await asyncio.wait_for(_ws_queue.get(), timeout=5.0) if response.get('success') and 'children' in response: response['_source'] = 'websocket' logger.info("✅ WebSocket GLIMPSE successful") # Attach preview_tree preview_tree = None try: root_obj = response.get("root") or {} children = response.get("children") or [] if isinstance(root_obj, dict): root_obj.setdefault("children", children) preview_tree = self._annotate_preview_ids_and_build_tree([root_obj], "WG") else: preview_tree = self._annotate_preview_ids_and_build_tree(children, "WG") except Exception: pass if preview_tree is not None: old = response response = { "success": old.get("success"), "_source": old.get("_source"), "node_count": old.get("node_count"), "depth": old.get("depth"), "preview_tree": preview_tree, "root": old.get("root"), "children": old.get("children"), } _log_glimpse_to_file("glimpse", node_id, response) # Write TERRAIN if requested if output_file is not None: return await self._write_glimpse_as_terrain(node_id, response, output_file) return response else: raise NetworkError("WebSocket invalid response") except asyncio.TimeoutError: raise NetworkError("WebSocket timeout (5s)") except Exception as e: raise NetworkError(f"WebSocket error: {e}") from e raise NetworkError("WebSocket unavailable - use workflowy_scry")
• src/workflowy_mcp/client/nexus_helper.py:93-132 (helper)

  Shared helper: merge_glimpses_for_terrain - detects hidden children by comparing WebSocket GLIMPSE (UI expansion) vs API GLIMPSE FULL, used in TERRAIN export when output_file provided.
  def merge_glimpses_for_terrain( workflowy_root_id: str, ws_glimpse: JsonDict, api_glimpse: Optional[JsonDict] = None, ) -> Tuple[List[JsonDict], str, bool]: """Merge WebSocket + API GLIMPSE results to annotate hidden children. This is the core logic that NEXUS uses to produce safe TERRAIN data from a WebSocket GLIMPSE (what Dan has expanded in the UI) and an optional API GLIMPSE FULL (the complete tree regardless of expansion state). When API data is available and successful: - Each WebSocket node gets ``has_hidden_children`` and ``children_status`` set by comparing its child count to the corresponding API node. - The root's children_status is computed similarly by comparing the top-level child counts. When API data is unavailable or failed: - All nodes are marked via :func:`mark_all_nodes_as_potentially_truncated`. - ``root_children_status`` is set to ``"truncated_by_expansion"``. Args: workflowy_root_id: UUID of the Workflowy root being glimpsed. ws_glimpse: Result dict from the WebSocket GLIMPSE (Dan's UI view). api_glimpse: Optional result dict from ``workflowy_scry`` / GLIMPSE FULL. If ``None`` or unsuccessful, the conservative fallback is used. Returns: (children, root_children_status, api_merge_performed) - children: List of child nodes to be used under ``"nodes"`` in the TERRAIN wrapper (mutated in-place from ``ws_glimpse['children']``). - root_children_status: ``"complete"`` or ``"truncated_by_expansion"`` describing the root's immediate children. - api_merge_performed: ``True`` if we actually compared against API structure; ``False`` if we fell back to WebSocket-only semantics. """