list_volume_files
List files and directories in a Databricks Unity Catalog volume to view file names, sizes, and modification times for dataset management.
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 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)}