Databricks MCP Server

Instructions

List files and directories in a Unity Catalog volume.

Args:
    volume_path: Volume path to list (e.g. '/Volumes/catalog/schema/volume/directory')

Returns:
    JSON with directory listing including file names, sizes, and modification times.
    
Example:
    # List files in volume directory
    files = list_volume_files('/Volumes/kbqa/stark_mas_eval/stark_raw_data/')
    
Note: Returns detailed file information including sizes for managing large datasets.

Implementation Reference

• src/server/simple_databricks_mcp_server.py:534-558 (handler)

  MCP tool registration and handler for 'list_volume_files'. Uses @mcp.tool() decorator. Includes tool description/docstring that serves as schema. Delegates to volumes.list_volume_files for core logic.

  @mcp.tool()
  async def list_volume_files(volume_path: str) -> str:
      """
      List files and directories in a Unity Catalog volume.
  
      Args:
          volume_path: Volume path to list (e.g. '/Volumes/catalog/schema/volume/directory')
  
      Returns:
          JSON with directory listing including file names, sizes, and modification times.
          
      Example:
          # List files in volume directory
          files = list_volume_files('/Volumes/kbqa/stark_mas_eval/stark_raw_data/')
          
      Note: Returns detailed file information including sizes for managing large datasets.
      """
      logger.info(f"Listing volume files in: {volume_path}")
      try:
          result = volumes.list_volume_files(volume_path)
          return json.dumps(result)
      except Exception as e:
          logger.error(f"Error listing volume files: {str(e)}")
          return json.dumps({"error": str(e)})

• src/api/volumes.py:98-137 (helper)

  Core helper function implementing the volume file listing logic using Databricks SDK WorkspaceClient.files.list_directory_contents.

  def list_volume_files(volume_path: str) -> Dict[str, Any]:
      """
      List files and directories in a Unity Catalog volume.
      
      Args:
          volume_path: Volume path to list (e.g. '/Volumes/catalog/schema/volume/directory')
          
      Returns:
          Response containing the directory listing with files and subdirectories
          
      Raises:
          Exception: If the SDK request fails
      """
      logger.info(f"Listing volume files in: {volume_path}")
      
      try:
          w = _get_workspace_client()
          
          # List directory contents using SDK
          files = w.files.list_directory_contents(directory_path=volume_path)
          
          # Convert to dict format similar to REST API response
          file_list = []
          for file_info in files:
              file_list.append({
                  "path": file_info.path,
                  "is_directory": file_info.is_directory,
                  "file_size": file_info.file_size,
                  "last_modified": file_info.last_modified
              })
          
          return {
              "files": file_list,
              "path": volume_path
          }
          
      except Exception as e:
          logger.error(f"Error listing volume files: {str(e)}")
          return {"error": str(e)}