Fix # root wildcard, top-level imports, regex cache

Author: elijahr

src/mcp/client/session.py

@@ -1,4 +1,5 @@
 import logging
+import re
 from datetime import timedelta
 from typing import Any, Protocol, overload
 
@@ -282,24 +283,25 @@ def decorator(fn: EventHandlerFnT) -> EventHandlerFnT:
 
     def _topic_matches_subscriptions(self, topic: str) -> bool:
         """Check if a topic matches any of our subscribed patterns."""
-        import re as _re
-
         for pattern in self._subscribed_patterns:
             parts = pattern.split("/")
             regex_parts: list[str] = []
             for i, part in enumerate(parts):
                 if part == "#":
-                    regex = "^" + "/".join(regex_parts) + "(/.*)?$"
-                    if _re.match(regex, topic):
+                    if regex_parts:
+                        regex = "^" + "/".join(regex_parts) + "(/.*)?$"
+                    else:
+                        regex = "^.*$"
+                    if re.match(regex, topic):
                         return True
                     break
                 elif part == "+":
                     regex_parts.append("[^/]+")
                 else:
-                    regex_parts.append(_re.escape(part))
+                    regex_parts.append(re.escape(part))
             else:
                 regex = "^" + "/".join(regex_parts) + "$"
-                if _re.match(regex, topic):
+                if re.match(regex, topic):
                     return True
         return False
 
@@ -312,23 +314,24 @@ async def _handle_event(self, params: types.EventParams) -> None:
             return
 
         if self._event_topic_filter is not None:
-            import re as _re
-
             parts = self._event_topic_filter.split("/")
             regex_parts: list[str] = []
             matched = False
             for i, part in enumerate(parts):
                 if part == "#":
-                    regex = "^" + "/".join(regex_parts) + "(/.*)?$"
-                    matched = bool(_re.match(regex, params.topic))
+                    if regex_parts:
+                        regex = "^" + "/".join(regex_parts) + "(/.*)?$"
+                    else:
+                        regex = "^.*$"
+                    matched = bool(re.match(regex, params.topic))
                     break
                 elif part == "+":
                     regex_parts.append("[^/]+")
                 else:
-                    regex_parts.append(_re.escape(part))
+                    regex_parts.append(re.escape(part))
             else:
                 regex = "^" + "/".join(regex_parts) + "$"
-                matched = bool(_re.match(regex, params.topic))
+                matched = bool(re.match(regex, params.topic))
             if not matched:
                 return
 

src/mcp/server/events.py

@@ -30,9 +30,14 @@ def _pattern_to_regex(pattern: str) -> re.Pattern[str]:
         if part == "#":
             if i != len(parts) - 1:
                 raise ValueError("'#' wildcard is only valid as the last segment")
-            # Use (/.*)?$ so that # matches zero or more trailing segments.
-            # e.g. "a/#" -> "^a(/.*)?$" matches "a", "a/b", "a/b/c"
-            return re.compile("^" + "/".join(regex_parts) + "(/.*)?$")
+            # # matches zero or more trailing segments.
+            # If preceding segments exist, the / before # is optional
+            # so "myapp/#" matches both "myapp" and "myapp/anything".
+            # If # is the sole segment, it matches everything.
+            if regex_parts:
+                return re.compile("^" + "/".join(regex_parts) + "(/.*)?$")
+            else:
+                return re.compile("^.*$")
         elif part == "+":
             regex_parts.append("[^/]+")
         else:
@@ -126,6 +131,7 @@ def __init__(self) -> None:
         self._lock = asyncio.Lock()
         self._store: dict[str, RetainedEvent] = {}
         self._expires: dict[str, str] = {}  # topic -> ISO 8601 expires_at
+        self._regex_cache: dict[str, re.Pattern[str]] = {}
 
     async def set(self, topic: str, event: RetainedEvent, expires_at: str | None = None) -> None:
         """Store or replace the retained value for *topic*."""
@@ -151,7 +157,9 @@ async def get(self, topic: str) -> RetainedEvent | None:
     async def get_matching(self, pattern: str) -> list[RetainedEvent]:
         """Return all non-expired retained events whose topic matches *pattern*."""
         async with self._lock:
-            regex = _pattern_to_regex(pattern)
+            if pattern not in self._regex_cache:
+                self._regex_cache[pattern] = _pattern_to_regex(pattern)
+            regex = self._regex_cache[pattern]
             result: list[RetainedEvent] = []
             expired_topics: list[str] = []
             for topic, event in self._store.items():

src/mcp/server/session.py

@@ -44,6 +44,7 @@ async def handle_list_prompts(ctx: RequestContext) -> list[types.Prompt]:
 import anyio.lowlevel
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from pydantic import AnyUrl
+from ulid import ULID
 
 import mcp.types as types
 from mcp.server.experimental.session_features import ExperimentalServerSessionFeatures
@@ -218,8 +219,6 @@ async def emit_event(
     ) -> None:
         """Push an event to the client on the given topic."""
         if event_id is None:
-            from ulid import ULID
-
             event_id = str(ULID())
         if timestamp is None:
             from datetime import datetime, timezone

tests/test_subscription_registry.py

@@ -109,6 +109,13 @@ async def test_rejects_pattern_exceeding_max_depth(self, registry: SubscriptionR
         with pytest.raises(ValueError, match="exceeds maximum depth of 8 segments"):
             await registry.add("s1", "a/b/c/d/e/f/g/h/i")
 
+    async def test_hash_root_wildcard_matches_everything(self, registry: SubscriptionRegistry):
+        """Pattern '#' (sole segment) should match any topic."""
+        await registry.add("s1", "#")
+        assert await registry.match("any/topic/at/all") == {"s1"}
+        assert await registry.match("single") == {"s1"}
+        assert await registry.match("a/b") == {"s1"}
+
     async def test_hash_matches_zero_trailing_no_slash(self, registry: SubscriptionRegistry):
         """# should match the prefix with no trailing slash (zero segments after prefix).