Databricks MCP Server

Instructions

Implementation Reference

• src/server/databricks_mcp_server.py:67-74 (handler)

  MCP tool handler for 'create_cluster'. Invokes the core clusters.create_cluster API and formats the response as TextContent for the model.

  async def create_cluster(params: Dict[str, Any]) -> List[TextContent]:
      logger.info(f"Creating cluster with params: {params}")
      try:
          result = await clusters.create_cluster(params)
          return [{"text": json.dumps(result)}]
      except Exception as e:
          logger.error(f"Error creating cluster: {str(e)}")
          return [{"text": json.dumps({"error": str(e)})}]

• src/server/databricks_mcp_server.py:63-66 (registration)

  Registers the 'create_cluster' tool in the FastMCP server, specifying the name and description which outlines the expected input parameters (serving as schema).

  @self.tool(
      name="create_cluster",
      description="Create a new Databricks cluster with parameters: cluster_name (required), spark_version (required), node_type_id (required), num_workers, autotermination_minutes",
  )

• src/api/clusters.py:14-28 (helper)

  Core helper function that performs the actual Databricks API call to create a cluster using make_api_request.

  async def create_cluster(cluster_config: Dict[str, Any]) -> Dict[str, Any]:
      """
      Create a new Databricks cluster.
      
      Args:
          cluster_config: Cluster configuration
          
      Returns:
          Response containing the cluster ID
          
      Raises:
          DatabricksAPIError: If the API request fails
      """
      logger.info("Creating new cluster")
      return make_api_request("POST", "/api/2.0/clusters/create", data=cluster_config)