list_clusters
Retrieve and display all active Databricks clusters to manage compute resources and monitor cluster status.
Instructions
List all Databricks clusters
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| params | Yes |
Implementation Reference
- MCP tool handler for 'list_clusters': registers the tool and implements the logic by calling the clusters.list_clusters API wrapper and formatting the JSON response.
@self.tool( name="list_clusters", description="List all Databricks clusters", ) async def list_clusters(params: Dict[str, Any]) -> List[TextContent]: logger.info(f"Listing clusters with params: {params}") try: result = await clusters.list_clusters() return [{"text": json.dumps(result)}] except Exception as e: logger.error(f"Error listing clusters: {str(e)}") return [{"text": json.dumps({"error": str(e)})}] - src/api/clusters.py:48-59 (helper)Core API wrapper function that performs the actual Databricks API GET request to /api/2.0/clusters/list to retrieve the list of clusters.
async def list_clusters() -> Dict[str, Any]: """ List all Databricks clusters. Returns: Response containing a list of clusters Raises: DatabricksAPIError: If the API request fails """ logger.info("Listing all clusters") return make_api_request("GET", "/api/2.0/clusters/list") - src/server/databricks_mcp_server.py:50-61 (registration)Tool registration via @self.tool decorator specifying name 'list_clusters' and description.
@self.tool( name="list_clusters", description="List all Databricks clusters", ) async def list_clusters(params: Dict[str, Any]) -> List[TextContent]: logger.info(f"Listing clusters with params: {params}") try: result = await clusters.list_clusters() return [{"text": json.dumps(result)}] except Exception as e: logger.error(f"Error listing clusters: {str(e)}") return [{"text": json.dumps({"error": str(e)})}]