Databricks MCP Server

Instructions

Implementation Reference

• src/api/clusters.py:31-45 (handler)

  Core handler implementing the cluster termination logic by calling the Databricks Clusters API delete endpoint.

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

• src/server/databricks_mcp_server.py:76-87 (registration)

  MCP tool registration and wrapper handler that registers the 'terminate_cluster' tool and delegates to the core API handler.

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

• src/server/app.py:47-50 (helper)

  FastAPI endpoint handler for cluster termination (compatibility stub).

  async def terminate_cluster(request_data: dict):
      """Terminate a cluster."""
      result = await clusters.terminate_cluster(request_data.get("cluster_id"))
      return result