Databricks MCP Server

Instructions

Implementation Reference

• src/server/databricks_mcp_server.py:50-61 (handler)

  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)})}]