Serialize subscription setup and keep event cursors safe to persist

Author: Kludex

docs/advanced/events.md

@@ -121,6 +121,7 @@ fell out of the upstream's retention window. Clients use it to know they missed
 from pydantic import BaseModel
 
 from mcp.experimental.events.server import Events, Subscriber
+from mcp.experimental.events.types import FORBIDDEN, events_error
 from mcp.server.mcpserver import MCPServer
 
 
@@ -140,6 +141,8 @@ watched: dict[str, int] = {}
 
 @incidents.on_subscribe
 async def start_watching(subscriber: Subscriber[IncidentFilter]) -> None:
+    if subscriber.principal is None:
+        raise events_error(FORBIDDEN, reason="sign in to watch incidents")
     service = subscriber.arguments.service
     watched[service] = watched.get(service, 0) + 1
 
@@ -152,7 +155,11 @@ async def stop_watching(subscriber: Subscriber[IncidentFilter]) -> None:
 mcp = MCPServer("pager", extensions=[events])
 ```
 
-Hooks let you set up and tear down upstream work per subscription. A poll client never says
+Hooks let you set up and tear down upstream work per subscription. `on_subscribe` is also where
+you check access: the subscriber carries the caller's `principal`, and raising an `MCPError`
+rejects the poll or subscription with that error.
+
+A poll client never says
 goodbye, so the SDK leases poll subscriptions: `on_subscribe` runs on the first poll for a
 principal, event type, and arguments, and `on_unsubscribe` runs once no poll renewed the lease
 for `poll_lease` (three poll intervals by default).
@@ -221,7 +228,8 @@ anyio.run(main)
 
 `watch` polls in a loop. It waits as long as the server asks (`nextPollMs`), never less than
 `Events(min_poll_interval=...)`, and polls again at once while the server has more. It yields
-only batches that have events or report a gap (`batch.truncated`).
+every batch, including empty ones, because an empty batch still moves the cursor. A batch with
+`batch.truncated` set means events were skipped.
 
 To resume after a restart, save `batch.cursor` once you handled a batch, and pass it back with
 `watch(..., cursor=saved)`. Delivery is at least once, so deduplicate on `event_id`.
@@ -273,7 +281,8 @@ anyio.run(main)
 ```
 
 `subscribe` holds the subscription for the duration of the block. It refreshes it before it
-expires, and unsubscribes on exit.
+expires, passing the latest cursor so a server that lost the subscription replays from there,
+and unsubscribes on exit.
 
 `WebhookReceiver` is an ASGI app for the callback URL. It answers the verification challenge,
 rejects requests that are unsigned, signed with another secret, or more than five minutes old,

src/mcp/experimental/events/_locks.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+import anyio
+
+
+class KeyedLocks:
+    """One lock per key, dropped once nobody holds or awaits it."""
+
+    def __init__(self) -> None:
+        self._locks: dict[str, anyio.Lock] = {}
+
+    @asynccontextmanager
+    async def __call__(self, key: str) -> AsyncIterator[None]:
+        lock = self._locks.setdefault(key, anyio.Lock())
+        try:
+            async with lock:
+                yield
+        finally:
+            if not lock.locked() and not lock.statistics().tasks_waiting:
+                del self._locks[key]

src/mcp/experimental/events/client.py

