Databricks MCP Server

Instructions

Implementation Reference

• src/server/databricks_mcp_server.py:102-113 (handler)

  MCP tool handler for 'start_cluster'. Registers the tool using @self.tool decorator and implements the execution logic by calling the underlying clusters.start_cluster API and formatting the response as MCP TextContent.

  @self.tool(
      name="start_cluster",
      description="Start a terminated Databricks cluster with parameter: cluster_id (required)",
  )
  async def start_cluster(params: Dict[str, Any]) -> List[TextContent]:
      logger.info(f"Starting cluster with params: {params}")
      try:
          result = await clusters.start_cluster(params.get("cluster_id"))
          return [{"text": json.dumps(result)}]
      except Exception as e:
          logger.error(f"Error starting cluster: {str(e)}")
          return [{"text": json.dumps({"error": str(e)})}]

• src/api/clusters.py:79-93 (handler)

  Core implementation of cluster starting logic. Makes a POST request to the Databricks Clusters API endpoint /api/2.0/clusters/start with the cluster_id.

  async def start_cluster(cluster_id: str) -> Dict[str, Any]:
      """
      Start a terminated Databricks cluster.
      
      Args:
          cluster_id: ID of the cluster to start
          
      Returns:
          Empty response on success
          
      Raises:
          DatabricksAPIError: If the API request fails
      """
      logger.info(f"Starting cluster: {cluster_id}")
      return make_api_request("POST", "/api/2.0/clusters/start", data={"cluster_id": cluster_id})

• src/server/databricks_mcp_server.py:102-105 (registration)

  Registration of the 'start_cluster' tool in the DatabricksMCPServer using the FastMCP @tool decorator, specifying name and description.

  @self.tool(
      name="start_cluster",
      description="Start a terminated Databricks cluster with parameter: cluster_id (required)",
  )