diff --git a/src/runtime/pkg/direct-volume/utils.go b/src/runtime/pkg/direct-volume/utils.go
new file mode 100644
index 0000000000..275b0508f8
--- /dev/null
+++ b/src/runtime/pkg/direct-volume/utils.go
@@ -0,0 +1,108 @@
+// Copyright (c) 2022 Databricks Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+package volume
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+const (
+	mountInfoFileName = "mountInfo.json"
+)
+
+var kataDirectVolumeRootPath = "/run/kata-containers/shared/direct-volumes"
+
+// MountInfo contains the information needed by Kata to consume a host block device and mount it as a filesystem inside the guest VM.
+type MountInfo struct {
+	// The type of the volume (ie. block)
+	VolumeType string `json:"volume-type"`
+	// The device backing the volume.
+	Device string `json:"device"`
+	// The filesystem type to be mounted on the volume.
+	FsType string `json:"fstype"`
+	// Additional metadata to pass to the agent regarding this volume.
+	Metadata map[string]string `json:"metadata,omitempty"`
+	// Additional mount options.
+	Options []string `json:"options,omitempty"`
+}
+
+// Add writes the mount info of a direct volume into a filesystem path known to Kata Container.
+func Add(volumePath string, mountInfo string) error {
+	volumeDir := filepath.Join(kataDirectVolumeRootPath, volumePath)
+	stat, err := os.Stat(volumeDir)
+	if err != nil && !errors.Is(err, os.ErrNotExist) {
+		return err
+	}
+	if stat != nil && !stat.IsDir() {
+		return fmt.Errorf("%s should be a directory", volumeDir)
+	}
+	if errors.Is(err, os.ErrNotExist) {
+		if err := os.MkdirAll(volumeDir, 0700); err != nil {
+			return err
+		}
+	}
+	var deserialized MountInfo
+	if err := json.Unmarshal([]byte(mountInfo), &deserialized); err != nil {
+		return err
+	}
+
+	return ioutil.WriteFile(filepath.Join(volumeDir, mountInfoFileName), []byte(mountInfo), 0600)
+}
+
+// Remove deletes the direct volume path including all the files inside it.
+func Remove(volumePath string) error {
+	// Find the base of the volume path to delete the whole volume path
+	base := strings.SplitN(volumePath, string(os.PathSeparator), 2)[0]
+	return os.RemoveAll(filepath.Join(kataDirectVolumeRootPath, base))
+}
+
+// VolumeMountInfo retrieves the mount info of a direct volume.
+func VolumeMountInfo(volumePath string) (*MountInfo, error) {
+	mountInfoFilePath := filepath.Join(kataDirectVolumeRootPath, volumePath, mountInfoFileName)
+	if _, err := os.Stat(mountInfoFilePath); err != nil {
+		return nil, err
+	}
+	buf, err := ioutil.ReadFile(mountInfoFilePath)
+	if err != nil {
+		return nil, err
+	}
+	var mountInfo MountInfo
+	if err := json.Unmarshal(buf, &mountInfo); err != nil {
+		return nil, err
+	}
+	return &mountInfo, nil
+}
+
+// RecordSandboxId associates a sandbox id with a direct volume.
+func RecordSandboxId(sandboxId string, volumePath string) error {
+	mountInfoFilePath := filepath.Join(kataDirectVolumeRootPath, volumePath, mountInfoFileName)
+	if _, err := os.Stat(mountInfoFilePath); err != nil {
+		return err
+	}
+
+	return ioutil.WriteFile(filepath.Join(kataDirectVolumeRootPath, volumePath, sandboxId), []byte(""), 0600)
+}
+
+func GetSandboxIdForVolume(volumePath string) (string, error) {
+	files, err := ioutil.ReadDir(filepath.Join(kataDirectVolumeRootPath, volumePath))
+	if err != nil {
+		return "", err
+	}
+	// Find the id of the first sandbox.
+	// We expect a direct-assigned volume is associated with only a sandbox at a time.
+	for _, file := range files {
+		if file.Name() != mountInfoFileName {
+			return file.Name(), nil
+		}
+	}
+	return "", fmt.Errorf("no sandbox found for %s", volumePath)
+}
diff --git a/src/runtime/pkg/direct-volume/utils_test.go b/src/runtime/pkg/direct-volume/utils_test.go
new file mode 100644
index 0000000000..6ca80dab81
--- /dev/null
+++ b/src/runtime/pkg/direct-volume/utils_test.go
@@ -0,0 +1,94 @@
+// Copyright (c) 2022 Databricks Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+package volume
+
+import (
+	"encoding/json"
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/kata-containers/kata-containers/src/runtime/pkg/uuid"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAdd(t *testing.T) {
+	var err error
+	kataDirectVolumeRootPath, err = os.MkdirTemp(os.TempDir(), "add-test")
+	assert.Nil(t, err)
+	defer os.RemoveAll(kataDirectVolumeRootPath)
+	var volumePath = "/a/b/c"
+	var basePath = "a"
+	actual := MountInfo{
+		VolumeType: "block",
+		Device:     "/dev/sda",
+		FsType:     "ext4",
+		Options:    []string{"journal_dev", "noload"},
+	}
+	buf, err := json.Marshal(actual)
+	assert.Nil(t, err)
+
+	// Add the mount info
+	assert.Nil(t, Add(volumePath, string(buf)))
+
+	// Validate the mount info
+	expected, err := VolumeMountInfo(volumePath)
+	assert.Nil(t, err)
+	assert.Equal(t, expected.Device, actual.Device)
+	assert.Equal(t, expected.FsType, actual.FsType)
+	assert.Equal(t, expected.Options, actual.Options)
+
+	// Remove the file
+	err = Remove(volumePath)
+	assert.Nil(t, err)
+	_, err = os.Stat(filepath.Join(kataDirectVolumeRootPath, basePath))
+	assert.True(t, errors.Is(err, os.ErrNotExist))
+
+	// Test invalid mount info json
+	assert.Error(t, Add(volumePath, "{invalid json}"))
+}
+
+func TestRecordSandboxId(t *testing.T) {
+	var err error
+	kataDirectVolumeRootPath, err = os.MkdirTemp(os.TempDir(), "recordSanboxId-test")
+	assert.Nil(t, err)
+	defer os.RemoveAll(kataDirectVolumeRootPath)
+
+	var volumePath = "/a/b/c"
+	mntInfo := MountInfo{
+		VolumeType: "block",
+		Device:     "/dev/sda",
+		FsType:     "ext4",
+		Options:    []string{"journal_dev", "noload"},
+	}
+	buf, err := json.Marshal(mntInfo)
+	assert.Nil(t, err)
+
+	// Add the mount info
+	assert.Nil(t, Add(volumePath, string(buf)))
+
+	sandboxId := uuid.Generate().String()
+	err = RecordSandboxId(sandboxId, volumePath)
+	assert.Nil(t, err)
+
+	id, err := GetSandboxIdForVolume(volumePath)
+	assert.Nil(t, err)
+	assert.Equal(t, sandboxId, id)
+}
+
+func TestRecordSandboxIdNoMountInfoFile(t *testing.T) {
+	var err error
+	kataDirectVolumeRootPath, err = os.MkdirTemp(os.TempDir(), "recordSanboxId-test")
+	assert.Nil(t, err)
+	defer os.RemoveAll(kataDirectVolumeRootPath)
+
+	var volumePath = "/a/b/c"
+	sandboxId := uuid.Generate().String()
+	err = RecordSandboxId(sandboxId, volumePath)
+	assert.Error(t, err)
+	assert.True(t, errors.Is(err, os.ErrNotExist))
+}