diff --git a/common/djangoapps/student/api.py b/common/djangoapps/student/api.py
index 56b0683f9f..2bf42f4828 100644
--- a/common/djangoapps/student/api.py
+++ b/common/djangoapps/student/api.py
@@ -8,6 +8,7 @@ import logging
 
 from django.contrib.auth import get_user_model
 from django.conf import settings
+from opaque_keys.edx.keys import CourseKey
 
 from common.djangoapps.student.models import CourseEnrollment
 from common.djangoapps.student.models_api import create_manual_enrollment_audit as _create_manual_enrollment_audit
@@ -23,7 +24,12 @@ from common.djangoapps.student.models_api import (
     ALLOWEDTOENROLL_TO_UNENROLLED as _ALLOWEDTOENROLL_TO_UNENROLLED,
     DEFAULT_TRANSITION_STATE as _DEFAULT_TRANSITION_STATE,
 )
-from common.djangoapps.student.roles import REGISTERED_ACCESS_ROLES as _REGISTERED_ACCESS_ROLES
+from common.djangoapps.student.roles import (
+    CourseInstructorRole,
+    CourseStaffRole,
+    GlobalStaff,
+    REGISTERED_ACCESS_ROLES as _REGISTERED_ACCESS_ROLES,
+)
 from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
 
 
@@ -114,3 +120,19 @@ def is_user_enrolled_in_course(student, course_key):
     """
     log.info(f"Checking if {student.id} is enrolled in course {course_key}")
     return CourseEnrollment.is_enrolled(student, course_key)
+
+
+def is_user_staff_or_instructor_in_course(user, course_key):
+    """
+    Determines if a user is an Instructor or part of the given course's course staff.
+
+    Also returns true for GlobalStaff.
+    """
+    if not isinstance(course_key, CourseKey):
+        course_key = CourseKey.from_string(course_key)
+
+    return (
+        GlobalStaff().has_user(user) or
+        CourseStaffRole(course_key).has_user(user) or
+        CourseInstructorRole(course_key).has_user(user)
+    )
diff --git a/common/djangoapps/student/tests/test_api.py b/common/djangoapps/student/tests/test_api.py
index 0b313e24eb..ad462830a1 100644
--- a/common/djangoapps/student/tests/test_api.py
+++ b/common/djangoapps/student/tests/test_api.py
@@ -1,12 +1,17 @@
 """
 Test Student api.py
 """
-
 from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory
 
-from common.djangoapps.student.api import is_user_enrolled_in_course
-from common.djangoapps.student.tests.factories import CourseEnrollmentFactory, UserFactory
+from common.djangoapps.student.api import is_user_enrolled_in_course, is_user_staff_or_instructor_in_course
+from common.djangoapps.student.tests.factories import (
+    CourseEnrollmentFactory,
+    GlobalStaffFactory,
+    InstructorFactory,
+    StaffFactory,
+    UserFactory,
+)
 
 
 class TestStudentApi(SharedModuleStoreTestCase):
@@ -55,3 +60,22 @@ class TestStudentApi(SharedModuleStoreTestCase):
         """
         result = is_user_enrolled_in_course(self.user, self.course_run_key)
         assert not result
+
+    def test_is_user_staff_or_instructor(self):
+        """
+        Verify the correct value is returned for users with different access levels.
+        """
+        course_id_string = str(self.course.id)
+        global_staff_user = GlobalStaffFactory.create()
+        staff_user = StaffFactory.create(course_key=self.course_run_key)
+        instructor = InstructorFactory.create(course_key=self.course_run_key)
+
+        different_course = CourseFactory.create()
+        instructor_different_course = InstructorFactory.create(course_key=different_course.id)
+
+        assert is_user_staff_or_instructor_in_course(instructor, course_id_string)
+        assert is_user_staff_or_instructor_in_course(global_staff_user, self.course_run_key)
+        assert is_user_staff_or_instructor_in_course(staff_user, self.course_run_key)
+        assert is_user_staff_or_instructor_in_course(instructor, self.course_run_key)
+        assert not is_user_staff_or_instructor_in_course(self.user, self.course_run_key)
+        assert not is_user_staff_or_instructor_in_course(instructor_different_course, self.course_run_key)
diff --git a/lms/djangoapps/bulk_email/api.py b/lms/djangoapps/bulk_email/api.py
index f89a27065b..c2f1666f70 100644
--- a/lms/djangoapps/bulk_email/api.py
+++ b/lms/djangoapps/bulk_email/api.py
@@ -2,8 +2,6 @@
 """
 Python APIs exposed by the bulk_email app to other in-process apps.
 """
