list_volume_files
Retrieve detailed directory listings, including file names, sizes, and modification times, from a Unity Catalog volume to manage large datasets efficiently.
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.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| volume_path | Yes |
Implementation Reference
- MCP tool handler function decorated with @mcp.tool(). Calls volumes.list_volume_files() and returns JSON response. This is the entrypoint for the MCP tool.@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 implementation using Databricks SDK WorkspaceClient.files.list_directory_contents(). Formats file info into standardized dict response with files list containing path, is_directory, size, and last_modified.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)}