diff --git a/templates/zerver/api/changelog.md b/templates/zerver/api/changelog.md
index 73a3cf57d3..41152ddec1 100644
--- a/templates/zerver/api/changelog.md
+++ b/templates/zerver/api/changelog.md
@@ -29,6 +29,9 @@ format used by the Zulip server that they are interacting with.
   /messages/flags`](/api/update-message-flags) no longer redundantly
   lists messages where the flag was set to the same state it was
   already in.
+* [`POST /messages/flags/narrow`](/api/update-message-flags-for-narrow):
+  This new endpoint allows updating message flags on a range of
+  messages within a narrow.
 
 **Feature level 154**
 
diff --git a/templates/zerver/help/include/rest-endpoints.md b/templates/zerver/help/include/rest-endpoints.md
index 1fb9b90d1f..e70655489b 100644
--- a/templates/zerver/help/include/rest-endpoints.md
+++ b/templates/zerver/help/include/rest-endpoints.md
@@ -13,6 +13,7 @@
 * [Check if messages match narrow](/api/check-messages-match-narrow)
 * [Get a message's edit history](/api/get-message-history)
 * [Update personal message flags](/api/update-message-flags)
+* [Update personal message flags for narrow](/api/update-message-flags-for-narrow)
 * [Mark messages as read in bulk](/api/mark-all-as-read)
 * [Get a message's read receipts](/api/get-read-receipts)
 
diff --git a/zerver/openapi/zulip.yaml b/zerver/openapi/zulip.yaml
index 9abd3e472c..43a8d532d7 100644
--- a/zerver/openapi/zulip.yaml
+++ b/zerver/openapi/zulip.yaml
@@ -5071,9 +5071,6 @@ paths:
             - `first_unread`: The oldest unread message matching the
               query, if any; otherwise, the most recent message.
 
-            The special values of `'newest'` and `'oldest'` are also supported
-            for anchoring the query at the most recent or oldest messages.
-
             **Changes**: String values are new in Zulip 3.0 (feature level 1). The
             `first_unread` functionality was supported in Zulip 2.1.x
             and older by not sending `anchor` and using `use_first_unread_anchor`.
@@ -5663,6 +5660,8 @@ paths:
         Add or remove personal message flags like `read` and `starred`
         on a collection of message IDs.
 
+        See also the endpoint for [updating flags on a range of
+        messages within a narrow](/api/update-message-flags-for-narrow).
         For updating the `read` flag on common collections of messages, see also
         the
         [special endpoints for marking message as read in bulk](/api/mark-all-as-read).
@@ -5792,6 +5791,205 @@ paths:
                           An array with the IDs of the modified messages.
                     example:
                       {"msg": "", "messages": [4, 18, 15], "result": "success"}
+  /messages/flags/narrow:
+    post:
+      operationId: update-message-flags-for-narrow
+      summary: Update personal message flags for narrow
+      tags: ["messages"]
+      description: |
+        Add or remove personal message flags like `read` and `starred`
+        on a range of messages within a narrow.
+
+        See also [the endpoint for updating flags on specific message
+        IDs](/api/update-message-flags).
+
+        **Changes**: New in Zulip 6.0 (feature level 155).
+      x-curl-examples-parameters:
+        oneOf:
+          - type: exclude
+            parameters:
+              enum:
+                - include_anchor
+      parameters:
+        - name: anchor
+          in: query
+          description: |
+            Integer message ID to anchor updating of flags. Supports special
+            string values for when the client wants the server to compute the anchor
+            to use:
+
+            - `newest`: The most recent message.
+            - `oldest`: The oldest message.
+            - `first_unread`: The oldest unread message matching the
+              query, if any; otherwise, the most recent message.
+          schema:
+            oneOf:
+              - type: string
+              - type: integer
+          example: 43
+          required: true
+        - name: include_anchor
+          in: query
+          description: |
+            Whether a message with the specified ID matching the narrow
+            should be included in the update range.
+          schema:
+            type: boolean
+            default: true
+          example: false
+        - name: num_before
+          in: query
+          description: |
+            Limit the number of messages preceding the anchor in the
+            update range. The server may decrease this to bound
+            transaction sizes.
+          schema:
+            type: integer
+            minimum: 0
+          example: 4
+          required: true
+        - name: num_after
+          in: query
+          description: |
+            Limit the number of messages following the anchor in the
+            update range. The server may decrease this to bound
+            transaction sizes.
+          schema:
+            type: integer
+            minimum: 0
+          example: 8
+          required: true
+        - name: narrow
+          in: query
+          description: |
+            The narrow you want update flags within. See how to
+            [construct a narrow](/api/construct-narrow).
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  oneOf:
+                    - type: object
+                      required:
+                        - operator
+                        - operand
+                      additionalProperties: false
+                      properties:
+                        operator:
+                          type: string
+                        operand:
+                          oneOf:
+                            - type: string
+                            - type: integer
+                            - type: array
+                              items:
+                                type: integer
+                        negated:
+                          type: boolean
+                    - type: array
+                      items:
+                        type: string
+                      minItems: 2
+                      maxItems: 2
+                default: []
+              example: [{"operand": "Denmark", "operator": "stream"}]
+          required: true
+        - name: op
+          in: query
+          description: |
+            Whether to `add` the flag or `remove` it.
+          schema:
+            type: string
+            enum:
+              - add
+              - remove
+          example: add
+          required: true
+        - name: flag
+          in: query
+          description: |
+            The flag that should be added/removed. See [available
+            flags](/api/update-message-flags#available-flags).
+          schema:
+            type: string
+          example: read
+          required: true
+      responses:
+        "200":
+          description: Success.
+          content:
+            application/json:
+              schema:
+                allOf:
+                  - $ref: "#/components/schemas/JsonSuccessBase"
+                  - $ref: "#/components/schemas/SuccessDescription"
+                  - additionalProperties: false
+                    required:
+                      - processed_count
+                      - updated_count
+                      - first_processed_id
+                      - last_processed_id
+                      - found_oldest
+                      - found_newest
+                    properties:
+                      result: {}
+                      msg: {}
+                      processed_count:
+                        type: integer
+                        description: |
+                          The number of messages that were within the
+                          update range (at most `num_before + 1 +
+                          num_after`).
+                      updated_count:
+                        type: integer
+                        description: |
+                          The number of messages where the flag's
+                          value was changed (at most
+                          `processed_count`).
+                      first_processed_id:
+                        type: integer
+                        nullable: true
+                        description: |
+                          The ID of the oldest message within the
+                          update range, or `null` if the range was
+                          empty.
+                      last_processed_id:
+                        type: integer
+                        nullable: true
+                        description: |
+                          The ID of the newest message within the
+                          update range, or `null` if the range was
+                          empty.
+                      found_oldest:
+                        type: boolean
+                        description: |
+                          Whether the update range reached backward
+                          far enough to include very oldest message
+                          matching the narrow (used by clients doing a
+                          bulk update to decide whether to issue
+                          another request anchored at
+                          `first_processed_id`).
+                      found_newest:
+                        type: boolean
+                        description: |
+                          Whether the update range reached forward far
+                          enough to include very oldest message
+                          matching the narrow (used by clients doing a
+                          bulk update to decide whether to issue
+                          another request anchored at
+                          `last_processed_id`).
+                    example:
+                      {
+                        "result": "success",
+                        "msg": "",
+                        "processed_count": 11,
+                        "updated_count": 8,
+                        "first_processed_id": 35,
+                        "last_processed_id": 55,
+                        "found_oldest": false,
+                        "found_newest": true,
+                      }
   /messages/render:
     post:
       operationId: render-message
diff --git a/zerver/tests/test_message_flags.py b/zerver/tests/test_message_flags.py
index 4d5f1e0521..f8f423ee80 100644
--- a/zerver/tests/test_message_flags.py
+++ b/zerver/tests/test_message_flags.py
@@ -224,6 +224,92 @@ class UnreadCountTests(ZulipTestCase):
             elif msg["id"] == self.unread_msg_ids[1]:
                 check_flags(msg["flags"], set())
 
+    def test_update_flags_for_narrow(self) -> None:
+        user = self.example_user("hamlet")
+        self.login_user(user)
+        message_ids = [
+            self.send_stream_message(
+                self.example_user("cordelia"), "Verona", topic_name=f"topic {i % 2}"
+            )
+            for i in range(10)
+        ]
+
+        response = self.assert_json_success(
+            self.client_post(
+                "/json/messages/flags/narrow",
+                {
+                    "anchor": message_ids[5],
+                    "num_before": 2,
+                    "num_after": 2,
+                    "narrow": "[]",
+                    "op": "add",
+                    "flag": "read",
+                },
+            )
+        )
+        self.assertEqual(response["processed_count"], 5)
+        self.assertEqual(response["updated_count"], 5)
+        self.assertEqual(response["first_processed_id"], message_ids[3])
+        self.assertEqual(response["last_processed_id"], message_ids[7])
+        self.assertEqual(response["found_oldest"], False)
+        self.assertEqual(response["found_newest"], False)
+        self.assertCountEqual(
+            UserMessage.objects.filter(user_profile_id=user.id, message_id__in=message_ids)
+            .extra(where=[UserMessage.where_read()])
+            .values_list("message_id", flat=True),
+            message_ids[3:8],
+        )
+
+        response = self.assert_json_success(
+            self.client_post(
+                "/json/messages/flags/narrow",
+                {
+                    "anchor": message_ids[3],
+                    "include_anchor": "false",
+                    "num_before": 0,
+                    "num_after": 5,
+                    "narrow": orjson.dumps(
+                        [
+                            {"operator": "stream", "operand": "Verona"},
+                            {"operator": "topic", "operand": "topic 1"},
+                        ]
+                    ).decode(),
+                    "op": "add",
+                    "flag": "starred",
+                },
+            )
+        )
+        # In this topic (1, 3, 5, 7, 9), processes everything after 3.
+        self.assertEqual(response["processed_count"], 3)
+        self.assertEqual(response["updated_count"], 3)
+        self.assertEqual(response["first_processed_id"], message_ids[5])
+        self.assertEqual(response["last_processed_id"], message_ids[9])
+        self.assertEqual(response["found_oldest"], False)
+        self.assertEqual(response["found_newest"], True)
+        self.assertCountEqual(
+            UserMessage.objects.filter(user_profile_id=user.id, message_id__in=message_ids)
+            .extra(where=[UserMessage.where_starred()])
+            .values_list("message_id", flat=True),
+            message_ids[5::2],
+        )
+
+    def test_update_flags_for_narrow_misuse(self) -> None:
+        self.login("hamlet")
+
+        response = self.client_post(
+            "/json/messages/flags/narrow",
+            {
+                "anchor": "0",
+                "include_anchor": "false",
+                "num_before": "1",
+                "num_after": "1",
+                "narrow": "[]",
+                "op": "add",
+                "flag": "read",
+            },
+        )
+        self.assert_json_error(response, "The anchor can only be excluded at an end of the range")
+
     def test_mark_all_in_stream_read(self) -> None:
         self.login("hamlet")
         user_profile = self.example_user("hamlet")
diff --git a/zerver/views/message_flags.py b/zerver/views/message_flags.py
index e7d92a477d..739a7c0fd0 100644
--- a/zerver/views/message_flags.py
+++ b/zerver/views/message_flags.py
@@ -9,18 +9,27 @@ from zerver.actions.message_flags import (
     do_update_message_flags,
 )
 from zerver.lib.exceptions import ErrorCode, JsonableError
+from zerver.lib.narrow import (
+    OptionalNarrowListT,
+    fetch_messages,
+    narrow_parameter,
+    parse_anchor_value,
+)
 from zerver.lib.request import REQ, RequestNotes, has_request_variables
 from zerver.lib.response import json_partial_success, json_success
 from zerver.lib.streams import access_stream_by_id
 from zerver.lib.timeout import TimeoutExpired, timeout
 from zerver.lib.topic import user_message_exists_for_topic
-from zerver.lib.validator import check_int, check_list
+from zerver.lib.validator import check_bool, check_int, check_list, to_non_negative_int
 from zerver.models import UserActivity, UserProfile
 
 
 def get_latest_update_message_flag_activity(user_profile: UserProfile) -> Optional[UserActivity]:
     return (
-        UserActivity.objects.filter(user_profile=user_profile, query="update_message_flags")
+        UserActivity.objects.filter(
+            user_profile=user_profile,
+            query__in=["update_message_flags", "update_message_flags_for_narrow"],
+        )
         .order_by("last_visit")
         .last()
     )
@@ -45,7 +54,66 @@ def update_message_flags(
     log_data_str = f"[{operation} {flag}/{target_count_str}] actually {count}"
     request_notes.log_data["extra"] = log_data_str
 
-    return json_success(request, data={"messages": messages})
+    return json_success(
+        request,
+        data={
+            "messages": messages,  # Useless, but included for backwards compatibility.
+        },
+    )
+
+
+MAX_MESSAGES_PER_UPDATE = 5000
+
+# NOTE: If this function name is changed, add the new name to the
+# query in get_latest_update_message_flag_activity
+@has_request_variables
+def update_message_flags_for_narrow(
+    request: HttpRequest,
+    user_profile: UserProfile,
+    anchor_val: str = REQ("anchor"),
+    include_anchor: bool = REQ(json_validator=check_bool, default=True),
+    num_before: int = REQ(converter=to_non_negative_int),
+    num_after: int = REQ(converter=to_non_negative_int),
+    narrow: OptionalNarrowListT = REQ("narrow", converter=narrow_parameter),
+    operation: str = REQ("op"),
+    flag: str = REQ(),
+) -> HttpResponse:
+    anchor = parse_anchor_value(anchor_val, use_first_unread_anchor=False)
+
+    if num_before > 0 and num_after > 0 and not include_anchor:
+        raise JsonableError(_("The anchor can only be excluded at an end of the range"))
+
+    # Clamp such that num_before + num_after <= MAX_MESSAGES_PER_UPDATE.
+    num_before = min(
+        num_before, max(MAX_MESSAGES_PER_UPDATE - num_after, MAX_MESSAGES_PER_UPDATE // 2)
+    )
+    num_after = min(num_after, MAX_MESSAGES_PER_UPDATE - num_before)
+
+    query_info = fetch_messages(
+        narrow=narrow,
+        user_profile=user_profile,
+        realm=user_profile.realm,
+        is_web_public_query=False,
+        anchor=anchor,
+        include_anchor=include_anchor,
+        num_before=num_before,
+        num_after=num_after,
+    )
+
+    messages = [row[0] for row in query_info.rows]
+    updated_count = do_update_message_flags(user_profile, operation, flag, messages)
+
+    return json_success(
+        request,
+        data={
+            "processed_count": len(messages),
+            "updated_count": updated_count,
+            "first_processed_id": messages[0] if messages else None,
+            "last_processed_id": messages[-1] if messages else None,
+            "found_oldest": query_info.found_oldest,
+            "found_newest": query_info.found_newest,
+        },
+    )
 
 
 @has_request_variables
diff --git a/zproject/urls.py b/zproject/urls.py
index bffb061f92..54eb315656 100644
--- a/zproject/urls.py
+++ b/zproject/urls.py
@@ -75,6 +75,7 @@ from zerver.views.message_flags import (
     mark_stream_as_read,
     mark_topic_as_read,
     update_message_flags,
+    update_message_flags_for_narrow,
 )
 from zerver.views.message_send import render_message_backend, send_message_backend, zcommand_backend
 from zerver.views.muting import mute_user, unmute_user, update_muted_topic
@@ -329,6 +330,7 @@ v1_api_and_json_patterns = [
     ),
     rest_path("messages/render", POST=render_message_backend),
     rest_path("messages/flags", POST=update_message_flags),
+    rest_path("messages/flags/narrow", POST=update_message_flags_for_narrow),
     rest_path("messages/<int:message_id>/history", GET=get_message_edit_history),
     rest_path("messages/matches_narrow", GET=messages_in_narrow_backend),
     rest_path("users/me/subscriptions/properties", POST=update_subscription_properties_backend),