-
-# Public Bulk Email Functions
 import logging
 
 from django.conf import settings
@@ -12,7 +10,6 @@ from django.urls import reverse
 from lms.djangoapps.bulk_email.models import CourseEmail
 from lms.djangoapps.bulk_email.models_api import (
     is_bulk_email_disabled_for_course,
-    is_bulk_email_enabled_for_course,
     is_bulk_email_feature_enabled,
     is_user_opted_out_for_course
 )
@@ -41,7 +38,6 @@ def get_emails_enabled(user, course_id):
 
 def get_unsubscribed_link(username, course_id):
     """
-
     :param username: username
     :param course_id:
     :return: AES encrypted token based on the user email
@@ -87,3 +83,21 @@ def create_course_email(course_id, sender, targets, subject, html_message, text_
     except ValueError as err:
         log.exception(f"Cannot create course email for {course_id} requested by user {sender} for targets {targets}")
         raise ValueError from err
+
+
+def get_course_email(email_id):
+    """
+    Utility function for retrieving a CourseEmail instance from a given CourseEmail id.
+
+    Args:
+        email_id (int): The ID of the CourseEmail instance you want to retrieve.
+
+    Returns:
+        CourseEmail: The CourseEmail instance, if it exists.
+    """
+    try:
+        return CourseEmail.objects.get(id=email_id)
+    except CourseEmail.DoesNotExist:
+        log.exception(f"CourseEmail instance with id '{email_id}' could not be found")
+
+    return None
diff --git a/lms/djangoapps/bulk_email/tests/test_api.py b/lms/djangoapps/bulk_email/tests/test_api.py
index f6c0804724..cc7fdbb431 100644
--- a/lms/djangoapps/bulk_email/tests/test_api.py
+++ b/lms/djangoapps/bulk_email/tests/test_api.py
@@ -7,10 +7,12 @@ from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory
 
 from common.djangoapps.student.tests.factories import InstructorFactory
-from lms.djangoapps.bulk_email.api import create_course_email
+from lms.djangoapps.bulk_email.api import create_course_email, get_course_email
 from lms.djangoapps.bulk_email.data import BulkEmailTargetChoices
 from openedx.core.lib.html_to_text import html_to_text
 
+LOG_PATH = "lms.djangoapps.bulk_email.api"
+
 
 class CreateCourseEmailTests(ModuleStoreTestCase):
     """
