diff --git a/zerver/actions/scheduled_messages.py b/zerver/actions/scheduled_messages.py
index 2d6352d3d6..abc1e592db 100644
--- a/zerver/actions/scheduled_messages.py
+++ b/zerver/actions/scheduled_messages.py
@@ -39,10 +39,12 @@ def check_schedule_message(
     if scheduled_message_id is not None:
         return edit_scheduled_message(scheduled_message_id, send_request, sender)
 
-    return do_schedule_messages([send_request])[0]
+    return do_schedule_messages([send_request], sender)[0]
 
 
-def do_schedule_messages(send_message_requests: Sequence[SendMessageRequest]) -> List[int]:
+def do_schedule_messages(
+    send_message_requests: Sequence[SendMessageRequest], sender: UserProfile
+) -> List[int]:
     scheduled_messages: List[ScheduledMessage] = []
 
     for send_request in send_message_requests:
@@ -66,6 +68,14 @@ def do_schedule_messages(send_message_requests: Sequence[SendMessageRequest]) ->
         scheduled_messages.append(scheduled_message)
 
     ScheduledMessage.objects.bulk_create(scheduled_messages)
+    event = {
+        "type": "scheduled_messages",
+        "op": "add",
+        "scheduled_messages": [
+            scheduled_message.to_dict() for scheduled_message in scheduled_messages
+        ],
+    }
+    send_event(sender.realm, event, [sender.id])
     return [scheduled_message.id for scheduled_message in scheduled_messages]
 
 
@@ -93,6 +103,13 @@ def edit_scheduled_message(
         assert send_request.deliver_at is not None
         scheduled_message_object.scheduled_timestamp = send_request.deliver_at
         scheduled_message_object.save()
+
+    event = {
+        "type": "scheduled_messages",
+        "op": "update",
+        "scheduled_message": scheduled_message_object.to_dict(),
+    }
+    send_event(sender.realm, event, [sender.id])
     return scheduled_message_id
 
 
@@ -102,7 +119,7 @@ def delete_scheduled_message(user_profile: UserProfile, scheduled_message_id: in
     scheduled_message_object.delete()
 
     event = {
-        "type": "scheduled_message",
+        "type": "scheduled_messages",
         "op": "remove",
         "scheduled_message_id": scheduled_message_id,
     }
diff --git a/zerver/lib/events.py b/zerver/lib/events.py
index abb7e63447..8d4ccfa5ef 100644
--- a/zerver/lib/events.py
+++ b/zerver/lib/events.py
@@ -40,6 +40,7 @@ from zerver.lib.presence import get_presence_for_user, get_presences_for_realm
 from zerver.lib.push_notifications import push_notifications_enabled
 from zerver.lib.realm_icon import realm_icon_url
 from zerver.lib.realm_logo import get_realm_logo_source, get_realm_logo_url
+from zerver.lib.scheduled_messages import get_undelivered_scheduled_messages
 from zerver.lib.soft_deactivation import reactivate_user_if_soft_deactivated
 from zerver.lib.sounds import get_available_notification_sounds
 from zerver.lib.stream_subscription import handle_stream_notifications_compatibility
@@ -204,6 +205,11 @@ def fetch_initial_state_data(
             user_draft_dicts = [draft.to_dict() for draft in user_draft_objects]
             state["drafts"] = user_draft_dicts
 
+    if want("scheduled_messages"):
+        state["scheduled_messages"] = (
+            [] if user_profile is None else get_undelivered_scheduled_messages(user_profile)
+        )
+
     if want("muted_topics") and (
         # Suppress muted_topics data for clients that explicitly
         # support user_topic. This allows clients to request both the
@@ -768,6 +774,41 @@ def apply_event(
             assert state_draft_idx is not None
             _draft_update_action(state_draft_idx)
 
+    elif event["type"] == "scheduled_messages":
+        if event["op"] == "add":
+            # Since bulk addition of scheduled messages will not be used by a normal user.
+            assert len(event["scheduled_messages"]) == 1
+
+            state["scheduled_messages"].append(event["scheduled_messages"][0])
+            # Sort in ascending order of scheduled_delivery_timestamp.
+            state["scheduled_messages"].sort(
+                key=lambda scheduled_message: scheduled_message["scheduled_delivery_timestamp"]
+            )
+
+        if event["op"] == "update":
+            for idx, scheduled_message in enumerate(state["scheduled_messages"]):
+                if (
+                    scheduled_message["scheduled_message_id"]
+                    == event["scheduled_message"]["scheduled_message_id"]
+                ):
+                    state["scheduled_messages"][idx] = event["scheduled_message"]
+                    # If scheduled_delivery_timestamp was changed, we need to sort it again.
+                    if (
+                        scheduled_message["scheduled_delivery_timestamp"]
+                        != event["scheduled_message"]["scheduled_delivery_timestamp"]
+                    ):
+                        state["scheduled_messages"].sort(
+                            key=lambda scheduled_message: scheduled_message[
+                                "scheduled_delivery_timestamp"
+                            ]
+                        )
+                    break
+
+        if event["op"] == "remove":
+            for idx, scheduled_message in enumerate(state["scheduled_messages"]):
+                if scheduled_message["scheduled_message_id"] == event["scheduled_message_id"]:
+                    del state["scheduled_messages"][idx]
+
     elif event["type"] == "hotspots":
         state["hotspots"] = event["hotspots"]
     elif event["type"] == "custom_profile_fields":
diff --git a/zerver/openapi/zulip.yaml b/zerver/openapi/zulip.yaml
index fac24d84d9..80a299423a 100644
--- a/zerver/openapi/zulip.yaml
+++ b/zerver/openapi/zulip.yaml
@@ -4430,6 +4430,106 @@ paths:
                                   "op": "update",
                                   "draft_id": 17,
                                 }
+                            - type: object
+                              additionalProperties: false
+                              description: |
+                                Event sent to a user's clients when scheduled messages
+                                are created.
+                              properties:
+                                id:
+                                  $ref: "#/components/schemas/EventIdSchema"
+                                type:
+                                  allOf:
+                                    - $ref: "#/components/schemas/EventTypeSchema"
+                                    - enum:
+                                        - scheduled_messages
+                                op:
+                                  type: string
+                                  enum:
+                                    - add
+                                scheduled_messages:
+                                  type: array
+                                  description: |
+                                    An array of objects containing details of the newly created
+                                    scheduled messages.
+                                  items:
+                                    $ref: "#/components/schemas/ScheduledMessage"
+                              example:
+                                {
+                                  "type": "scheduled_messages",
+                                  "op": "add",
+                                  "scheduled_messages":
+                                    [
+                                      {
+                                        "scheduled_message_id": 17,
+                                        "type": "private",
+                                        "to": [6],
+                                        "content": "Hello there!",
+                                        "rendered_content": "<p>Hello there!</p>",
+                                        "scheduled_delivery_timestamp": 1681662420,
+                                      },
+                                    ],
+                                }
+                            - type: object
+                              additionalProperties: false
+                              description: |
+                                Event sent to a user's clients when a scheduled message
+                                is edited.
+                              properties:
+                                id:
+                                  $ref: "#/components/schemas/EventIdSchema"
+                                type:
+                                  allOf:
+                                    - $ref: "#/components/schemas/EventTypeSchema"
+                                    - enum:
+                                        - scheduled_messages
+                                op:
+                                  type: string
+                                  enum:
+                                    - update
+                                scheduled_message:
+                                  $ref: "#/components/schemas/ScheduledMessage"
+                              example:
+                                {
+                                  "type": "scheduled_messages",
+                                  "op": "update",
+                                  "scheduled_message":
+                                    {
+                                      "scheduled_message_id": 17,
+                                      "type": "private",
+                                      "to": [6],
+                                      "content": "Hello there!",
+                                      "rendered_content": "<p>Hello there!</p>",
+                                      "scheduled_delivery_timestamp": 1681662420,
+                                    },
+                                }
+                            - type: object
+                              additionalProperties: false
+                              description: |
+                                Event sent to a user's clients when a scheduled message
+                                is deleted.
+                              properties:
+                                id:
+                                  $ref: "#/components/schemas/EventIdSchema"
+                                type:
+                                  allOf:
+                                    - $ref: "#/components/schemas/EventTypeSchema"
+                                    - enum:
+                                        - scheduled_messages
+                                op:
+                                  type: string
+                                  enum:
+                                    - remove
+                                scheduled_message_id:
+                                  type: integer
+                                  description: |
+                                    The ID of the scheduled message that was deleted.
+                              example:
+                                {
+                                  "type": "scheduled_messages",
+                                  "op": "remove",
+                                  "scheduled_message_id": 17,
+                                }
                       queue_id:
                         type: string
                         description: |
@@ -5039,7 +5139,7 @@ paths:
                               "content": "Hi",
                               "rendered_content": "<p>Hi</p>",
                               "topic": "Introduction",
-                              "deliver_at": 1681662420000,
+                              "scheduled_delivery_timestamp": 1681662420,
                             },
                           ],
                       }
@@ -5786,13 +5886,6 @@ paths:
                         type: integer
                         description: |
                           The unique ID assigned to the sent message.
-                      deliver_at:
-                        type: string
-                        description: |
-                          Present for scheduled messages, encodes the time when the message will
-                          be sent. Note that scheduled messages ("Send later") is a beta API and
-                          may change before it's a finished feature.
-                        example: "2020-06-24 11:19:54.337533+00:00"
                     example: {"msg": "", "id": 42, "result": "success"}
         "400":
           description: Bad request.
@@ -10758,6 +10851,14 @@ paths:
                           **Changes**: New in Zulip 7.0 (feature level 164). Clients should use 140
                           for older Zulip servers, since that's the value that was hardcoded in the
                           Zulip client apps prior to this parameter being introduced.
+                      scheduled_messages:
+                        type: array
+                        description: |
+                          Present if `scheduled_messages` is present in `fetch_event_types`.
+
+                          An array of all undelivered scheduled messages by the user.
+                        items:
+                          $ref: "#/components/schemas/ScheduledMessage"
                       muted_topics:
                         type: array
                         deprecated: true
diff --git a/zerver/tests/test_event_system.py b/zerver/tests/test_event_system.py
index d4afedf5af..05eec875d1 100644
--- a/zerver/tests/test_event_system.py
+++ b/zerver/tests/test_event_system.py
@@ -1240,7 +1240,7 @@ class FetchQueriesTest(ZulipTestCase):
         self.login_user(user)
 
         flush_per_request_caches()
-        with self.assert_database_query_count(37):
+        with self.assert_database_query_count(38):
             with mock.patch("zerver.lib.events.always_want") as want_mock:
                 fetch_initial_state_data(user)
 
@@ -1268,6 +1268,7 @@ class FetchQueriesTest(ZulipTestCase):
             realm_user_groups=3,
             realm_user_settings_defaults=1,
             recent_private_conversations=1,
+            scheduled_messages=1,
             starred_messages=1,
             stream=2,
             stop_words=0,
diff --git a/zerver/tests/test_events.py b/zerver/tests/test_events.py
index b2c526df09..7833fb08af 100644
--- a/zerver/tests/test_events.py
+++ b/zerver/tests/test_events.py
@@ -12,6 +12,7 @@ from typing import Any, Callable, Dict, List, Optional, Set
 from unittest import mock
 
 import orjson
+from dateutil.parser import parse as dateparser
 from django.utils.timezone import now as timezone_now
 
 from zerver.actions.alert_words import do_add_alert_words, do_remove_alert_words
@@ -76,6 +77,10 @@ from zerver.actions.realm_settings import (
     do_set_realm_signup_notifications_stream,
     do_set_realm_user_default_setting,
 )
+from zerver.actions.scheduled_messages import (
+    check_schedule_message,
+    delete_scheduled_message,
+)
 from zerver.actions.streams import (
     bulk_add_subscriptions,
     bulk_remove_subscriptions,
@@ -198,6 +203,7 @@ from zerver.lib.test_helpers import (
     reset_email_visibility_to_everyone_in_zulip_realm,
     stdout_suppressed,
 )
+from zerver.lib.timestamp import convert_to_UTC
 from zerver.lib.topic import TOPIC_NAME
 from zerver.lib.types import ProfileDataElementUpdateDict
 from zerver.models import (
@@ -3224,3 +3230,103 @@ class DraftActionTest(BaseAction):
         draft_id = do_create_drafts([dummy_draft], self.user_profile)[0].id
         action = lambda: do_delete_draft(draft_id, self.user_profile)
         self.verify_action(action)
+
+
+class ScheduledMessagesEventsTest(BaseAction):
+    def test_stream_scheduled_message_create_event(self) -> None:
+        # Create stream scheduled message
+        action = lambda: check_schedule_message(
+            self.user_profile,
+            get_client("website"),
+            "stream",
+            [self.get_stream_id("Verona")],
+            "Test topic",
+            "Stream message",
+            None,
+            convert_to_UTC(dateparser("2023-04-19 18:24:56")),
+            self.user_profile.realm,
+        )
+        self.verify_action(action)
+
+    def test_create_event_with_existing_scheduled_messages(self) -> None:
+        # Create stream scheduled message
+        check_schedule_message(
+            self.user_profile,
+            get_client("website"),
+            "stream",
+            [self.get_stream_id("Verona")],
+            "Test topic",
+            "Stream message 1",
+            None,
+            convert_to_UTC(dateparser("2023-04-19 17:24:56")),
+            self.user_profile.realm,
+        )
+
+        # Check that the new scheduled message gets appended correctly.
+        action = lambda: check_schedule_message(
+            self.user_profile,
+            get_client("website"),
+            "stream",
+            [self.get_stream_id("Verona")],
+            "Test topic",
+            "Stream message 2",
+            None,
+            convert_to_UTC(dateparser("2023-04-19 18:24:56")),
+            self.user_profile.realm,
+        )
+        self.verify_action(action)
+
+    def test_private_scheduled_message_create_event(self) -> None:
+        # Create private scheduled message
+        action = lambda: check_schedule_message(
+            self.user_profile,
+            get_client("website"),
+            "private",
+            [self.example_user("hamlet").id],
+            None,
+            "Private message",
+            None,
+            convert_to_UTC(dateparser("2023-04-19 18:24:56")),
+            self.user_profile.realm,
+        )
+        self.verify_action(action)
+
+    def test_scheduled_message_edit_event(self) -> None:
+        scheduled_message_id = check_schedule_message(
+            self.user_profile,
+            get_client("website"),
+            "stream",
+            [self.get_stream_id("Verona")],
+            "Test topic",
+            "Stream message",
+            None,
+            convert_to_UTC(dateparser("2023-04-19 18:24:56")),
+            self.user_profile.realm,
+        )
+        action = lambda: check_schedule_message(
+            self.user_profile,
+            get_client("website"),
+            "stream",
+            [self.get_stream_id("Verona")],
+            "Edited test topic",
+            "Edited stream message",
+            scheduled_message_id,
+            convert_to_UTC(dateparser("2023-04-20 18:24:56")),
+            self.user_profile.realm,
+        )
+        self.verify_action(action)
+
+    def test_scheduled_message_delete_event(self) -> None:
+        scheduled_message_id = check_schedule_message(
+            self.user_profile,
+            get_client("website"),
+            "stream",
+            [self.get_stream_id("Verona")],
+            "Test topic",
+            "Stream message",
+            None,
+            convert_to_UTC(dateparser("2023-04-19 18:24:56")),
+            self.user_profile.realm,
+        )
+        action = lambda: delete_scheduled_message(self.user_profile, scheduled_message_id)
+        self.verify_action(action)
diff --git a/zerver/tests/test_home.py b/zerver/tests/test_home.py
index b21fcbde61..3d638d0067 100644
--- a/zerver/tests/test_home.py
+++ b/zerver/tests/test_home.py
@@ -188,6 +188,7 @@ class HomeTest(ZulipTestCase):
         "realm_wildcard_mention_policy",
         "recent_private_conversations",
         "request_language",
+        "scheduled_messages",
         "search_pills_enabled",
         "server_avatar_changes_disabled",
         "server_emoji_data_url",
@@ -248,7 +249,7 @@ class HomeTest(ZulipTestCase):
 
         # Verify succeeds once logged-in
         flush_per_request_caches()
-        with self.assert_database_query_count(48):
+        with self.assert_database_query_count(49):
             with patch("zerver.lib.cache.cache_set") as cache_mock:
                 result = self._get_home_page(stream="Denmark")
                 self.check_rendered_logged_in_app(result)
@@ -439,7 +440,7 @@ class HomeTest(ZulipTestCase):
         # Verify number of queries for Realm admin isn't much higher than for normal users.
         self.login("iago")
         flush_per_request_caches()
-        with self.assert_database_query_count(45):
+        with self.assert_database_query_count(46):
             with patch("zerver.lib.cache.cache_set") as cache_mock:
                 result = self._get_home_page()
                 self.check_rendered_logged_in_app(result)
@@ -471,7 +472,7 @@ class HomeTest(ZulipTestCase):
 
         # Then for the second page load, measure the number of queries.
         flush_per_request_caches()
-        with self.assert_database_query_count(43):
+        with self.assert_database_query_count(44):
             result = self._get_home_page()
 
         # Do a sanity check that our new streams were in the payload.