diff --git a/changelog.d/15004.feature b/changelog.d/15004.feature
new file mode 100644
index 000000000000..d11d0aca91da
--- /dev/null
+++ b/changelog.d/15004.feature
@@ -0,0 +1 @@
+Implement [MSC3873](https://github.com/matrix-org/matrix-spec-proposals/pull/3873) to unambiguate push rule keys with dots in them.
diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py
index 53c0682dfdb6..5e3a889081b0 100644
--- a/synapse/config/experimental.py
+++ b/synapse/config/experimental.py
@@ -169,6 +169,11 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None:
         # MSC3925: do not replace events with their edits
         self.msc3925_inhibit_edit = experimental.get("msc3925_inhibit_edit", False)
 
+        # MSC3873: Disambiguate event_match keys.
+        self.msc3783_escape_event_match_key = experimental.get(
+            "msc3783_escape_event_match_key", False
+        )
+
         # MSC3952: Intentional mentions
         self.msc3952_intentional_mentions = experimental.get(
             "msc3952_intentional_mentions", False
diff --git a/synapse/push/bulk_push_rule_evaluator.py b/synapse/push/bulk_push_rule_evaluator.py
index d9c0a98f44ee..39d2f88f0349 100644
--- a/synapse/push/bulk_push_rule_evaluator.py
+++ b/synapse/push/bulk_push_rule_evaluator.py
@@ -271,7 +271,10 @@ async def _related_events(self, event: EventBase) -> Dict[str, Dict[str, str]]:
                     related_event_id, allow_none=True
                 )
                 if related_event is not None:
-                    related_events[relation_type] = _flatten_dict(related_event)
+                    related_events[relation_type] = _flatten_dict(
+                        related_event,
+                        msc3783_escape_event_match_key=self.hs.config.experimental.msc3783_escape_event_match_key,
+                    )
 
             reply_event_id = (
                 event.content.get("m.relates_to", {})
@@ -286,7 +289,10 @@ async def _related_events(self, event: EventBase) -> Dict[str, Dict[str, str]]:
                 )
 
                 if related_event is not None:
-                    related_events["m.in_reply_to"] = _flatten_dict(related_event)
+                    related_events["m.in_reply_to"] = _flatten_dict(
+                        related_event,
+                        msc3783_escape_event_match_key=self.hs.config.experimental.msc3783_escape_event_match_key,
+                    )
 
                     # indicate that this is from a fallback relation.
                     if relation_type == "m.thread" and event.content.get(
@@ -405,7 +411,10 @@ async def _action_for_event_by_user(
             room_mention = mentions.get("room") is True
 
         evaluator = PushRuleEvaluator(
-            _flatten_dict(event),
+            _flatten_dict(
+                event,
+                msc3783_escape_event_match_key=self.hs.config.experimental.msc3783_escape_event_match_key,
+            ),
             has_mentions,
             user_mentions,
             room_mention,
@@ -493,6 +502,8 @@ def _flatten_dict(
     d: Union[EventBase, Mapping[str, Any]],
     prefix: Optional[List[str]] = None,
     result: Optional[Dict[str, str]] = None,
+    *,
+    msc3783_escape_event_match_key: bool = False,
 ) -> Dict[str, str]:
     """
     Given a JSON dictionary (or event) which might contain sub dictionaries,
@@ -521,11 +532,22 @@ def _flatten_dict(
     if result is None:
         result = {}
     for key, value in d.items():
+        if msc3783_escape_event_match_key:
+            # Escape periods in the key with a backslash (and backslashes with an
+            # extra backslash). This is since a period is used as a separator between
+            # nested fields.
+            key = key.replace("\\", "\\\\").replace(".", "\\.")
+
         if isinstance(value, str):
             result[".".join(prefix + [key])] = value.lower()
         elif isinstance(value, Mapping):
             # do not set `room_version` due to recursion considerations below
-            _flatten_dict(value, prefix=(prefix + [key]), result=result)
+            _flatten_dict(
+                value,
+                prefix=(prefix + [key]),
+                result=result,
+                msc3783_escape_event_match_key=msc3783_escape_event_match_key,
+            )
 
     # `room_version` should only ever be set when looking at the top level of an event
     if (
diff --git a/tests/push/test_push_rule_evaluator.py b/tests/push/test_push_rule_evaluator.py
index da3342387121..516b65cc3c97 100644
--- a/tests/push/test_push_rule_evaluator.py
+++ b/tests/push/test_push_rule_evaluator.py
@@ -48,6 +48,14 @@ def test_nested(self) -> None:
         input = {"foo": {"bar": "abc"}}
         self.assertEqual({"foo.bar": "abc"}, _flatten_dict(input))
 
+        # If a field has a dot in it, escape it.
+        input = {"m.foo": {"b\\ar": "abc"}}
+        self.assertEqual({"m.foo.b\\ar": "abc"}, _flatten_dict(input))
+        self.assertEqual(
+            {"m\\.foo.b\\\\ar": "abc"},
+            _flatten_dict(input, msc3783_escape_event_match_key=True),
+        )
+
     def test_non_string(self) -> None:
         """Non-string items are dropped."""
         input: Dict[str, Any] = {