@@ -101,18 +101,18 @@ async def poll(
     async def watch(
         self, name: str, arguments: dict[str, Any] | None = None, *, cursor: str | None = None
     ) -> AsyncGenerator[PollEventsResult, None]:
-        """Poll forever, yielding each batch that has events or reports a gap (`truncated`).
+        """Poll forever, yielding every batch, including empty ones.
 
         Persist `batch.cursor` after handling a batch, and pass it back after a restart
-        to resume: delivery is at-least-once, so deduplicate on `event_id`.
+        to resume: delivery is at-least-once, so deduplicate on `event_id`. Empty batches
+        still move the cursor, which keeps it inside the upstream's retention window.
         """
         if cursor is None:
             cursor = (await self.poll(name, arguments)).cursor
         while True:
             batch = await self.poll(name, arguments, cursor=cursor)
             cursor = batch.cursor
-            if batch.events or batch.truncated:
-                yield batch
+            yield batch
             if not batch.has_more:
                 wait = max(timedelta(milliseconds=batch.next_poll_ms), self._min_poll_interval)
                 await anyio.sleep(wait.total_seconds())
@@ -145,7 +145,7 @@ async def subscribe(
         )
         result = await self._session.send_request(SubscribeEventsRequest(params=params), SubscribeEventsResult)
         async with anyio.create_task_group() as task_group:
-            task_group.start_soon(self._refresh, params, result.refresh_before)
+            task_group.start_soon(self._refresh, params, result)
             try:
                 yield result
             finally:
@@ -155,11 +155,10 @@ async def subscribe(
                     request = UnsubscribeEventsRequest(params=unsubscribe)
                     await self._session.send_request(request, UnsubscribeEventsResult)
 
-    async def _refresh(self, params: SubscribeEventsParams, refresh_before: datetime | None) -> None:
-        while refresh_before is not None:
+    async def _refresh(self, params: SubscribeEventsParams, result: SubscribeEventsResult) -> None:
+        while result.refresh_before is not None:
             # Refresh at two thirds of the remaining lifetime, leaving room for a failed attempt.
-            await anyio.sleep((refresh_before - datetime.now(UTC)).total_seconds() * 2 / 3)
-            # Resend the original cursor-less params: the server keeps the subscription's own position.
-            request = SubscribeEventsRequest(params=params.model_copy(update={"cursor": None}))
+            await anyio.sleep((result.refresh_before - datetime.now(UTC)).total_seconds() * 2 / 3)
+            # The latest safe cursor is a no-op for a live subscription, and the replay point if the server lost it.
+            request = SubscribeEventsRequest(params=params.model_copy(update={"cursor": result.cursor}))
             result = await self._session.send_request(request, SubscribeEventsResult)
-            refresh_before = result.refresh_before

src/mcp/experimental/events/receiver.py

@@ -53,8 +53,8 @@ class WebhookReceiver:
     """ASGI app for the callback URL of webhook subscriptions made with one `secret`.
 
     Requests that are unsigned, signed with another secret, or older than `tolerance`
-    get a 401 and never reach `handler`. Events are deduplicated on `event_id` across
-    the last `dedupe_window` events, since servers retry deliveries.
+    get a 401 and never reach `handler`. Events are deduplicated on their subscription and
+    `event_id` across the last `dedupe_window` events, since servers retry deliveries.
     """
 
     def __init__(
@@ -69,7 +69,7 @@ def __init__(
         self._handler = handler
         self._tolerance = tolerance
         self._dedupe_window = dedupe_window
-        self._seen: OrderedDict[str, None] = OrderedDict()
+        self._seen: OrderedDict[tuple[str, str], None] = OrderedDict()
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         request = Request(scope, receive)
@@ -100,11 +100,13 @@ async def _handle(self, request: Request) -> Response:
                     delivery = EventOccurrence.model_validate(payload)
         except (ValueError, TypeError, KeyError, AttributeError, ValidationError):
             return Response(status_code=400)
-        if isinstance(delivery, EventOccurrence) and delivery.event_id in self._seen:
+        # One upstream event can match several subscriptions, and each must receive it.
+        seen_key = (subscription_id, delivery.event_id) if isinstance(delivery, EventOccurrence) else None
+        if seen_key in self._seen:
             return Response(status_code=204)
         await self._handler(delivery)
-        if isinstance(delivery, EventOccurrence):
-            self._seen[delivery.event_id] = None
+        if seen_key is not None:
+            self._seen[seen_key] = None
             if len(self._seen) > self._dedupe_window:
                 self._seen.popitem(last=False)
         return Response(status_code=204)

src/mcp/experimental/events/server.py

@@ -31,6 +31,7 @@ def same_service(arguments: IncidentFilter, incident: Incident) -> bool:
 from pydantic import BaseModel, ValidationError
 
 from mcp.experimental.events._buffer import EmitBuffer
+from mcp.experimental.events._locks import KeyedLocks
 from mcp.experimental.events.event_type import EventBatch, EventQuery, EventType, NoArguments, Occurrence, Subscriber
 from mcp.experimental.events.types import (
     EXTENSION_ID,
@@ -117,6 +118,7 @@ def __init__(
         self._buffer_max_age = buffer_max_age
         self._event_types: dict[str, EventType[Any, Any]] = {}
         self._leases: dict[str, anyio.CancelScope] = {}
+        self._lease_setup = KeyedLocks()
         self._task_group: TaskGroup | None = None
         self._subscriptions: WebhookSubscriptions | None = None
 
@@ -221,20 +223,17 @@ def _resolve(self, name: str, arguments: dict[str, Any] | None) -> tuple[EventTy
     async def _renew_lease(self, event_type: EventType[Any, Any], arguments: BaseModel, principal: str | None) -> None:
         key = json.dumps([principal, event_type.name, arguments.model_dump(mode="json")], sort_keys=True)
         deadline = anyio.current_time() + self._poll_lease.total_seconds()
-        if (lease := self._leases.get(key)) is not None:
-            lease.deadline = deadline
-            return
         assert self._task_group is not None, "Events runs inside the server's lifespan"
         subscription_id = f"sub_{hashlib.sha256(key.encode()).hexdigest()[:16]}"
         subscriber = Subscriber(id=subscription_id, arguments=arguments, principal=principal, mode="poll")
-        lease = self._leases[key] = anyio.CancelScope(deadline=deadline)
-        try:
+        async with self._lease_setup(key):
+            if (lease := self._leases.get(key)) is not None:
+                lease.deadline = deadline
+                return
             for hook in event_type.subscribe_hooks:
                 await hook(subscriber)
-        except BaseException:
-            del self._leases[key]
-            raise
-        self._task_group.start_soon(self._expire_lease, key, lease, event_type, subscriber)
+            lease = self._leases[key] = anyio.CancelScope(deadline=deadline)
+            self._task_group.start_soon(self._expire_lease, key, lease, event_type, subscriber)
 
     async def _expire_lease(
         self, key: str, lease: anyio.CancelScope, event_type: EventType[Any, Any], subscriber: Subscriber[Any]

src/mcp/experimental/events/webhooks.py

@@ -18,6 +18,7 @@
 from pydantic import BaseModel
 
 from mcp.experimental.events._callbacks import CallbackFailure, Callbacks, NonPublicAddress
+from mcp.experimental.events._locks import KeyedLocks
 from mcp.experimental.events.event_type import EventBatch, EventQuery, EventType, Subscriber
 from mcp.experimental.events.signing import decode_secret
 from mcp.experimental.events.types import (
@@ -103,7 +104,8 @@ def __init__(
         self._callbacks = Callbacks(client, config.trusted_origins)
         self._task_group = task_group
         self._subscriptions: dict[str, _Subscription] = {}
-        self._verified: set[tuple[str, str]] = set()
+        self._verified: set[tuple[str | None, str]] = set()
+        self._setup = KeyedLocks()
 
     async def subscribe(
         self, event_type: EventType[Any, Any], arguments: BaseModel, principal: str, params: SubscribeEventsParams
@@ -117,8 +119,26 @@ async def subscribe(
         key = json.dumps([principal, url, event_type.name, params.arguments or {}], sort_keys=True)
         subscription_id = f"sub_{hashlib.sha256(key.encode()).hexdigest()[:16]}"
         ttl = self._grant(params)
-        if (existing := self._subscriptions.get(key)) is not None:
-            return self._refresh(existing, secret, ttl)
+        async with self._setup(key):
+            if (existing := self._subscriptions.get(key)) is not None:
+                return self._refresh(existing, secret, ttl)
+            try:
+                return await self._create(event_type, arguments, principal, params, key, subscription_id, ttl)
+            except BaseException:
+                self._forget_unused_verification(principal, url)
+                raise
+
+    async def _create(
+        self,
+        event_type: EventType[Any, Any],
+        arguments: BaseModel,
+        principal: str,
+        params: SubscribeEventsParams,
+        key: str,
+        subscription_id: str,
+        ttl: timedelta,
+    ) -> SubscribeEventsResult:
+        url, secret = params.delivery.url, params.delivery.secret
         if (principal, url) not in self._verified:
             try:
                 await self._callbacks.verify(url, subscription_id, secret)
@@ -138,8 +158,8 @@ async def subscribe(
             url=url,
             secret=secret,
             expires_at=datetime.now(UTC) + ttl,
-            # Not yet delivered, so only the supplied cursor is safe to persist; after a gap it is stale.
-            cursor=params.cursor if batch.events and not batch.truncated else batch.cursor,
+            # The batch is not delivered yet, so only the supplied cursor is safe to persist.
+            cursor=params.cursor if batch.events else batch.cursor,
         )
         subscription.scope.deadline = anyio.current_time() + ttl.total_seconds()
         for hook in event_type.subscribe_hooks:
@@ -160,6 +180,11 @@ def unsubscribe(self, event_type: EventType[Any, Any], principal: str, params: U
             raise events_error(NOT_FOUND, kind="subscription")
         subscription.scope.cancel()
 
+    def _forget_unused_verification(self, principal: str | None, url: str) -> None:
+        # The spec scopes verification to the subscriptions it covers, so the cache cannot outgrow them.
+        if not any(s.subscriber.principal == principal and s.url == url for s in self._subscriptions.values()):
+            self._verified.discard((principal, url))
+
     def _grant(self, params: SubscribeEventsParams) -> timedelta:
         # In-memory subscriptions cannot honour "no expiry", so an omitted or null ttlMs gets the cap.
         if params.ttl_ms is None:
@@ -184,6 +209,8 @@ async def _run(self, subscription: _Subscription, batch: EventBatch[Any]) -> Non
         try:
             with subscription.scope:
                 while True:
+                    # shortcut: events go out one at a time, so a failing event delays later ones by its
+                    # retries, where the spec retries each independently; needs per-event acks behind the cursor.
                     for index, occurrence in enumerate(batch.events):
                         last = index == len(batch.events) - 1
                         body = event_type.wire(occurrence, batch.cursor if last else subscription.cursor)
@@ -195,6 +222,7 @@ async def _run(self, subscription: _Subscription, batch: EventBatch[Any]) -> Non
         finally:
             if self._subscriptions.get(subscription.key) is subscription:
                 del self._subscriptions[subscription.key]
+            self._forget_unused_verification(subscriber.principal, subscription.url)
         for hook in event_type.unsubscribe_hooks:
             await hook(subscriber)
 

tests/experimental/events/test_client.py

@@ -35,8 +35,8 @@ class Ping(BaseModel):
 
 
 async def test_watch_yields_batches_with_events_and_resumes_from_the_latest_cursor() -> None:
-    """`watch` polls in a loop, drains `hasMore` batches at once, skips empty polls, and yields
-    batches that carry events or a gap (SDK-defined loop over spec-mandated poll semantics)."""
+    """`watch` polls in a loop and drains `hasMore` batches at once (SDK-defined loop over
+    spec-mandated poll semantics)."""
     events = Events(poll_interval=timedelta(0), max_events=2)
     pings = events.event("ping", arguments=NoArguments, payload=Ping)
     mcp = MCPServer("pinger", extensions=[events])
@@ -73,9 +73,10 @@ async def test_watch_from_a_saved_cursor_yields_the_gap_when_the_cursor_is_unkno
     assert (batch.events, batch.truncated) == ([], True)
 
 
-async def test_watch_waits_at_least_the_clients_floor_between_polls() -> None:
-    """The client's `min_poll_interval` overrides a server asking for a tighter loop (spec-mandated floor;
-    a time-based feature, so the test lets one floor elapse)."""
+async def test_watch_yields_empty_batches_and_waits_at_least_the_clients_floor_between_polls() -> None:
+    """Empty batches are yielded because they still advance the cursor (spec-mandated quiet-period
+    advancement), and the client's `min_poll_interval` overrides a server asking for a tighter loop
+    (spec-mandated floor; a time-based feature, so the test lets the floor elapse)."""
     polls: list[str | None] = []
     events = Events(poll_interval=timedelta(0))
     pings = events.event("ping", arguments=NoArguments, payload=Ping)
@@ -84,19 +85,24 @@ async def test_watch_waits_at_least_the_clients_floor_between_polls() -> None:
     async def check(query: EventQuery[NoArguments]) -> EventBatch[Ping]:
         polls.append(query.cursor)
         if len(polls) < 3:
-            return EventBatch(cursor="c")
-        return EventBatch(cursor="c", events=[Occurrence(Ping(n=len(polls)))])
+            return EventBatch(cursor=f"c{len(polls)}")
+        return EventBatch(cursor=f"c{len(polls)}", events=[Occurrence(Ping(n=len(polls)))])
 
     mcp = MCPServer("pinger", extensions=[events])
 
     async with Client(mcp, extensions=[ClientEvents(min_poll_interval=timedelta(milliseconds=20))]) as client:
         watch = client.extension(ClientEvents).watch("ping")
+        started = anyio.current_time()
         with anyio.fail_after(5):
-            batch = await anext(watch)
+            quiet = await anext(watch)
+            busy = await anext(watch)
             await watch.aclose()
+        elapsed = anyio.current_time() - started
 
-    assert polls == [None, "c", "c"]
-    assert [e.data for e in batch.events] == [{"n": 3}]
+    assert polls == [None, "c1", "c2"]
+    assert (quiet.events, quiet.cursor) == ([], "c2")
+    assert [e.data for e in busy.events] == [{"n": 3}]
+    assert elapsed >= 0.02
 
 
 class _OtherEventsServer(Extension):
@@ -127,7 +133,8 @@ async def _subscribe(
     ) -> SubscribeEventsResult:
         self.subscribes.append(params)
         if len(self.subscribes) == 1:
-            return SubscribeEventsResult(id="sub-1", refresh_before=datetime.now(UTC) + timedelta(milliseconds=30))
+            refresh_before = datetime.now(UTC) + timedelta(milliseconds=30)
+            return SubscribeEventsResult(id="sub-1", refresh_before=refresh_before, cursor="c1")
         self.refreshed.set()
         return SubscribeEventsResult(id="sub-1", refresh_before=None)
 
@@ -149,8 +156,9 @@ async def test_list_follows_pagination_to_the_last_page() -> None:
 
 
 async def test_subscribe_refreshes_before_the_subscription_expires_and_unsubscribes_on_exit() -> None:
-    """`subscribe` resends the subscription before `refreshBefore`, stops once the server grants no
-    expiry, and unsubscribes when the block exits (spec-mandated refresh; a time-based feature)."""
+    """`subscribe` resends the subscription with the latest cursor before `refreshBefore`, stops once
+    the server grants no expiry, and unsubscribes when the block exits (spec-mandated refresh; a
+    time-based feature)."""
     server = _OtherEventsServer()
     mcp = MCPServer("other", extensions=[server])
     secret = "whsec_" + "a" * 43 + "="
@@ -164,5 +172,5 @@ async def test_subscribe_refreshes_before_the_subscription_expires_and_unsubscri
             await server.unsubscribed.wait()
 
     first, refresh = server.subscribes
-    assert (first.cursor, refresh.cursor) == ("c0", None)
+    assert (first.cursor, refresh.cursor) == ("c0", "c1")
     assert refresh.delivery == first.delivery