WorkFlowy MCP Server

Instructions

Implementation Reference

• src/workflowy_mcp/server.py:1335-1336 (registration)

  MCP tool registration decorator for 'workflowy_export_node' tool.

  @mcp.tool(name="workflowy_export_node", description="Export a WorkFlowy node with all its children")
  async def export_node(

• src/workflowy_mcp/server.py:1337-1363 (handler)

  Primary handler function for the MCP tool. Applies rate limiting and delegates to WorkFlowyClient.export_nodes.

      node_id: str | None = None,
  ) -> dict:
      """Export all nodes or filter to specific node's subtree.
  
      Args:
          node_id: ID of the node to export (omit to export all nodes).
                   If provided, exports only that node and all its descendants.
  
      Returns:
          Dictionary containing 'nodes' list with exported node data.
          Rate limit: 1 request per minute for full export.
      """
      client = get_client()
  
      if _rate_limiter:
          await _rate_limiter.acquire()
  
      try:
          data = await client.export_nodes(node_id)
          if _rate_limiter:
              _rate_limiter.on_success()
          return data
      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_etch.py:72-89 (handler)

  WorkFlowyClient.export_nodes method - thin wrapper delegating to export_nodes_impl.

      self,
      node_id: str | None = None,
      max_retries: int = 10,
      use_cache: bool = True,
      force_refresh: bool = False,
  ) -> dict[str, Any]:
      """Export all nodes or filter to specific node's subtree.
      
      Delegates to module-level export_nodes_impl to avoid circular imports.
      """
      return await export_nodes_impl(
          self,
          node_id=node_id,
          max_retries=max_retries,
          use_cache=use_cache,
          force_refresh=force_refresh,
      )

• src/workflowy_mcp/client/api_client_etch.py:819-945 (handler)

  Core implementation: handles /nodes-export API calls with caching, dirty ID tracking, retries, dewhitening names/notes, and subtree filtering.

  async def export_nodes_impl(
      client: WorkFlowyClientCore,
      node_id: str | None = None,
      max_retries: int = 10,
      use_cache: bool = True,
      force_refresh: bool = False,
  ) -> dict[str, Any]:
      """Export all nodes or filter to specific node's subtree (implementation).
      
      This is extracted as a module-level function to avoid circular imports.
      """
      import asyncio
  
      logger = _ClientLogger()
  
      async def fetch_and_cache() -> dict[str, Any]:
          """Call /nodes-export with retries and update cache."""
          retry_count = 0
          base_delay = 1.0
  
          while retry_count < max_retries:
              await asyncio.sleep(API_RATE_LIMIT_DELAY)
  
              try:
                  response = await client.client.get("/nodes-export")
                  data = await client._handle_response(response)
  
                  all_nodes = data.get("nodes", []) or []
  
                  # Dewhiten names/notes
                  for node in all_nodes:
                      for key in ("name", "nm", "note", "no"):
                          if key in node:
                              node[key] = client._dewhiten_text(node.get(key))
  
                  total_before_filter = len(all_nodes)
                  data["_total_fetched_from_api"] = total_before_filter
  
                  # Update cache
                  client._nodes_export_cache = data
                  client._nodes_export_cache_timestamp = datetime.now()
                  client._nodes_export_dirty_ids.clear()
  
                  if retry_count > 0:
                      success_msg = f"export_nodes succeeded after {retry_count + 1}/{max_retries} attempts"
                      logger.info(success_msg)
                      _log_to_file_helper(success_msg, "reconcile")
  
                  return data
  
              except Exception as e:
                  retry_count += 1
                  logger.warning(f"Export error: {e}. Retry {retry_count}/{max_retries}")
                  if retry_count < max_retries:
                      await asyncio.sleep(base_delay * (2 ** retry_count))
                  else:
                      raise
  
          raise NetworkError("export_nodes failed after maximum retries")
  
      # Decide cache vs fetch
      if (not use_cache) or force_refresh or client._nodes_export_cache is None:
          data = await fetch_and_cache()
      else:
          data = client._nodes_export_cache
  
          # Check dirty IDs for subtree requests
          if node_id is not None and client._nodes_export_dirty_ids:
              all_nodes = data.get("nodes", []) or []
              nodes_by_id = {n.get("id"): n for n in all_nodes if n.get("id")}
  
              if node_id not in nodes_by_id:
                  logger.info(f"Node {node_id} not in cache; refreshing")
                  data = await fetch_and_cache()
              else:
                  dirty = client._nodes_export_dirty_ids
                  path_hits_dirty = False
                  cur = node_id
                  visited: set[str] = set()
  
                  while cur and cur not in visited:
                      visited.add(cur)
                      if cur in dirty or "*" in dirty:
                          path_hits_dirty = True
                          break
                      parent_id = (
                          nodes_by_id[cur].get("parent_id")
                          or nodes_by_id[cur].get("parentId")
                      )
                      cur = parent_id
  
                  if path_hits_dirty:
                      logger.info(f"Path from {node_id} hits dirty; refreshing")
                      data = await fetch_and_cache()
                  else:
                      logger.info(f"Using cached export for {node_id}")
  
      # Filter if needed
      all_nodes = data.get("nodes", []) or []
      total_before_filter = len(all_nodes)
  
      if node_id is None:
          if "_total_fetched_from_api" not in data:
              data["_total_fetched_from_api"] = total_before_filter
          return data
  
      # Filter to subtree
      included_ids = {node_id}
      nodes_by_id = {node["id"]: node for node in all_nodes if node.get("id")}
  
      def add_descendants(parent_id: str) -> None:
          for node in all_nodes:
              if node.get("parent_id") == parent_id and node["id"] not in included_ids:
                  included_ids.add(node["id"])
                  add_descendants(node["id"])
  
      if node_id in nodes_by_id:
          add_descendants(node_id)
  
      filtered_nodes = [node for node in all_nodes if node["id"] in included_ids]
  
      return {
          "nodes": filtered_nodes,
          "_total_fetched_from_api": data.get("_total_fetched_from_api", total_before_filter),
          "_filtered_count": len(filtered_nodes),
      }