@@ -77,10 +79,10 @@ class CreateCourseEmailTests(ModuleStoreTestCase):
         """
         targets = ["humpty dumpty"]
 
-        expected_messages = [
+        expected_message = (
             f"Cannot create course email for {self.course.id} requested by user {self.instructor} for targets "
-            f"{targets}",
-        ]
+            f"{targets}"
+        )
 
         with self.assertRaises(ValueError):
             with LogCapture() as log:
@@ -92,5 +94,33 @@ class CreateCourseEmailTests(ModuleStoreTestCase):
                     self.html_message
                 )
 
-        for index, message in enumerate(expected_messages):
-            assert message in log.records[index].getMessage()
+        log.check_present(
+            (LOG_PATH, "ERROR", expected_message),
+        )
+
+    def test_get_course_email(self):
+        """
+        A test to verify the happy path behavior of the `get_course_email` utility function and the presence of an
+        expected log message when an email instance can't be found for a given id.
+        """
+        course_email = create_course_email(
+            self.course.id,
+            self.instructor,
+            self.target,
+            self.subject,
+            self.html_message,
+        )
+
+        email_instance = get_course_email(course_email.id)
+        assert email_instance.id == course_email.id
+
+        email_id_dne = 3463435
+        expected_message = (
+            f"CourseEmail instance with id '{email_id_dne}' could not be found"
+        )
+        with LogCapture() as log:
+            get_course_email(email_id_dne)
+
+        log.check_present(
+            (LOG_PATH, "ERROR", expected_message),
+        )
diff --git a/lms/djangoapps/instructor/tests/test_email.py b/lms/djangoapps/instructor/tests/test_email.py
index 18cffa4740..a84c728c30 100644
--- a/lms/djangoapps/instructor/tests/test_email.py
+++ b/lms/djangoapps/instructor/tests/test_email.py
@@ -10,7 +10,7 @@ from django.urls import reverse
 from opaque_keys.edx.keys import CourseKey
 
 from common.djangoapps.student.tests.factories import AdminFactory
-from lms.djangoapps.bulk_email.api import is_bulk_email_enabled_for_course, is_bulk_email_feature_enabled
+from lms.djangoapps.bulk_email.models_api import is_bulk_email_enabled_for_course, is_bulk_email_feature_enabled
 
 from lms.djangoapps.bulk_email.models import BulkEmailFlag, CourseAuthorization, DisabledCourse
 from xmodule.modulestore.tests.django_utils import TEST_DATA_MIXED_MODULESTORE, SharedModuleStoreTestCase  # lint-amnesty, pylint: disable=wrong-import-order
diff --git a/lms/djangoapps/instructor_task/migrations/0005_alter_instructortaskschedule_task.py b/lms/djangoapps/instructor_task/migrations/0005_alter_instructortaskschedule_task.py
new file mode 100644
index 0000000000..d3f7f426e1
--- /dev/null
+++ b/lms/djangoapps/instructor_task/migrations/0005_alter_instructortaskschedule_task.py
@@ -0,0 +1,19 @@
+# Generated by Django 3.2.13 on 2022-04-26 13:32
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('instructor_task', '0004_historicalinstructortaskschedule_instructortaskschedule'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='instructortaskschedule',
+            name='task',
+            field=models.OneToOneField(on_delete=django.db.models.deletion.DO_NOTHING, to='instructor_task.instructortask'),
+        ),
+    ]
diff --git a/lms/djangoapps/instructor_task/models.py b/lms/djangoapps/instructor_task/models.py
index 92268c3090..3655bccf17 100644
--- a/lms/djangoapps/instructor_task/models.py
+++ b/lms/djangoapps/instructor_task/models.py
@@ -203,7 +203,7 @@ class InstructorTaskSchedule(TimeStampedModel):
     class Meta:
         app_label = "instructor_task"
 
-    task = models.OneToOneField(InstructorTask, on_delete=models.CASCADE)
+    task = models.OneToOneField(InstructorTask, on_delete=models.DO_NOTHING)
     task_args = models.TextField(null=False, blank=False)
     task_due = models.DateTimeField(null=False)
 
diff --git a/lms/djangoapps/instructor_task/rest_api/__init__.py b/lms/djangoapps/instructor_task/rest_api/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lms/djangoapps/instructor_task/rest_api/urls.py b/lms/djangoapps/instructor_task/rest_api/urls.py
new file mode 100644
index 0000000000..cfffc382b1
--- /dev/null
+++ b/lms/djangoapps/instructor_task/rest_api/urls.py
@@ -0,0 +1,14 @@
+"""
+Instructor Task Django app root REST API URLs.
+"""
+from django.conf.urls import include
+from django.urls import path
+
+from lms.djangoapps.instructor_task.rest_api.v1 import urls as v1_urls
+
+
+app_name = "lms.djangoapps.instructor_task"
+
+urlpatterns = [
+    path("v1/", include(v1_urls))
+]
diff --git a/lms/djangoapps/instructor_task/rest_api/v1/__init__.py b/lms/djangoapps/instructor_task/rest_api/v1/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lms/djangoapps/instructor_task/rest_api/v1/permissions.py b/lms/djangoapps/instructor_task/rest_api/v1/permissions.py
new file mode 100644
index 0000000000..7685e5624c
--- /dev/null
+++ b/lms/djangoapps/instructor_task/rest_api/v1/permissions.py
@@ -0,0 +1,23 @@
+"""
+Instructor Task Django app REST API permission classes.
+"""
+from rest_framework import permissions
+
+from common.djangoapps.student.api import is_user_staff_or_instructor_in_course
+
+
+class CanViewOrDeleteScheduledBulkCourseEmails(permissions.BasePermission):
+    """
+    Permission class that ensures a user is allowed to interact with the bulk course messages in a given course-run.
+    """
+    def has_permission(self, request, view):
+        """
+        Only course-staff/instructors or staff users should be able to modify the bulk course email messages or
+        schedules.
+        """
+        user = request.user
+        if user and user.is_authenticated:
+            course_id = view.kwargs['course_id']
+            return is_user_staff_or_instructor_in_course(user, course_id)
+
+        return False
diff --git a/lms/djangoapps/instructor_task/rest_api/v1/serializers.py b/lms/djangoapps/instructor_task/rest_api/v1/serializers.py
new file mode 100644
index 0000000000..5bc3335703
--- /dev/null
+++ b/lms/djangoapps/instructor_task/rest_api/v1/serializers.py
@@ -0,0 +1,79 @@
+"""
+Instructor Task Django app REST API serializers.
+"""
+import json
+
+from django.apps import apps
+from rest_framework import serializers
+
+from lms.djangoapps.bulk_email.api import get_course_email
+from lms.djangoapps.instructor_task.models import InstructorTaskSchedule
+
+
+class SenderField(serializers.RelatedField):
+    """
+    Serializer field that converts the user object id to a human readable name (username)
+    """
+    def to_representation(self, value):
+        return value.username
+
+
+class TargetListField(serializers.RelatedField):
+    """
+    Serializer field that converts the email target values (recipient groups) of a message from an int to a human
+    readable name.
+    """
+    def to_representation(self, value):
+        return value.short_display()
+
+
+class CourseEmailSerializer(serializers.ModelSerializer):
+    """
+    Serializer for the course email instance belonging to each scheduled task.
+    """
+    targets = TargetListField(many=True, read_only=True)
+    sender = SenderField(many=False, read_only=True)
+
+    class Meta:
+        # use the bulk_email app's CourseEmail model without adding the direct import of the model
+        course_email_model = apps.get_model('bulk_email.CourseEmail')
+        model = course_email_model
+        fields = (
+            "id",
+            "subject",
+            "html_message",
+            "text_message",
+            "course_id",
+            "to_option",
+            "sender",
+            "targets"
+        )
+
+
+class ScheduledBulkEmailSerializer(serializers.ModelSerializer):
+    """
+    Serializer for scheduled bulk email instructor tasks.
+    """
+    course_email = serializers.SerializerMethodField()
+
+    class Meta:
+        model = InstructorTaskSchedule
+        fields = (
+            "id",
+            "course_email",
+            "task",
+            "task_due",
+        )
+
+    def get_course_email(self, obj):
+        """
+        This function is responsible for retrieving and including course email instance information associated with
+        each individual scheduled task. Uses the task related to the schedule to extract the email id of the scheduled
+        message. From here we can lookup the individual message and attach it with the other API results.
+        """
+        # extract the id of the course email instance for this task
+        task_input = json.loads(obj.task.task_input)
+        email_id = task_input["email_id"]
+        # retrieve the course_email instance
+        course_email = get_course_email(email_id)
+        return CourseEmailSerializer(course_email).data
diff --git a/lms/djangoapps/instructor_task/rest_api/v1/tests/__init__.py b/lms/djangoapps/instructor_task/rest_api/v1/tests/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/lms/djangoapps/instructor_task/rest_api/v1/tests/test_views.py b/lms/djangoapps/instructor_task/rest_api/v1/tests/test_views.py
new file mode 100644
index 0000000000..b2103b7733
--- /dev/null
+++ b/lms/djangoapps/instructor_task/rest_api/v1/tests/test_views.py
@@ -0,0 +1,191 @@
+"""
+Tests for the instructor_task app's REST API v1 views.
+"""
+import json
+from uuid import uuid4
+
+from celery.states import REVOKED
+import ddt
+from django.contrib.auth import get_user_model
+from rest_framework.test import APITestCase
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
+from xmodule.modulestore.tests.factories import CourseFactory
+
+from common.djangoapps.student.tests.factories import (
+    GlobalStaffFactory,
+    InstructorFactory,
+    StaffFactory,
+    UserFactory,
+)
+from lms.djangoapps.bulk_email.api import create_course_email
+from lms.djangoapps.bulk_email.data import BulkEmailTargetChoices
+from lms.djangoapps.instructor_task.data import InstructorTaskTypes
+from lms.djangoapps.instructor_task.models import InstructorTask, PROGRESS, SCHEDULED
+from lms.djangoapps.instructor_task.tests.factories import InstructorTaskFactory, InstructorTaskScheduleFactory
+
+User = get_user_model()
+
+
+@ddt.ddt
+class TestScheduledBulkEmailAPIViews(APITestCase, ModuleStoreTestCase):
+    """
+    Tests for the ListScheduledBulkEmailInstructorTasks and DeleteScheduledBulkEmailInstructorTask views.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.course1 = CourseFactory()
+        self.course2 = CourseFactory()
+        self.global_staff = GlobalStaffFactory(username="globalstaff")
+        self.instructor_course1 = InstructorFactory.create(
+            course_key=self.course1.id,
+            username="instructor_course1")
+        self.instructor_course2 = InstructorFactory.create(
+            course_key=self.course2.id,
+            username="instructor_course2")
+        self.staff = StaffFactory.create(
+            course_key=self.course1.id,
+            username="staff")
+        self.user = UserFactory(username="user")
+
+    def tearDown(self):
+        super().tearDown()
+        self.client.logout()
+
+    def _build_api_url(self, course_id):
+        """
+        Utility function to build the URL to retrieve scheduled bulk email tasks from a given course-id.
+        """
+        return f"/api/instructor_task/v1/schedules/{course_id}/bulk_email/"
+
+    def _create_course_email(self, course_id, author, targets, subject, message):
+        """
+        Utility function to create CourseEmail objects for the test suite.
+        """
+        return create_course_email(course_id, author, targets, subject, message)
+
+    def _create_scheduled_course_emails_for_course(self, course_id, email_author, task_status, num_emails):
+        """
+        Utility function to create (bulk course email) scheduled instructor tasks for the test suite.
+        """
+        for i in range(1, (num_emails + 1)):
+            # create the course email instance
+            email = self._create_course_email(
+                course_id,
+                email_author,
+                [BulkEmailTargetChoices.SEND_TO_MYSELF],
+                f"Test Subject{i}",
+                "<p>Test message.</p>"
+            )
+            # associate the course_email instance with the task
+            task_input = {'email_id': email.id}
+            task_id = str(uuid4())
+            task = InstructorTaskFactory.create(
+                course_id=course_id,
+                requester=email_author,
+                task_id=task_id,
+                task_input=json.dumps(task_input),
+                task_key='dummy value',
+                task_state=task_status,
+                task_type=InstructorTaskTypes.BULK_COURSE_EMAIL
+            )
+            # associate the task with an instructor task schedule
+            task_args = {
+                "request_info": {
+                    "username": self.instructor_course1.username,
+                    "user_id": self.instructor_course1.id,
+                    "ip": "192.168.1.100",
+                    "agent": "Mozilla",
+                    "host": "localhost:18000"
+                },
+                "task_id": task.task_id
+            }
+            InstructorTaskScheduleFactory.create(
+                task=task,
+                task_args=json.dumps(task_args),
+            )
+
+    @ddt.data(
+        ("globalstaff", 200),
+        ("instructor_course1", 200),
+        ("instructor_course2", 403),
+        ("staff", 200),
+        ("user", 403)
+    )
+    @ddt.unpack
+    def test_list_tasks_basic_permissions(self, username, expected_status):
+        """
+        Test case that verifies the permissions of the ListScheduledBulkEmailInstructorTasks view. A user without staff,
+        instructor, or GlobalStaff access should not be able to retrieve the bulk email schedules for a course.
+        """
+        self.client.login(username=username, password="test")
+        response = self.client.get(self._build_api_url(self.course1.id))
+        assert response.status_code == expected_status
+
+    def test_list_tasks(self):
+        """
+        Test case that verifies the functionality of the ListScheduledBulkEmailInstructorTasks view.
+
+        This test verifies that the results returned are for scheduled tasks awaiting execution and also confirms that
+        the email data associated with the results belong to the correct task and schedule.
+        """
+        self._create_scheduled_course_emails_for_course(self.course1.id, self.instructor_course1, SCHEDULED, 3)
+        # add a "In Progress" task which shouldn't be returned in our results
+        self._create_scheduled_course_emails_for_course(self.course1.id, self.instructor_course1, PROGRESS, 1)
+
+        self.client.login(username=self.instructor_course1.username, password="test")
+        response = self.client.get(self._build_api_url(self.course1.id))
+        results = response.data.get("results")
+
+        assert response.status_code == 200
+        assert len(results) == 3
+        # Verify the serializer is returning the correct data
+        for result in results:
+            # Get the full task associated with the individual response instance, then verify that the email referenced
+            # in the response instance belongs to the apporpriate task.
+            email_data = result.get("course_email")
+            task = InstructorTask.objects.get(id=result.get("task"))
+            task_input = json.loads(task.task_input)
+            task_email_id = task_input.get("email_id")
+            assert email_data.get("id") == task_email_id
+
+    def test_delete_schedules(self):
+        """
+        Test case that verifies the functionality of the DeleteScheduledBulkEmailInstructorTask view.
+
+        This test verifies that, when a task schedule is cancelled, the data is in the correct state and that the task
+        is no longer returned with GET requests.
+        """
+        self._create_scheduled_course_emails_for_course(self.course1.id, self.instructor_course1, SCHEDULED, 3)
+        self.client.login(username=self.instructor_course1.username, password="test")
+        response = self.client.get(self._build_api_url(self.course1.id))
+        results = response.data.get("results")
+
+        assert len(results) == 3
+
+        # cancel/delete the first scheduled bulk email task in the list
+        schedule_id = results[0].get("id")
+        task_id = results[0].get("task")
+        delete_response = self.client.delete(f"{self._build_api_url(self.course1.id)}{schedule_id}")
+        assert delete_response.status_code == 204
+
+        # get the list of scheduled tasks again and verify the one we deleted doesn't appear in the results
+        response = self.client.get(self._build_api_url(self.course1.id))
+        results = response.data.get("results")
+        assert len(results) == 2
+        for result in results:
+            assert not result.get("id") == schedule_id
+
+        # verify the task status is REVOKED for the scheduled we cancelled
+        task = InstructorTask.objects.get(id=task_id)
+        assert task.task_state == REVOKED
+
+    def test_delete_schedules_schedule_does_not_exist(self):
+        """
+        Test case that verifies the functionality of the DeleteScheduledBulkEmailInstructorTask view.
+
+        This test verifies the response received when we try to delete a task schedule that does not exist.
+        """
+        self.client.login(username=self.instructor_course1.username, password="test")
+        response = self.client.delete(f"{self._build_api_url(self.course1.id)}123456789")
+        assert response.status_code == 404
diff --git a/lms/djangoapps/instructor_task/rest_api/v1/urls.py b/lms/djangoapps/instructor_task/rest_api/v1/urls.py
new file mode 100644
index 0000000000..a25915e19a
--- /dev/null
+++ b/lms/djangoapps/instructor_task/rest_api/v1/urls.py
@@ -0,0 +1,24 @@
+"""
+Instructor Task Django app REST API URLs.
+"""
+from django.conf import settings
+from django.urls import re_path
+
+from lms.djangoapps.instructor_task.rest_api.v1.views import (
+    ListScheduledBulkEmailInstructorTasks,
+    DeleteScheduledBulkEmailInstructorTask
+)
+
+
+urlpatterns = [
+    re_path(
+        fr"schedules/{settings.COURSE_ID_PATTERN}/bulk_email/$",
+        ListScheduledBulkEmailInstructorTasks.as_view(),
+        name="get-scheduled-bulk-email-messages"
+    ),
+    re_path(
+        fr"schedules/{settings.COURSE_ID_PATTERN}/bulk_email/(?P<schedule_id>[0-9]+)$",
+        DeleteScheduledBulkEmailInstructorTask.as_view(),
+        name="delete-scheduled-bulk-email-messages"
+    )
+]
diff --git a/lms/djangoapps/instructor_task/rest_api/v1/views.py b/lms/djangoapps/instructor_task/rest_api/v1/views.py
new file mode 100644
index 0000000000..b3c019d647
--- /dev/null
+++ b/lms/djangoapps/instructor_task/rest_api/v1/views.py
@@ -0,0 +1,88 @@
+"""
+Instructor Task Django app REST API views.
+"""
+import logging
+
+from celery.states import REVOKED
+from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
+from rest_framework.authentication import SessionAuthentication
+from rest_framework.response import Response
+from rest_framework import generics, status
+
+from lms.djangoapps.instructor_task.data import InstructorTaskTypes
+from lms.djangoapps.instructor_task.models import InstructorTaskSchedule, SCHEDULED
+from lms.djangoapps.instructor_task.rest_api.v1.serializers import ScheduledBulkEmailSerializer
+from lms.djangoapps.instructor_task.rest_api.v1.permissions import CanViewOrDeleteScheduledBulkCourseEmails
+
+log = logging.getLogger(__name__)
+
+
+class ListScheduledBulkEmailInstructorTasks(generics.ListAPIView):
+    """
+    Read only view to list all scheduled bulk email messages for a course-run.
+
+    Path: GET `api/instructor_task/v1/schedules/{course_id}/bulk_email`
+
+    Returns:
+        * 200: OK - Contains a list of all scheduled bulk email instructor tasks that haven't been executed yet. This
+               data also includes information about the and course email instance associated with each task.
+        * 403: User does not have the required role to view this data.
+    """
+    authentication_classes = (
+        JwtAuthentication,
+        SessionAuthentication,
+    )
+    permission_classes = (
+        CanViewOrDeleteScheduledBulkCourseEmails,
+    )
+    serializer_class = ScheduledBulkEmailSerializer
+
+    def get_queryset(self):
+        """
+        Filters the results so that only scheduled bulk email tasks for the specific course-run are returned.
+        """
+        course_id = self.kwargs["course_id"]
+        return (
+            InstructorTaskSchedule.objects
+            .filter(task__course_id=course_id)
+            .filter(task__task_state=SCHEDULED)
+            .filter(task__task_type=InstructorTaskTypes.BULK_COURSE_EMAIL)
+        )
+
+
+class DeleteScheduledBulkEmailInstructorTask(generics.DestroyAPIView):
+    """
+    A view that deletes an instructor task schedule instance and revokes the associated instructor task.
+
+    Path: DELETE `api/instructor_task/v1/schedules/{course_id}/bulk_email/{task_schedule_id}`
+
+    Returns:
+        * 204: No Content - Deleting the schedule was successful.
+        * 404: Requested schedule object could not be found and thus could not be deleted.
+    """
+    authentication_classes = (
+        JwtAuthentication,
+        SessionAuthentication,
+    )
+    permission_classes = (
+        CanViewOrDeleteScheduledBulkCourseEmails,
+    )
+
+    def destroy(self, request, *args, **kwargs):
+        course_id = kwargs["course_id"]
+        schedule_id = kwargs["schedule_id"]
+
+        log.info(f"Cancelling instructor task schedule with id '{schedule_id}' in course '{course_id}'")
+        try:
+            schedule = InstructorTaskSchedule.objects.get(id=schedule_id)
+        except InstructorTaskSchedule.DoesNotExist:
+            return Response(status=status.HTTP_404_NOT_FOUND)
+
+        # update the task's status to REVOKED and then delete the task schedule instance
+        task = schedule.task
+        log.info(f"Revoking instructor task with id '{task.id}' for course '{task.course_id}'")
+        task.task_state = REVOKED
+        task.save()
+        schedule.delete()
+
+        return Response(status=status.HTTP_204_NO_CONTENT)
diff --git a/lms/djangoapps/instructor_task/tests/test_api.py b/lms/djangoapps/instructor_task/tests/test_api.py
index 9b8f02699a..082227c968 100644
--- a/lms/djangoapps/instructor_task/tests/test_api.py
+++ b/lms/djangoapps/instructor_task/tests/test_api.py
@@ -16,11 +16,13 @@ from xmodule.modulestore.exceptions import ItemNotFoundError
 
 from common.djangoapps.student.tests.factories import UserFactory
 from common.test.utils import normalize_repr
