diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/ResourceBackend.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/ResourceBackend.java
index 253fa16..fe6d311 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/ResourceBackend.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/ResourceBackend.java
@@ -66,4 +66,14 @@
     public DropboxResource createDropboxResource(DropboxResourceCreateRequest request) throws Exception;
     public boolean updateDropboxResource(DropboxResourceUpdateRequest request) throws Exception;
     public boolean deleteDropboxResource(DropboxResourceDeleteRequest request) throws Exception;
+
+    Optional<FTPStorage> getFTPStorage(FTPStorageGetRequest request) throws Exception;
+    FTPStorage createFTPStorage(FTPStorageCreateRequest request) throws Exception;
+    boolean updateFTPStorage(FTPStorageUpdateRequest request) throws Exception;
+    boolean deleteFTPStorage(FTPStorageDeleteRequest request) throws Exception;
+
+    Optional<FTPResource> getFTPResource(FTPResourceGetRequest request) throws Exception;
+    FTPResource createFTPResource(FTPResourceCreateRequest request) throws Exception;
+    boolean updateFTPResource(FTPResourceUpdateRequest request) throws Exception;
+    boolean deleteFTPResource(FTPResourceDeleteRequest request) throws Exception;
 }
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/airavata/AiravataResourceBackend.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/airavata/AiravataResourceBackend.java
index 17b0bd1..1f05394 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/airavata/AiravataResourceBackend.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/airavata/AiravataResourceBackend.java
@@ -270,4 +270,44 @@
     public boolean deleteDropboxResource(DropboxResourceDeleteRequest request) throws Exception {
         throw new UnsupportedOperationException("Operation is not supported in backend");
     }
