diff --git a/zerver/lib/events.py b/zerver/lib/events.py
index 6e7e3efab8..59c567b613 100644
--- a/zerver/lib/events.py
+++ b/zerver/lib/events.py
@@ -35,7 +35,11 @@ from zerver.lib.message import (
     remove_message_id_from_unread_mgs,
 )
 from zerver.lib.muted_users import get_user_mutes
-from zerver.lib.narrow import check_supported_events_narrow_filter, read_stop_words
+from zerver.lib.narrow import (
+    check_supported_events_narrow_filter,
+    narrow_dataclasses_from_tuples,
+    read_stop_words,
+)
 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
@@ -1466,10 +1470,14 @@ def do_events_register(
     spectator_requested_language: Optional[str] = None,
     pronouns_field_type_supported: bool = True,
 ) -> Dict[str, Any]:
+    # TODO: We eventually want to upstream this code to the caller, but
+    # serialization concerns make it a bit difficult.
+    modern_narrow = narrow_dataclasses_from_tuples(narrow)
+
     # Technically we don't need to check this here because
     # build_narrow_filter will check it, but it's nicer from an error
     # handling perspective to do it before contacting Tornado
-    check_supported_events_narrow_filter(narrow)
+    check_supported_events_narrow_filter(modern_narrow)
 
     notification_settings_null = client_capabilities.get("notification_settings_null", False)
     bulk_message_deletion = client_capabilities.get("bulk_message_deletion", False)
diff --git a/zerver/lib/narrow.py b/zerver/lib/narrow.py
index a5962aca0b..9cf9bd869a 100644
--- a/zerver/lib/narrow.py
+++ b/zerver/lib/narrow.py
@@ -85,9 +85,25 @@ from zerver.models import (
     get_user_including_cross_realm,
 )
 
+
+@dataclass
+class NarrowTerm:
+    # In our current use case we don't yet handle negated narrow terms.
+    operator: str
+    operand: str
+
+
 stop_words_list: Optional[List[str]] = None
 
 
+def narrow_dataclasses_from_tuples(tups: Collection[Sequence[str]]) -> Collection[NarrowTerm]:
+    """
+    This method assumes that the callers are in our event-handling codepath, and
+    therefore as of summer 2023, they do not yet support the "negated" flag.
+    """
+    return [NarrowTerm(operator=tup[0], operand=tup[1]) for tup in tups]
+
+
 def read_stop_words() -> List[str]:
     global stop_words_list
     if stop_words_list is None:
@@ -100,9 +116,9 @@ def read_stop_words() -> List[str]:
     return stop_words_list
 
 
-def check_supported_events_narrow_filter(narrow: Iterable[Sequence[str]]) -> None:
-    for element in narrow:
-        operator = element[0]
+def check_supported_events_narrow_filter(narrow: Collection[NarrowTerm]) -> None:
+    for narrow_term in narrow:
+        operator = narrow_term.operator
         if operator not in ["stream", "topic", "sender", "is"]:
             raise JsonableError(_("Operator {} not supported.").format(operator))
 
@@ -133,16 +149,14 @@ def is_web_public_narrow(narrow: Optional[Iterable[Dict[str, Any]]]) -> bool:
 
 
 def build_narrow_filter(
-    narrow: Collection[Sequence[str]],
+    narrow: Collection[NarrowTerm],
 ) -> Callable[[NamedArg(Dict[str, Any], "message"), NamedArg(List[str], "flags")], bool]:
     """Changes to this function should come with corresponding changes to
     NarrowLibraryTest."""
     check_supported_events_narrow_filter(narrow)
 
     def narrow_filter(*, message: Dict[str, Any], flags: List[str]) -> bool:
-        for element in narrow:
-            operator = element[0]
-            operand = element[1]
+        def satisfies_operator(*, operator: str, operand: str) -> bool:
             if operator == "stream":
                 if message["type"] != "stream":
                     return False