-from lms.djangoapps.bulk_email.models import SEND_TO_LEARNERS, SEND_TO_MYSELF, SEND_TO_STAFF, CourseEmail
+from lms.djangoapps.bulk_email.api import create_course_email
+from lms.djangoapps.bulk_email.data import BulkEmailTargetChoices
 from lms.djangoapps.certificates.data import CertificateStatuses
 from lms.djangoapps.certificates.models import CertificateGenerationHistory
 from lms.djangoapps.instructor_task.api import (
     SpecificStudentIdMissingError,
+    generate_anonymous_ids,
     generate_certificates_for_students,
     get_instructor_task_history,
     get_running_instructor_tasks,
@@ -42,7 +44,6 @@ from lms.djangoapps.instructor_task.api import (
     submit_rescore_problem_for_student,
     submit_reset_problem_attempts_for_all_students,
     submit_reset_problem_attempts_in_entrance_exam,
-    generate_anonymous_ids
 )
 from lms.djangoapps.instructor_task.api_helper import AlreadyRunningError, QueueConnectionError
 from lms.djangoapps.instructor_task.models import PROGRESS, SCHEDULED, InstructorTask
@@ -217,12 +218,15 @@ class InstructorTaskCourseSubmitTest(TestReportMixin, InstructorTaskCourseTestCa
 
     def _define_course_email(self):
         """Create CourseEmail object for testing."""
-        # TODO: convert to use bulk_email app's `create_course_email` API function and remove direct import and use of
-        # bulk_email model
-        course_email = CourseEmail.create(
+        email_recipient_groups = [
+            BulkEmailTargetChoices.SEND_TO_MYSELF,
+            BulkEmailTargetChoices.SEND_TO_STAFF,
+            BulkEmailTargetChoices.SEND_TO_LEARNERS
+        ]
+        course_email = create_course_email(
             self.course.id,
             self.instructor,
-            [SEND_TO_MYSELF, SEND_TO_STAFF, SEND_TO_LEARNERS],
+            email_recipient_groups,
             "Test Subject",
             "<p>This is a test message</p>"
         )
diff --git a/lms/urls.py b/lms/urls.py
index 21fb97fba4..80f77a6440 100644
--- a/lms/urls.py
+++ b/lms/urls.py
@@ -1024,3 +1024,8 @@ if settings.ENABLE_SAVE_FOR_LATER:
 urlpatterns += [
     path('api/ora_staff_grader/', include('lms.djangoapps.ora_staff_grader.urls', 'ora-staff-grader')),
 ]
+
+# Scheduled Bulk Email (Instructor Task) URLs
+urlpatterns += [
+    path('api/instructor_task/', include('lms.djangoapps.instructor_task.rest_api.urls')),
+]