+
+    @Override
+    public Optional<FTPResource> getFTPResource(FTPResourceGetRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
+
+    @Override
+    public FTPResource createFTPResource(FTPResourceCreateRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
+
+    @Override
+    public boolean updateFTPResource(FTPResourceUpdateRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
+
+    @Override
+    public boolean deleteFTPResource(FTPResourceDeleteRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
+
+    @Override
+    public Optional<FTPStorage> getFTPStorage(FTPStorageGetRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
+
+    @Override
+    public FTPStorage createFTPStorage(FTPStorageCreateRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
+
+    @Override
+    public boolean updateFTPStorage(FTPStorageUpdateRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
+
+    @Override
+    public boolean deleteFTPStorage(FTPStorageDeleteRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
 }
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/file/FileBasedResourceBackend.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/file/FileBasedResourceBackend.java
index fd8a73f..ae29eb4 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/file/FileBasedResourceBackend.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/file/FileBasedResourceBackend.java
@@ -25,12 +25,14 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.util.List;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
+@SuppressWarnings("unchecked")
 public class FileBasedResourceBackend implements ResourceBackend {
 
     private static final Logger logger = LoggerFactory.getLogger(FileBasedResourceBackend.class);
@@ -383,4 +385,75 @@
     public boolean deleteDropboxResource(DropboxResourceDeleteRequest request) throws Exception {
         throw new UnsupportedOperationException("Operation is not supported in backend");
     }
+
+    @Override
+    public Optional<FTPResource> getFTPResource(FTPResourceGetRequest request) throws Exception {
+        InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream("resources.json");
+
+        JSONParser jsonParser = new JSONParser();
+
+        if (inputStream == null) {
+            throw new IOException("resources file not found");
+        }
+
+        try (InputStreamReader reader = new InputStreamReader(inputStream)) {
+
+            Object obj = jsonParser.parse(reader);
+
+            JSONArray resourceList = (JSONArray) obj;
+
+            List<FTPResource> ftpResources = (List<FTPResource>) resourceList.stream()
+                    .filter(resource -> "FTP".equals(((JSONObject) resource).get("type").toString()))
+                    .map(resource -> {
+                        JSONObject r = (JSONObject) resource;
+
+                        FTPStorage storage = FTPStorage.newBuilder()
+                                .setStorageId(((JSONObject)r.get("ftpStorage")).get("storageId").toString())
+                                .setHost(((JSONObject)r.get("ftpStorage")).get("host").toString())
+                                .setPort(Integer.parseInt(((JSONObject)r.get("ftpStorage")).get("port").toString())).build();
+
+                        return FTPResource.newBuilder()
+                                .setResourcePath(r.get("resourcePath").toString())
+                                .setResourceId(r.get("resourceId").toString())
+                                .setFtpStorage(storage).build();
+                    }).collect(Collectors.toList());
+
+            return ftpResources.stream().filter(r -> request.getResourceId().equals(r.getResourceId())).findFirst();
+        }
+    }
+
+    @Override
+    public FTPResource createFTPResource(FTPResourceCreateRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
+
+    @Override
+    public boolean updateFTPResource(FTPResourceUpdateRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
+
+    @Override
+    public boolean deleteFTPResource(FTPResourceDeleteRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
+
+    @Override
+    public Optional<FTPStorage> getFTPStorage(FTPStorageGetRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
+
+    @Override
+    public FTPStorage createFTPStorage(FTPStorageCreateRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
+
+    @Override
+    public boolean updateFTPStorage(FTPStorageUpdateRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
+
+    @Override
+    public boolean deleteFTPStorage(FTPStorageDeleteRequest request) {
+        throw new UnsupportedOperationException("Operation is not supported in backend");
+    }
 }
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/SQLResourceBackend.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/SQLResourceBackend.java
index 1718ba8..adea1a0 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/SQLResourceBackend.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/SQLResourceBackend.java
@@ -18,12 +18,8 @@
 package org.apache.airavata.mft.resource.server.backend.sql;
 
 import org.apache.airavata.mft.resource.server.backend.ResourceBackend;
-import org.apache.airavata.mft.resource.server.backend.sql.entity.LocalResourceEntity;
-import org.apache.airavata.mft.resource.server.backend.sql.entity.SCPResourceEntity;
-import org.apache.airavata.mft.resource.server.backend.sql.entity.SCPStorageEntity;
-import org.apache.airavata.mft.resource.server.backend.sql.repository.LocalResourceRepository;
-import org.apache.airavata.mft.resource.server.backend.sql.repository.SCPResourceRepository;
-import org.apache.airavata.mft.resource.server.backend.sql.repository.SCPStorageRepository;
+import org.apache.airavata.mft.resource.server.backend.sql.entity.*;
+import org.apache.airavata.mft.resource.server.backend.sql.repository.*;
 import org.apache.airavata.mft.resource.service.*;
 import org.dozer.DozerBeanMapper;
 import org.slf4j.Logger;
@@ -45,6 +41,12 @@
     @Autowired
     private LocalResourceRepository localResourceRepository;
 
+    @Autowired
+    private FTPResourceRepository ftpResourceRepository;
+
+    @Autowired
+    private FTPStorageRepository ftpStorageRepository;
+
     private DozerBeanMapper mapper = new DozerBeanMapper();
 
     @Override
@@ -236,4 +238,54 @@
         throw new UnsupportedOperationException("Operation is not supported in backend");
     }
 
+    @Override
+    public Optional<FTPResource> getFTPResource(FTPResourceGetRequest request) {
+        Optional<FTPResourceEntity> resourceEntity = ftpResourceRepository.findByResourceId(request.getResourceId());
+
+        return resourceEntity.map(ftpResourceEntity -> mapper.map(ftpResourceEntity, FTPResource.newBuilder().getClass())
+                .setFtpStorage(mapper.map(ftpResourceEntity.getFtpStorage(), FTPStorage.newBuilder().getClass())).build());
+    }
+
+    @Override
+    public FTPResource createFTPResource(FTPResourceCreateRequest request) {
+        FTPResourceEntity savedEntity = ftpResourceRepository.save(mapper.map(request, FTPResourceEntity.class));
+        return getFTPResource(FTPResourceGetRequest.newBuilder().setResourceId(savedEntity.getResourceId()).build()).orElse(null);
+    }
+
+    @Override
+    public boolean updateFTPResource(FTPResourceUpdateRequest request) {
+        ftpResourceRepository.save(mapper.map(request, FTPResourceEntity.class));
+        return true;
+    }
+
+    @Override
+    public boolean deleteFTPResource(FTPResourceDeleteRequest request) {
+        ftpResourceRepository.deleteById(request.getResourceId());
+        return true;
+    }
+
+    @Override
+    public Optional<FTPStorage> getFTPStorage(FTPStorageGetRequest request) {
+        Optional<FTPStorageEntity> storageEty = ftpStorageRepository.findByStorageId(request.getStorageId());
+        return storageEty.map(ftpStorageEntity -> mapper.map(ftpStorageEntity, FTPStorage.newBuilder().getClass()).build());
+    }
+
+    @Override
+    public FTPStorage createFTPStorage(FTPStorageCreateRequest request) {
+        FTPStorageEntity savedEntity = ftpStorageRepository.save(mapper.map(request, FTPStorageEntity.class));
+        return mapper.map(savedEntity, FTPStorage.newBuilder().getClass()).build();
+    }
+
+    @Override
+    public boolean updateFTPStorage(FTPStorageUpdateRequest request) {
+        ftpStorageRepository.save(mapper.map(request, FTPStorageEntity.class));
+        return true;
+    }
+
+    @Override
+    public boolean deleteFTPStorage(FTPStorageDeleteRequest request) {
+        ftpResourceRepository.deleteById(request.getStorageId());
+        return true;
+    }
+
 }
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/entity/FTPResourceEntity.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/entity/FTPResourceEntity.java
new file mode 100644
index 0000000..5aba915
--- /dev/null
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/entity/FTPResourceEntity.java
@@ -0,0 +1,57 @@
+package org.apache.airavata.mft.resource.server.backend.sql.entity;
+
+import org.hibernate.annotations.GenericGenerator;
+
+import javax.persistence.*;
+
+@Entity
+public class FTPResourceEntity {
+
+    @Id
+    @Column(name = "FTP_RESOURCE_ID")
+    @GeneratedValue(generator = "uuid")
+    @GenericGenerator(name = "uuid", strategy = "uuid2")
+    private String resourceId;
+
+    @Column(name = "FTP_STORAGE_ID")
+    private String ftpStorageId;
+
+    @ManyToOne(fetch = FetchType.EAGER)
+    @JoinColumn(name = "FTP_STORAGE_ID", referencedColumnName = "FTP_STORAGE_ID", nullable = false, insertable = false, updatable = false)
+    private FTPStorageEntity ftpStorage;
+
+    @Column(name = "RESOURCE_PATH")
+    private String resourcePath;
+
+    public String getResourceId() {
+        return resourceId;
+    }
+
+    public void setResourceId(String resourceId) {
+        this.resourceId = resourceId;
+    }
+
+    public String getFtpStorageId() {
+        return ftpStorageId;
+    }
+
+    public void setFtpStorageId(String ftpStorageId) {
+        this.ftpStorageId = ftpStorageId;
+    }
+
+    public FTPStorageEntity getFtpStorage() {
+        return ftpStorage;
+    }
+
+    public void setFtpStorage(FTPStorageEntity ftpStorage) {
+        this.ftpStorage = ftpStorage;
+    }
+
+    public String getResourcePath() {
+        return resourcePath;
+    }
+
+    public void setResourcePath(String resourcePath) {
+        this.resourcePath = resourcePath;
+    }
+}
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/entity/FTPStorageEntity.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/entity/FTPStorageEntity.java
new file mode 100644
index 0000000..2c42a8c
--- /dev/null
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/entity/FTPStorageEntity.java
@@ -0,0 +1,48 @@
+package org.apache.airavata.mft.resource.server.backend.sql.entity;
+
+import org.hibernate.annotations.GenericGenerator;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.GeneratedValue;
+import javax.persistence.Id;
+
+@Entity
+public class FTPStorageEntity {
+
+    @Id
+    @Column(name = "FTP_STORAGE_ID")
+    @GeneratedValue(generator = "uuid")
+    @GenericGenerator(name = "uuid", strategy = "uuid2")
+    private String storageId;
+
+    @Column(name = "HOST")
+    private String host;
+
+    @Column(name = "PORT")
+    private int port;
+
+    public String getStorageId() {
+        return storageId;
+    }
+
+    public void setStorageId(String storageId) {
+        this.storageId = storageId;
+    }
+
+    public String getHost() {
+        return host;
+    }
+
+    public void setHost(String host) {
+        this.host = host;
+    }
+
+    public int getPort() {
+        return port;
+    }
+
+    public void setPort(int port) {
+        this.port = port;
+    }
+}
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/repository/FTPResourceRepository.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/repository/FTPResourceRepository.java
new file mode 100644
index 0000000..ad1f855
--- /dev/null
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/repository/FTPResourceRepository.java
@@ -0,0 +1,10 @@
+package org.apache.airavata.mft.resource.server.backend.sql.repository;
+
+import org.apache.airavata.mft.resource.server.backend.sql.entity.FTPResourceEntity;
+import org.springframework.data.repository.CrudRepository;
+
+import java.util.Optional;
+
+public interface FTPResourceRepository extends CrudRepository<FTPResourceEntity, String> {
+    Optional<FTPResourceEntity> findByResourceId(String resourceId);
+}
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/repository/FTPStorageRepository.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/repository/FTPStorageRepository.java
new file mode 100644
index 0000000..c80187a
--- /dev/null
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/repository/FTPStorageRepository.java
@@ -0,0 +1,10 @@
+package org.apache.airavata.mft.resource.server.backend.sql.repository;
+
+import org.apache.airavata.mft.resource.server.backend.sql.entity.FTPStorageEntity;
+import org.springframework.data.repository.CrudRepository;
+
+import java.util.Optional;
+
+public interface FTPStorageRepository extends CrudRepository<FTPStorageEntity, String> {
+    Optional<FTPStorageEntity> findByStorageId(String storageId);
+}
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/ResourceServiceHandler.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/ResourceServiceHandler.java
index 7e0ae28..1b66319 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/ResourceServiceHandler.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/ResourceServiceHandler.java
@@ -571,4 +571,139 @@
                     .asRuntimeException());
         }
     }
+
+    @Override
+    public void getFTPStorage(FTPStorageGetRequest request, StreamObserver<FTPStorage> responseObserver) {
+        try {
+            this.backend.getFTPStorage(request).ifPresentOrElse(storage -> {
+                responseObserver.onNext(storage);
+                responseObserver.onCompleted();
+            }, () -> responseObserver.onError(Status.INTERNAL
+                    .withDescription("No FTP Storage with id " + request.getStorageId())
+                    .asRuntimeException()));
+        } catch (Exception e) {
+            logger.error("Failed in retrieving FTP storage with id " + request.getStorageId(), e);
+
+            responseObserver.onError(Status.INTERNAL.withCause(e)
+                    .withDescription("Failed in retrieving FTP storage with id " + request.getStorageId())
+                    .asRuntimeException());
+        }
+    }
+
+    @Override
+    public void createFTPStorage(FTPStorageCreateRequest request, StreamObserver<FTPStorage> responseObserver) {
+        try {
+            responseObserver.onNext(this.backend.createFTPStorage(request));
+            responseObserver.onCompleted();
+        } catch (Exception e) {
+            logger.error("Failed in creating the FTP storage", e);
+
+            responseObserver.onError(Status.INTERNAL.withCause(e)
+                    .withDescription("Failed in creating the FTP storage")
+                    .asRuntimeException());
+        }
+    }
+
+    @Override
+    public void updateFTPStorage(FTPStorageUpdateRequest request, StreamObserver<Empty> responseObserver) {
+        try {
+            this.backend.updateFTPStorage(request);
+            responseObserver.onCompleted();
+        } catch (Exception e) {
+            logger.error("Failed in updating the FTP storage {}", request.getStorageId(), e);
+
+            responseObserver.onError(Status.INTERNAL.withCause(e)
+                    .withDescription("Failed in updating the FTP storage")
+                    .asRuntimeException());
+        }
+    }
+
+    @Override
+    public void deleteFTPStorage(FTPStorageDeleteRequest request, StreamObserver<Empty> responseObserver) {
+        try {
+            boolean res = this.backend.deleteFTPStorage(request);
+            if (res) {
+                responseObserver.onCompleted();
+            } else {
+                logger.error("Failed to delete FTP Storage with id " + request.getStorageId());
+
+                responseObserver.onError(Status.INTERNAL
+                        .withDescription("Failed to delete FTP Storage with id " + request.getStorageId())
+                        .asRuntimeException());
+            }
+        } catch (Exception e) {
+            logger.error("Failed in deleting the FTP storage {}", request.getStorageId(), e);
+
+            responseObserver.onError(Status.INTERNAL.withCause(e)
+                    .withDescription("Failed in deleting the FTP storage")
+                    .asRuntimeException());
+        }
+    }
+
+    @Override
+    public void getFTPResource(FTPResourceGetRequest request, StreamObserver<FTPResource> responseObserver) {
+        try {
+            this.backend.getFTPResource(request).ifPresentOrElse(resource -> {
+                responseObserver.onNext(resource);
+                responseObserver.onCompleted();
+            }, () -> responseObserver.onError(Status.INTERNAL
+                    .withDescription("No FTP Resource with id " + request.getResourceId())
+                    .asRuntimeException()));
+        } catch (Exception e) {
+            logger.error("Failed in retrieving FTP resource with id {}", request.getResourceId(), e);
+
+            responseObserver.onError(Status.INTERNAL.withCause(e)
+                    .withDescription("Failed in retrieving FTP resource with id " + request.getResourceId())
+                    .asRuntimeException());
+        }
+    }
+
+    @Override
+    public void createFTPResource(FTPResourceCreateRequest request, StreamObserver<FTPResource> responseObserver) {
+        try {
+            responseObserver.onNext(this.backend.createFTPResource(request));
+            responseObserver.onCompleted();
+        } catch (Exception e) {
+            logger.error("Failed in creating the FTP resource", e);
+
+            responseObserver.onError(Status.INTERNAL.withCause(e)
+                    .withDescription("Failed in creating the FTP resource")
+                    .asRuntimeException());
+        }
+    }
+
+    @Override
+    public void updateFTPResource(FTPResourceUpdateRequest request, StreamObserver<Empty> responseObserver) {
+        try {
+            this.backend.updateFTPResource(request);
+            responseObserver.onCompleted();
+        } catch (Exception e) {
+            logger.error("Failed in updating the FTP resource {}", request.getResourceId(), e);
+
+            responseObserver.onError(Status.INTERNAL.withCause(e)
+                    .withDescription("Failed in updating the FTP resource")
+                    .asRuntimeException());
+        }
+    }
+
+    @Override
+    public void deleteFTPResource(FTPResourceDeleteRequest request, StreamObserver<Empty> responseObserver) {
+        try {
+            boolean res = this.backend.deleteFTPResource(request);
+            if (res) {
+                responseObserver.onCompleted();
+            } else {
+
+                responseObserver.onError(Status.INTERNAL
+                        .withDescription("Failed to delete FTP Resource with id " + request.getResourceId())
+                        .asRuntimeException());
+            }
+        } catch (Exception e) {
+            logger.error("Failed in deleting the scp resource {}", request.getResourceId(), e);
+
+            responseObserver.onError(Status.INTERNAL.withCause(e)
+                    .withDescription("Failed in deleting the FTP resource")
+                    .asRuntimeException());
+        }
+    }
 }