@@ -176,6 +190,12 @@ def build_narrow_filter(
                 topic_name = get_topic_from_message_info(message)
                 if not topic_name.startswith(RESOLVED_TOPIC_PREFIX):
                     return False
+            return True
+
+        for narrow_term in narrow:
+            # TODO: Eventually handle negated narrow terms.
+            if not satisfies_operator(operator=narrow_term.operator, operand=narrow_term.operand):
+                return False
 
         return True
 
diff --git a/zerver/tests/test_message_fetch.py b/zerver/tests/test_message_fetch.py
index 4025406c42..030175963d 100644
--- a/zerver/tests/test_message_fetch.py
+++ b/zerver/tests/test_message_fetch.py
@@ -30,6 +30,7 @@ from zerver.lib.narrow import (
     LARGER_THAN_MAX_MESSAGE_ID,
     BadNarrowOperatorError,
     NarrowBuilder,
+    NarrowTerm,
     build_narrow_filter,
     exclude_muting_conditions,
     find_first_unread_anchor,
@@ -592,7 +593,7 @@ class NarrowBuilderTest(ZulipTestCase):
 
 class NarrowLibraryTest(ZulipTestCase):
     def test_build_narrow_filter(self) -> None:
-        narrow_filter = build_narrow_filter([["stream", "devel"]])
+        narrow_filter = build_narrow_filter([NarrowTerm(operator="stream", operand="devel")])
 
         self.assertTrue(
             narrow_filter(
@@ -616,7 +617,7 @@ class NarrowLibraryTest(ZulipTestCase):
 
         ###
 
-        narrow_filter = build_narrow_filter([["topic", "bark"]])
+        narrow_filter = build_narrow_filter([NarrowTerm(operator="topic", operand="bark")])
 
         self.assertTrue(
             narrow_filter(
@@ -652,7 +653,12 @@ class NarrowLibraryTest(ZulipTestCase):
 
         ###
 
-        narrow_filter = build_narrow_filter([["stream", "devel"], ["topic", "python"]])
+        narrow_filter = build_narrow_filter(
+            [
+                NarrowTerm(operator="stream", operand="devel"),
+                NarrowTerm(operator="topic", operand="python"),
+            ]
+        )
 
         self.assertTrue(
             narrow_filter(
@@ -682,7 +688,9 @@ class NarrowLibraryTest(ZulipTestCase):
 
         ###
 
-        narrow_filter = build_narrow_filter([["sender", "hamlet@zulip.com"]])
+        narrow_filter = build_narrow_filter(
+            [NarrowTerm(operator="sender", operand="hamlet@zulip.com")]
+        )
 
         self.assertTrue(
             narrow_filter(
@@ -700,7 +708,7 @@ class NarrowLibraryTest(ZulipTestCase):
 
         ###
 
-        narrow_filter = build_narrow_filter([["is", "dm"]])
+        narrow_filter = build_narrow_filter([NarrowTerm(operator="is", operand="dm")])
 
         self.assertTrue(
             narrow_filter(
@@ -718,7 +726,7 @@ class NarrowLibraryTest(ZulipTestCase):
 
         ###
 
-        narrow_filter = build_narrow_filter([["is", "private"]])
+        narrow_filter = build_narrow_filter([NarrowTerm(operator="is", operand="private")])
 
         self.assertTrue(
             narrow_filter(
@@ -736,7 +744,7 @@ class NarrowLibraryTest(ZulipTestCase):
 
         ###
 
-        narrow_filter = build_narrow_filter([["is", "starred"]])
+        narrow_filter = build_narrow_filter([NarrowTerm(operator="is", operand="starred")])
 
         self.assertTrue(
             narrow_filter(
@@ -754,7 +762,7 @@ class NarrowLibraryTest(ZulipTestCase):
 
         ###
 
-        narrow_filter = build_narrow_filter([["is", "alerted"]])
+        narrow_filter = build_narrow_filter([NarrowTerm(operator="is", operand="alerted")])
 
         self.assertTrue(
             narrow_filter(
@@ -772,7 +780,7 @@ class NarrowLibraryTest(ZulipTestCase):
 
         ###
 
-        narrow_filter = build_narrow_filter([["is", "mentioned"]])
+        narrow_filter = build_narrow_filter([NarrowTerm(operator="is", operand="mentioned")])
 
         self.assertTrue(
             narrow_filter(
@@ -790,7 +798,7 @@ class NarrowLibraryTest(ZulipTestCase):
 
         ###
 
-        narrow_filter = build_narrow_filter([["is", "unread"]])
+        narrow_filter = build_narrow_filter([NarrowTerm(operator="is", operand="unread")])
 
         self.assertTrue(
             narrow_filter(
@@ -808,7 +816,7 @@ class NarrowLibraryTest(ZulipTestCase):
 
         ###
 
-        narrow_filter = build_narrow_filter([["is", "resolved"]])
+        narrow_filter = build_narrow_filter([NarrowTerm(operator="is", operand="resolved")])
 
         self.assertTrue(
             narrow_filter(
@@ -832,7 +840,7 @@ class NarrowLibraryTest(ZulipTestCase):
 
     def test_build_narrow_filter_invalid(self) -> None:
         with self.assertRaises(JsonableError):
-            build_narrow_filter(["invalid_operator", "operand"])
+            build_narrow_filter([NarrowTerm(operator="invalid_operator", operand="operand")])
 
     def test_is_spectator_compatible(self) -> None:
         self.assertTrue(is_spectator_compatible([]))
diff --git a/zerver/tornado/event_queue.py b/zerver/tornado/event_queue.py
index 3fdb90ad5f..26addb0bd2 100644
--- a/zerver/tornado/event_queue.py
+++ b/zerver/tornado/event_queue.py
@@ -39,7 +39,7 @@ from tornado import autoreload
 from version import API_FEATURE_LEVEL, ZULIP_MERGE_BASE, ZULIP_VERSION
 from zerver.lib.exceptions import JsonableError
 from zerver.lib.message import MessageDict
-from zerver.lib.narrow import build_narrow_filter
+from zerver.lib.narrow import build_narrow_filter, narrow_dataclasses_from_tuples
 from zerver.lib.notification_data import UserMessageNotificationsData
 from zerver.lib.queue import queue_json_publish, retry_event
 from zerver.middleware import async_request_timer_restart
@@ -97,6 +97,10 @@ class ClientDescriptor:
         pronouns_field_type_supported: bool = True,
         linkifier_url_template: bool = False,
     ) -> None:
+        # TODO: We eventually want to upstream this code to the caller, but
+        # serialization concerns make it a bit difficult.
+        modern_narrow = narrow_dataclasses_from_tuples(narrow)
+
         # These objects are serialized on shutdown and restored on restart.
         # If fields are added or semantics are changed, temporary code must be
         # added to load_event_queues() to update the restored objects.
@@ -115,7 +119,7 @@ class ClientDescriptor:
         self.client_type_name = client_type_name
         self._timeout_handle: Any = None  # TODO: should be return type of ioloop.call_later
         self.narrow = narrow
-        self.narrow_filter = build_narrow_filter(narrow)
+        self.narrow_filter = build_narrow_filter(modern_narrow)
         self.bulk_message_deletion = bulk_message_deletion
         self.stream_typing_notifications = stream_typing_notifications
         self.user_settings_object = user_settings_object