Repository navigation
Conversation
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
📚 Documentation preview
|
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: ad16a1cb32
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| self._subscriptions[key] = subscription | ||
| self._task_group.start_soon(self._run, subscription, batch) |
There was a problem hiding this comment.
When two concurrent subscribe requests use the same principal, URL, event type, and arguments, both can pass the existing-subscription check before either finishes endpoint verification or the initial read. Both then start delivery loops, but the second assignment overwrites the first subscription in _subscriptions. Subsequent refreshes and unsubscribe calls affect only the retained subscription, leaving the other loop delivering duplicate events until its original TTL expires. Serialize creation per key so overlapping requests refresh one subscription rather than create orphaned loops.
Useful? React with 👍 / 👎.
| oldest = self._entries[0][0] if self._entries else self._head + 1 | ||
| truncated = position < oldest - 1 |
There was a problem hiding this comment.
If every event after a reader's cursor has expired, eviction empties _entries, but position remains the supplied cursor because the iteration never runs. The response therefore returns that same stale cursor with truncated=True, and polling with the returned cursor reports the identical gap indefinitely until another event arrives. This also makes watch() repeatedly yield the same gap and webhook delivery repeatedly send gap messages. Advance the position past the discarded entries even when there are no retained events.
Useful? React with 👍 / 👎.
| finally: | ||
| if self._subscriptions.get(subscription.key) is subscription: | ||
| del self._subscriptions[subscription.key] | ||
| for hook in event_type.unsubscribe_hooks: | ||
| await hook(subscriber) |
There was a problem hiding this comment.
When the server lifespan cancels its task group, cancellation propagates through the subscription's own cancel scope rather than being suppressed by it. The finally block removes the subscription, but execution skips the unsubscribe hooks below it. Consequently, upstream listeners allocated by on_subscribe are not torn down when a running server is stopped or restarted within the same process. Put hook cleanup in the cancellation path and shield asynchronous teardown; poll leases in _expire_lease have the same shutdown problem.
Useful? React with 👍 / 👎.
| assert polls == [None, "c", "c"] | ||
| assert [e.data for e in batch.events] == [{"n": 3}] |
There was a problem hiding this comment.
This test passes unchanged if watch() stops sleeping entirely or ignores min_poll_interval: its assertions only check the three scripted polls and final payload. It therefore does not protect the tight-loop prevention behavior named in the test. The test-quality guide required by AGENTS.md explicitly requires an assertion proving the claimed behavior; use deterministic clock or sleep instrumentation to verify that the 20 ms floor is honored without adding a wall-clock timing assertion.
AGENTS.md reference: AGENTS.md:L68-L69
Useful? React with 👍 / 👎.
| assert timedelta(minutes=59) < short.refresh_before - before <= timedelta(hours=1, minutes=1) | ||
| assert timedelta(hours=1, minutes=59) < long.refresh_before - before <= timedelta(hours=2, minutes=1) |
There was a problem hiding this comment.
These assertions measure expiration against an earlier real-time reading and allow only a one-minute margin, so process suspension, a sufficiently delayed request, or a system-clock adjustment can fail the test despite correct TTL clamping. The test-quality guide required by AGENTS.md explicitly forbids wall-clock assertions even with generous margins. Control the clock used for expiration calculation and assert the exact granted TTLs instead.
AGENTS.md reference: AGENTS.md:L68-L69
Useful? React with 👍 / 👎.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
21 issues found across 23 files
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="tests/experimental/events/test_client.py">
<violation number="1" location="tests/experimental/events/test_client.py:162">
P2: The fake server accepts refreshes unconditionally, so a client that refreshes after `refresh_before` still passes this test. Record the refresh arrival and assert it precedes the first response's deadline, or reject late refreshes in the fake server.</violation>
</file>
<file name="src/mcp/experimental/events/signing.py">
<violation number="1" location="src/mcp/experimental/events/signing.py:48">
P2: A non-ASCII `v1,` candidate makes `hmac.compare_digest` raise `TypeError`, so the receiver returns an unhandled error instead of rejecting the invalid signature. Ignore non-ASCII candidates before comparing them.</violation>
</file>
<file name="src/mcp/experimental/events/event_type.py">
<violation number="1" location="src/mcp/experimental/events/event_type.py:162">
P1: This makes every non-replayable check source permanently silent: `cursor=None` is sent on every poll, so this branch discards every batch's events. Track initialization separately from cursor availability so later polls can deliver newly found events.</violation>
</file>
<file name="src/mcp/experimental/events/_callbacks.py">
<violation number="1" location="src/mcp/experimental/events/_callbacks.py:75">
P2: `parts.port or 443` silently maps an explicit `:0` callback to port 443, while an out-of-range port raises `ValueError` before the `try` and escapes verification as an internal error. Validate the port in `validate()` and default only when it is `None`.</violation>
<violation number="2" location="src/mcp/experimental/events/_callbacks.py:121">
P1: On a NAT64-enabled host, `64:ff9b::a00:1` passes this check but translates to private IPv4 `10.0.0.1`, bypassing the private-address refusal. Reject NAT64/other IPv4-translation addresses or validate their embedded IPv4 destination.</violation>
</file>
<file name="src/mcp/server/mcpserver/server.py">
<violation number="1" location="src/mcp/server/mcpserver/server.py:287">
P2: This wraps extension lifespans around each SSE connection because `handle_sse` calls `_lowlevel_server.run()` per client. `Events.lifespan()` shares one task group across connections, so the first disconnect cancels webhook delivery for the others; run extension lifespans once for the transport app or otherwise keep the shared lifetime open until all connections exit.</violation>
</file>
<file name="src/mcp/experimental/events/receiver.py">
<violation number="1" location="src/mcp/experimental/events/receiver.py:80">
P1: The receiver buffers the full POST body before validating its signature, and this app imposes no size limit. Enforce a bounded read and return 413 so unauthenticated requests cannot exhaust the process's memory.</violation>
<violation number="2" location="src/mcp/experimental/events/receiver.py:84">
P2: `.isdigit()` accepts superscript digits such as `²`, but `int()` rejects them; this uncaught conversion turns an invalid request into a 500. Restrict timestamps to ASCII decimal and catch conversion errors so malformed headers return 401.</violation>
<violation number="3" location="src/mcp/experimental/events/receiver.py:103">
P2: Concurrent retries for the same event can both pass this check before either records the ID, because `_handler` is awaited first. Track in-flight IDs atomically and clear them if the handler fails.</violation>
</file>
<file name="src/mcp/experimental/events/server.py">
<violation number="1" location="src/mcp/experimental/events/server.py:167">
P2: Graceful shutdown cancels `_expire_lease` while it waits in `sleep_forever`; cancellation escapes `with lease`, so `on_unsubscribe` never runs. Run lease teardown in a shielded cleanup path before clearing leases.</violation>
<violation number="2" location="src/mcp/experimental/events/server.py:188">
P2: A check function cannot request an immediate next poll: `timedelta(0)` is falsy, so this returns `poll_interval` instead of the override. Use the default only when `next_poll is None`.</violation>
<violation number="3" location="src/mcp/experimental/events/server.py:245">
P1: An exception from an `on_unsubscribe` hook escapes this task-group child and cancels the server lifespan. Catch and log teardown failures so one subscriber cannot stop the server.</violation>
</file>
<file name="docs/advanced/events.md">
<violation number="1" location="docs/advanced/events.md:62">
P2: The default emit buffer holds only 1,000 events and 10 minutes, so a burst can evict unread events even when polling every few seconds. Qualify this guarantee and tell readers to handle `batch.truncated`.</violation>
<violation number="2" location="docs/advanced/events.md:288">
P2: Deduplication covers only the last 1,000 IDs in memory; eviction or a receiver restart lets retried deliveries reach the handler again. Document the limit and require durable handler-side idempotency for duplicate-sensitive work.</violation>
</file>
<file name="src/mcp/experimental/events/webhooks.py">
<violation number="1" location="src/mcp/experimental/events/webhooks.py:142">
P1: An initial truncated batch sets `subscription.cursor` to its final position before delivery. Every non-final event then advertises a cursor that can skip remaining events if the receiver resumes after processing the first; retain a pre-batch checkpoint until delivery completes.</violation>
<violation number="2" location="src/mcp/experimental/events/webhooks.py:146">
P2: An `on_unsubscribe` exception escapes this background task and fails the shared task group, cancelling unrelated webhook loops. Catch and log each hook failure so one subscriber's cleanup cannot stop other deliveries.</violation>
<violation number="3" location="src/mcp/experimental/events/webhooks.py:227">
P2: The non-public-address and HTTP 410/413 paths set `last_error` but return before recording `failed_since`, so refresh reports an ongoing failure with no start time. Set `failed_since` before these early returns.</violation>
</file>
<file name="src/mcp/experimental/events/_buffer.py">
<violation number="1" location="src/mcp/experimental/events/_buffer.py:54">
P2: This removes only a stale prefix, but `EventType.emit` accepts caller-supplied timestamps; an old occurrence behind a fresh one remains readable beyond `buffer_max_age`. Filter expired entries throughout the read and report skipped sequences as truncated.</violation>
<violation number="2" location="src/mcp/experimental/events/_buffer.py:60">
P2: When eviction removes every retained entry after a cursor, this reports `truncated` but leaves `position` unchanged because the loop is empty. Advance it to `oldest - 1` so clients can resume instead of receiving the same gap on every poll.</violation>
<violation number="3" location="src/mcp/experimental/events/_buffer.py:80">
P2: A suffix such as `²` passes `str.isdigit()` but `int()` raises `ValueError`, so a malformed poll cursor causes an error instead of unknown-cursor recovery. Accept ASCII decimal digits and handle conversion failures.</violation>
</file>
<file name="src/mcp/experimental/events/client.py">
<violation number="1" location="src/mcp/experimental/events/client.py:164">
P2: A failed refresh request escapes this task and cancels the surrounding `subscribe` block, so one transient transport or RPC error interrupts the caller and ends the subscription. Catch refresh failures and retry with bounded backoff before the lease expires.</violation>
</file>
Tip: cubic can generate docs of your entire codebase and keep them up to date. Try it here.
Turn on auto-fix | Re-trigger cubic
| ) | ||
| batch = await self._check(query) | ||
| # A null cursor means "start from now", whatever the check function returned. | ||
| return replace(batch, events=()) if query.cursor is None else batch |
There was a problem hiding this comment.
P1: This makes every non-replayable check source permanently silent: cursor=None is sent on every poll, so this branch discards every batch's events. Track initialization separately from cursor availability so later polls can deliver newly found events.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At src/mcp/experimental/events/event_type.py, line 162:
<comment>This makes every non-replayable check source permanently silent: `cursor=None` is sent on every poll, so this branch discards every batch's events. Track initialization separately from cursor availability so later polls can deliver newly found events.</comment>
<file context>
@@ -0,0 +1,185 @@
+ )
+ batch = await self._check(query)
+ # A null cursor means "start from now", whatever the check function returned.
+ return replace(batch, events=()) if query.cursor is None else batch
+
+ def descriptor(self, delivery: list[Literal["poll", "push", "webhook"]]) -> EventDescriptor:
</file context>
| except OSError: | ||
| raise CallbackFailure("connection_refused") from None | ||
| addresses = [str(info[4][0]) for info in infos] | ||
| if not all(ipaddress.ip_address(address).is_global for address in addresses): |
There was a problem hiding this comment.
P1: On a NAT64-enabled host, 64:ff9b::a00:1 passes this check but translates to private IPv4 10.0.0.1, bypassing the private-address refusal. Reject NAT64/other IPv4-translation addresses or validate their embedded IPv4 destination.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At src/mcp/experimental/events/_callbacks.py, line 121:
<comment>On a NAT64-enabled host, `64:ff9b::a00:1` passes this check but translates to private IPv4 `10.0.0.1`, bypassing the private-address refusal. Reject NAT64/other IPv4-translation addresses or validate their embedded IPv4 destination.</comment>
<file context>
@@ -0,0 +1,132 @@
+ except OSError:
+ raise CallbackFailure("connection_refused") from None
+ addresses = [str(info[4][0]) for info in infos]
+ if not all(ipaddress.ip_address(address).is_global for address in addresses):
+ raise NonPublicAddress(f"{host} resolves to a non-public address")
+ return addresses[0]
</file context>
| await response(scope, receive, send) | ||
|
|
||
| async def _handle(self, request: Request) -> Response: | ||
| body = await request.body() |
There was a problem hiding this comment.
P1: The receiver buffers the full POST body before validating its signature, and this app imposes no size limit. Enforce a bounded read and return 413 so unauthenticated requests cannot exhaust the process's memory.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At src/mcp/experimental/events/receiver.py, line 80:
<comment>The receiver buffers the full POST body before validating its signature, and this app imposes no size limit. Enforce a bounded read and return 413 so unauthenticated requests cannot exhaust the process's memory.</comment>
<file context>
@@ -0,0 +1,110 @@
+ await response(scope, receive, send)
+
+ async def _handle(self, request: Request) -> Response:
+ body = await request.body()
+ message_id = request.headers.get("webhook-id", "")
+ timestamp = request.headers.get("webhook-timestamp", "")
</file context>
| with lease: | ||
| await anyio.sleep_forever() | ||
| del self._leases[key] | ||
| for hook in event_type.unsubscribe_hooks: |
There was a problem hiding this comment.
P1: An exception from an on_unsubscribe hook escapes this task-group child and cancels the server lifespan. Catch and log teardown failures so one subscriber cannot stop the server.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At src/mcp/experimental/events/server.py, line 245:
<comment>An exception from an `on_unsubscribe` hook escapes this task-group child and cancels the server lifespan. Catch and log teardown failures so one subscriber cannot stop the server.</comment>
<file context>
@@ -0,0 +1,246 @@
+ with lease:
+ await anyio.sleep_forever()
+ del self._leases[key]
+ for hook in event_type.unsubscribe_hooks:
+ await hook(subscriber)
</file context>
| select: Callable[[PayloadT], PayloadT | None], | ||
| ) -> EventBatch[PayloadT]: | ||
| now = datetime.now(UTC) | ||
| while self._entries and self._entries[0][1].timestamp < now - self._max_age: |
There was a problem hiding this comment.
P2: This removes only a stale prefix, but EventType.emit accepts caller-supplied timestamps; an old occurrence behind a fresh one remains readable beyond buffer_max_age. Filter expired entries throughout the read and report skipped sequences as truncated.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At src/mcp/experimental/events/_buffer.py, line 54:
<comment>This removes only a stale prefix, but `EventType.emit` accepts caller-supplied timestamps; an old occurrence behind a fresh one remains readable beyond `buffer_max_age`. Filter expired entries throughout the read and report skipped sequences as truncated.</comment>
<file context>
@@ -0,0 +1,82 @@
+ select: Callable[[PayloadT], PayloadT | None],
+ ) -> EventBatch[PayloadT]:
+ now = datetime.now(UTC)
+ while self._entries and self._entries[0][1].timestamp < now - self._max_age:
+ self._entries.popleft()
+ position = self._position(cursor)
</file context>
| if position is None: | ||
| return EventBatch(cursor=f"{self._epoch}.{self._head}", truncated=cursor is not None) | ||
| oldest = self._entries[0][0] if self._entries else self._head + 1 | ||
| truncated = position < oldest - 1 |
There was a problem hiding this comment.
P2: When eviction removes every retained entry after a cursor, this reports truncated but leaves position unchanged because the loop is empty. Advance it to oldest - 1 so clients can resume instead of receiving the same gap on every poll.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At src/mcp/experimental/events/_buffer.py, line 60:
<comment>When eviction removes every retained entry after a cursor, this reports `truncated` but leaves `position` unchanged because the loop is empty. Advance it to `oldest - 1` so clients can resume instead of receiving the same gap on every poll.</comment>
<file context>
@@ -0,0 +1,82 @@
+ if position is None:
+ return EventBatch(cursor=f"{self._epoch}.{self._head}", truncated=cursor is not None)
+ oldest = self._entries[0][0] if self._entries else self._head + 1
+ truncated = position < oldest - 1
+ floor = now - max_age if max_age is not None else None
+ events: list[Occurrence[PayloadT]] = []
</file context>
| truncated = position < oldest - 1 | |
| truncated = position < oldest - 1 | |
| if truncated: | |
| position = oldest - 1 |
| 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})) | ||
| result = await self._session.send_request(request, SubscribeEventsResult) |
There was a problem hiding this comment.
P2: A failed refresh request escapes this task and cancels the surrounding subscribe block, so one transient transport or RPC error interrupts the caller and ends the subscription. Catch refresh failures and retry with bounded backoff before the lease expires.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At src/mcp/experimental/events/client.py, line 164:
<comment>A failed refresh request escapes this task and cancels the surrounding `subscribe` block, so one transient transport or RPC error interrupts the caller and ends the subscription. Catch refresh failures and retry with bounded backoff before the lease expires.</comment>
<file context>
@@ -0,0 +1,165 @@
+ 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}))
+ result = await self._session.send_request(request, SubscribeEventsResult)
+ refresh_before = result.refresh_before
</file context>
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| except httpx2.TimeoutException: | ||
| raise CallbackFailure("timeout") from None | ||
| except httpx2.TransportError as exc: |
There was a problem hiding this comment.
🔴 A subscriber's callback endpoint can stop the whole server by answering one delivery with a body that fails to decode. Callbacks.post catches only httpx2.TransportError (_callbacks.py:86), and httpx2.DecodingError is not one. It escapes _deliver and _run into the extension's shared task group. Fix: turn every httpx request error from the POST into a CallbackFailure, for example by catching httpx2.HTTPError, so a bad response counts as a failed delivery and is retried. This differs from guarding on_unsubscribe hooks: the trigger here is remote and client-controlled. [also at: src/mcp/experimental/events/_callbacks.py:87 - A webhook subscriber can shut down event delivery for everyone by answering a delivery with an undecodable body. Callbacks.post catches only…]
Why this was flagged
An authenticated client subscribes with its own https endpoint, which echoes the challenge (webhooks.py:124), then answers a later event POST with Content-Encoding: gzip and a body that is not gzip. self._client.post (_callbacks.py:81) reads and decodes the body and raises httpx2.DecodingError, a RequestError that is not a TransportError, so the handlers at _callbacks.py:84-87 do not catch it. Neither do except NonPublicAddress and except CallbackFailure in _deliver (webhooks.py:225-229). The exception leaves _run (webhooks.py:182), which runs in the task group created at server.py:161, so every other subscription and poll lease is cancelled and the server lifespan fails. The base branch sends no callback requests. The except Exception at webhooks.py:209 wraps only event_type.read, not _deliver. On events/subscribe the same response makes verify fail with an internal error instead of CALLBACK_ENDPOINT_ERROR. The exception hierarchy is stated from the httpx API that httpx2 mirrors; httpx2 is not installed in this checkout.
Verification: Triggered when a verified subscriber's callback endpoint answers an event POST with a body that fails content decoding. src/mcp/experimental/events/_callbacks.py:80-87: Callbacks.post wraps self._client.post(...) in exactly two handlers, except httpx2.TimeoutException and except httpx2.TransportError. One unhandled child exception cancels the Events task group.
Finding ID: f-d209ba94
| for hook in event_type.unsubscribe_hooks: | ||
| await hook(subscriber) |
There was a problem hiding this comment.
🔴 One failing on_unsubscribe hook shuts down the whole MCP server, not just the subscription it belongs to. Background tasks run user hooks with no exception isolation, inside the task group that Events.lifespan holds around the server. An uncaught error there cancels the lifespan body. Fix: catch and log failures at the top of each per-subscription task so only that subscription ends, which covers the 2 sites listed. Same pattern at 2 sites (webhooks.py:199, server.py:246). _read already isolates the check function for this reason. [also at: src/mcp/experimental/events/server.py:246 - A failing on_unsubscribe hook stops all event background work and can bring the server down. _expire_lease (server.py:245) and _run (webhooks.py:198) await user hooks directly inside the extension's task group.]
Why this was flagged
A webhook subscription's TTL lapses or the client calls events/unsubscribe, so _run leaves with subscription.scope and awaits hook(subscriber) at webhooks.py:198-199. A poll lease expiring does the same in _expire_lease at server.py:245-246. If the user's hook raises (an upstream teardown call failing, a KeyError on shared state), the exception leaves the task started with self._task_group.start_soon. That group is entered in Events.lifespan (server.py:161), which _with_extension_lifespans holds around the server lifespan, so anyio cancels the serving code and the lifespan exits with an ExceptionGroup. All clients lose the server. The comment at webhooks.py:210 says a crash "must not take the server's task group down", but that except Exception only wraps event_type.read. Any non-TransportError raised by self._client.post inside _deliver (webhooks.py:222) escapes the same way. The base branch has no extension background tasks.
Verification: Triggers when a user-registered on_unsubscribe hook raises as a webhook subscription ends or a poll lease lapses. src/mcp/experimental/events/webhooks.py:198-199 and src/mcp/experimental/events/server.py:242-246 await the hook bare inside the task group entered in Events.lifespan at server.py:161, so the serve loop is cancelled. No guard found elsewhere.
Finding ID: f-743adcc1
| if self._task_group is not None: | ||
| # shortcut: a nested run (an in-memory Client per connection) shares the outer one's | ||
| # background work, which stops when the outer run does; give each server its own Events. | ||
| yield | ||
| return |
There was a problem hiding this comment.
🔴 On SSE, one client disconnecting silently drops every other client's webhook subscriptions and poll leases, and their next subscribe call fails. Events.lifespan lets only the first entrant own the task group (server.py:151-155). When that connection ends, the finally cancels the group and sets _subscriptions to None while other connections are still open. Fix: tear the background work down only when the last entrant exits (for example a reference count), which covers SSE, stdio and in-memory runs. Server.run enters the lifespan once per connection there.
Why this was flagged
MCPServer.sse_app calls self._lowlevel_server.run(...) per SSE connection (src/mcp/server/mcpserver/server.py:1194), and Server.run does async with self.lifespan(self) (src/mcp/server/lowlevel/server.py:722), now wrapped by _with_extension_lifespans. Client A connects first and creates the task group; client B connects and takes the yield; return shortcut. When A disconnects, server.py:167-169 cancels the group, sets _task_group = self._subscriptions = None and clears _leases. B's delivery loops stop with no terminated body and no on_unsubscribe hook. B's next events/subscribe or events/unsubscribe hits assert self._subscriptions is not None (server.py:199, 208) and hooked polls hit the assert at server.py:227. The comment at server.py:152 treats this as an in-memory Client shortcut only; nothing guards SSE. Streamable HTTP enters the lifespan once (streamable_http_manager.py:156) and is unaffected. The base branch has no Events extension.
Verification: Triggering condition: an MCPServer with the Events extension served over SSE with two or more overlapping connections, where the first-connected one disconnects first. The owner's finally (server.py:167-169) runs self._task_group.cancel_scope.cancel(), self._task_group = self._subscriptions = None, self._leases.clear(), with no check for other live entrants. B's next events/subscribe or events/unsubscribe hits assert self._subscriptions is not None (server.py:199, 208).
Finding ID: f-7f565fba
| def access_token_principal(ctx: ServerRequestContext[Any, Any]) -> str | None: | ||
| """The caller's OAuth identity (client, issuer, subject), or `None` on an unauthenticated server.""" | ||
| token = get_access_token() | ||
| return compact_json(principal_components(token)) if token is not None else None |
There was a problem hiding this comment.
🔴 Users who share one OAuth client can cancel or take over each other's webhook subscriptions when the token verifier sets no subject. access_token_principal (server.py:80) builds the principal from client, issuer and subject, and AccessToken.subject defaults to None (src/mcp/server/auth/provider.py:59). All such tokens then map to one principal, which is the only owner check in the subscription key (webhooks.py:117). Fix: answer Forbidden for webhook calls whose token has no subject unless a custom principal= resolver is set, and document it at docs/advanced/events.md:173.
Why this was flagged
If a deployment's verifier does not populate AccessToken.subject (default None at src/mcp/server/auth/provider.py:59; nothing under src/mcp sets it), two users holding tokens for the same client_id reach events/subscribe and events/unsubscribe (server.py:197, server.py:206) with the same principal string. User B subscribing with A's url, name and arguments hits the existing key at webhooks.py:120, and _refresh replaces A's secret at webhooks.py:172, so after the 5 minute _ROTATION_GRACE A's receiver rejects every delivery. B's events/unsubscribe pops A's subscription at webhooks.py:158. The base branch has no events extension, so this is new surface. The principal is None Forbidden guard does not help: the JSON triple is never None for an authenticated token. docs/advanced/events.md:173-177 says the default principal prevents exactly this cancel or hijack. Remedy: treat a subject-less token as no principal for webhook calls.
Verification: Triggers when the deployment's token verifier leaves AccessToken.subject unset and more than one user holds tokens for the same client_id. The is None guards at server.py:197 and server.py:206 therefore do not fire. src/mcp/experimental/events/webhooks.py:117: the principal is the only ownership component. webhooks.py:158-161 pops A's subscription and cancels its scope.
Finding ID: f-0980fba1
| return await self._client.post( | ||
| url, content=content, headers=headers, extensions=extensions, follow_redirects=False | ||
| ) |
There was a problem hiding this comment.
🔴 A hostile callback endpoint can exhaust server memory or hold an events/subscribe request open with no end. self._client.post at _callbacks.py:81 buffers and decompresses the whole response body. There is no size cap, and the only timeout is a per-read 5 seconds. Fix: bound every callback response in size and in total time, for example stream it under anyio.fail_after, read only the status for deliveries, and cap the verification body at a few kilobytes.
Why this was flagged
Trigger: any principal allowed to call events/subscribe names an endpoint that answers with an endless or very large body, or a small gzip body that expands to gigabytes. Callbacks.post calls self._client.post(...) (_callbacks.py:81-83), which returns only after the full body is read and decoded; verify then calls response.json() on it (_callbacks.py:104). The default client is httpx2.AsyncClient(timeout=5) (events/server.py:160), a per-operation timeout, so a body that keeps sending a chunk every few seconds never times out. In the subscribe path (webhooks.py:124) no deadline applies at all. In the delivery path (webhooks.py:222) only the subscription TTL scope applies, and the client can refresh it. The server gets unbounded memory growth and stuck handlers; the base branch makes no outbound calls. docs/advanced/events.md:180 says callback URLs are treated as hostile, but only the request side (scheme, address, redirects) is guarded, not the response.
Verification: Triggers when a principal allowed to call events/subscribe names a callback endpoint that answers with an endless, very large, or highly compressed body. src/mcp/experimental/events/_callbacks.py:81-83 is a non-streaming call, so the whole body is read and content-decoded before it returns. There is no byte cap, and httpx2.AsyncClient(timeout=5) at src/mcp/experimental/events/server.py:160 is not a total deadline.
Finding ID: f-a7cff6c3
| url = urlunsplit(parts._replace(netloc=f"{host}:{parts.port or 443}")) | ||
| headers["host"] = parts.netloc | ||
| extensions["sni_hostname"] = parts.hostname |
There was a problem hiding this comment.
🔴 Events for one callback host can be sent over a TLS connection opened and certificate-checked for a different host on the same IP. Callbacks.post rewrites the URL to the pinned IP (_callbacks.py:77) and sets sni_hostname per request. All subscriptions share one client, whose pool matches connections by scheme, IP and port if httpx2 pools like httpx. Fix: never reuse a pinned connection across hostnames, e.g. send Connection: close on pinned POSTs or keep one pool per callback host. Shared front ends may then answer 421, and the receiver's certificate goes unchecked.
Why this was flagged
Condition: httpx2/httpcore2 pick a pooled connection by URL origin only and read sni_hostname only when opening a connection, as httpx/httpcore do (the library source is not in this checkout). Two subscriptions use https://a.example/hook and https://b.example/hook, and both names resolve to one address, as tenants of the same CDN or hosting platform do. _public_address returns that IP for both (_callbacks.py:75), the request URL becomes https://<ip>:443/hook (_callbacks.py:77), and the single client built at events/server.py:160 keeps the connection alive. A POST for b inside the keep-alive window reuses the connection whose handshake used SNI and certificate checks for a. b's event bodies are sent without checking b's certificate, and front ends that reject an SNI/Host mismatch answer 421, which _deliver records as http_4xx and retries (webhooks.py:236). The module docstring promises "the original Host and TLS SNI" (_callbacks.py:5-6). The sender is new, so the base has no such path.
Verification: When two callback hosts resolve to the same IP, src/mcp/experimental/events/_callbacks.py:74-79 replaces the netloc with the resolved IP, so the request origin is https://<ip>:<port> for every host on that IP. _callbacks.py:81-83: the request goes through the single self._client, with no Connection: close. The only test uses a MockTransport, so pooling is never exercised.
Finding ID: f-7436d314
| ) | ||
| batch = await self._check(query) | ||
| # A null cursor means "start from now", whatever the check function returned. | ||
| return replace(batch, events=()) if query.cursor is None else batch |
There was a problem hiding this comment.
🔴 A check function that returns cursor=None, as the EventBatch docstring allows for non-replayable event types, never delivers any event by poll or webhook. EventType.read strips the events whenever query.cursor is None (event_type.py:162). With a None cursor returned, every later read also has a None cursor. Fix: make a non-replayable check type able to deliver after its first read, for example strip events only on the caller's initial read, or remove the cursor=None option from the EventBatch docstring and reject it.
Why this was flagged
A server registers @ event.check returning EventBatch(cursor=None, events=[...]), following event_type.py:58 "Return cursor=None for an event type that cannot replay". EventsClient.watch first polls with no cursor and stores batch.cursor, which is None (client.py:110), then polls with cursor=None forever. Each read reaches replace(batch, events=()) if query.cursor is None else batch and returns no events. Webhooks behave the same: subscribe stores cursor=batch.cursor (None) at webhooks.py:142, and _read builds its EventQuery from subscription.cursor (webhooks.py:202), so every batch is emptied. The client sees an endless stream of empty polls and no error. No guard exists; test_poll_answers_from_the_check_function_with_the_callers_query only covers a non-null cursor. The base branch has no Events extension.
Verification: When a server author registers a check function that returns EventBatch(cursor=None, ...), no event is ever delivered to a reader that started without a cursor. src/mcp/experimental/events/event_type.py:160-162: Events are dropped on every read whose query cursor is None, not only on the first one. Nothing guards this: no validation rejects a None cursor from a check function, and no test covers it.
Finding ID: f-c064a5ff
| def append(self, occurrence: Occurrence[PayloadT]) -> None: | ||
| self._head += 1 | ||
| self._entries.append((self._head, occurrence)) | ||
| self._appended.set() | ||
| self._appended = anyio.Event() |
There was a problem hiding this comment.
🔴 Following the docs' own example, events emitted from a sync tool reach webhooks late and can make a poll fail. EventType.emit mutates the buffer and calls anyio.Event.set() with no thread hand-off (_buffer.py:34-38). Sync tools run on a worker thread (func_metadata.py:176), and docs/advanced/events.md:52-54 emits from one. Fix: make emit safe from worker threads, for example by moving the append and the wake-up onto the event loop, or document it as loop-only and make the docs example async def.
Why this was flagged
open_incident in docs/advanced/events.md:52-54 is a plain def tool, so call_fn runs it with anyio.to_thread.run_sync (src/mcp/server/mcpserver/utilities/func_metadata.py:176). incidents.emit then calls EmitBuffer.append (src/mcp/experimental/events/_buffer.py:34-38) off the event-loop thread. self._appended.set() is not thread-safe: on asyncio the waiter in EmitBuffer.wait (_buffer.py:42-43) is scheduled without waking the loop, so the webhook loop at webhooks.py:193 may only notice at its poll_interval timeout (5 seconds by default) instead of at once. The deque append can also land while read iterates self._entries on the loop thread (_buffer.py:64), which raises RuntimeError: deque mutated during iteration and fails that events/poll. Nothing in the package takes a lock or uses anyio.from_thread. This is new code, so the base branch has no such path.
Verification: Triggers when emit is called from a sync (def) tool, which is exactly what the new docs example does. src/mcp/server/mcpserver/utilities/func_metadata.py:176 runs a sync function with anyio.to_thread.run_sync(...). _buffer.py:34-38: append does self._entries.append(...) and self._appended.set() on the calling thread. A worker-thread append during the loop at _buffer.py:64 raises RuntimeError: deque mutated during iteration. The tests only emit from the loop thread.
Finding ID: f-e256d621
| position = self._position(cursor) | ||
| if position is None: | ||
| return EventBatch(cursor=f"{self._epoch}.{self._head}", truncated=cursor is not None) | ||
| oldest = self._entries[0][0] if self._entries else self._head + 1 | ||
| truncated = position < oldest - 1 |
There was a problem hiding this comment.
🔴 Consumers that resume from a cursor whose events have all aged out get a truncated gap on every poll instead of once. EmitBuffer.read flags the gap at _buffer.py:60 but, when pruning left the buffer empty, returns the caller's stale position, so the next read is truncated again. Fix: once a gap is detected, return a cursor at or past oldest - 1 (e.g. position = max(position, oldest - 1)), so the follow-up read is clean whether or not entries remain. Until the next emit, watch yields a gap batch per poll and a webhook loop POSTs gap every poll interval.
Why this was flagged
A client resumes with a saved cursor <epoch>.5 (the flow documented at docs/advanced/events.md:226) after every later event passed buffer_max_age; EventType.read calls EmitBuffer.read (event_type.py:154). The prune loop at _buffer.py:54-55 empties the deque, oldest becomes self._head + 1 (_buffer.py:59), truncated is True (_buffer.py:60), the scan loop has no entry to advance position, and _buffer.py:76 returns cursor=f"{self._epoch}.{position}", the same stale cursor. watch stores it and yields the batch because batch.truncated is set (client.py:113-115), so it yields a gap every next_poll_ms; a webhook _read POSTs {"type": "gap"} each interval (webhooks.py:213-214). This is new code, so the base has no such behaviour. The unknown-cursor path does return a fresh cursor (_buffer.py:58) and tests/experimental/events/test_server.py:186 pins a clean follow-up poll there; the aged-out path (test_server.py:225-237) never polls again, so nothing catches it.
Verification: Triggers when a reader presents a same-epoch cursor behind _head after every later entry has aged past buffer_max_age. In src/mcp/experimental/events/_buffer.py, line 60 sets truncated True and line 76 returns the caller's own stale cursor. The next read with that returned cursor repeats the same result. No test covers the empty-after-prune case, and no guard exists on either path.
Finding ID: f-89b372c9
There was a problem hiding this comment.
Findings marked 🟡 are optional suggestions and need no follow-up push.
Still open from earlier reviews (9):
- 🔴
src/mcp/experimental/events/_buffer.py:38—Following the docs' own example, events emitted from a sync tool reach webhooks late and can make a poll fail. EventTyp… - 🔴
src/mcp/experimental/events/_buffer.py:60—Consumers that resume from a cursor whose events have all aged out get a truncated gap on every poll instead of once. E… - 🔴
src/mcp/experimental/events/_callbacks.py:79—Events for one callback host can be sent over a TLS connection opened and certificate-checked for a different host on t… - 🔴
src/mcp/experimental/events/_callbacks.py:83—A hostile callback endpoint can exhaust server memory or hold an events/subscribe request open with no end. self._clien… - 🔴
src/mcp/experimental/events/_callbacks.py:86—A subscriber's callback endpoint can stop the whole server by answering one delivery with a body that fails to decode.… - 🔴
src/mcp/experimental/events/event_type.py:162—A check function that returns cursor=None, as the EventBatch docstring allows for non-replayable event types, never del… - 🔴
src/mcp/experimental/events/server.py:81—Users who share one OAuth client can cancel or take over each other's webhook subscriptions when the token verifier set… - 🔴
src/mcp/experimental/events/server.py:157—On SSE, one client disconnecting silently drops every other client's webhook subscriptions and poll leases, and their n… - 🔴
src/mcp/experimental/events/webhooks.py:227—One failing on_unsubscribe hook shuts down the whole MCP server, not just the subscription it belongs to. Background ta…
If you have decided not to act on one of these findings, resolve its thread (a reply alone leaves it open) and the next review stops counting it. To review this commit again now, use Re-run on its "Claude Code Review" check.
| 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. |
There was a problem hiding this comment.
🔴 Server authors who follow the new docs get an access check that runs once, so a principal whose access is later revoked keeps receiving events. docs/advanced/events.md:158 says on_subscribe is where to check access. A webhook refresh returns through _refresh (webhooks.py:124) and a renewed poll lease returns at server.py:232, both without running the hook. Fix: re-check access on every refresh and lease renewal, or give emit-fed types a per-read principal check and say in the docs that the hook is subscribe-time only. match gets no principal, so emit-fed types have no other gate. [also at: src/mcp/experimental/events/server.py:232 - A caller whose access is revoked keeps receiving events for as long as it keeps polling or refreshing. The docs now name on_subscribe as the place to check access (docs/advanced/events.md:158-160), but the hook runs only when a lease or subscription is created.]
Why this was flagged
A server registers an emit-fed event type and checks subscriber.principal in on_subscribe, as docs/advanced/events.md:142-145 and 158-160 show. A poll client keeps polling: _renew_lease finds the lease, sets lease.deadline = deadline and returns (server.py:230-232), so the hook ran only on the first poll. A webhook client refreshes before refresh_before: subscribe returns self._refresh(existing, secret, ttl) (webhooks.py:123-124), which extends the deadline and runs no hook. The operator then revokes that principal's access. Polls and deliveries continue for as long as the client keeps polling or refreshing; max_ttl does not bound this because each refresh re-arms it (webhooks.py:198-199). _select calls self._match(arguments, payload) with no principal (event_type.py:182-185), so an emit-fed type has no other place to refuse. The base branch has no Events extension, and before this push the docs did not present the hook as the access check.
Verification: Triggers when a server author puts a revocable access check in on_subscribe of an emit-fed event type, as the docs direct, and that access is later revoked while the caller's token stays valid. The hooks at server.py:233-234 run only when no lease exists. _refresh (webhooks.py:194-205) only rotates the secret and extends expires_at and scope.deadline; no hook runs.
Finding ID: f-4d2fed41
| 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((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) |
There was a problem hiding this comment.
🔴 Webhook subscribers silently miss events when a refresh re-creates a lost subscription, despite the new "replays from there" promise. _refresh keeps only refresh_before and cursor from each refresh result (client.py:164), so truncated=True is discarded. The server reports that gap only in the subscribe result (webhooks.py:173), never as a gap delivery. Fix: make a gap found while creating a subscription reach the receiver, e.g. POST a gap body from _create when batch.truncated, so refresh-created subscriptions surface it too. Emit-based event types always hit this after a server restart (_buffer.py:58).
Why this was flagged
A client holds events.subscribe(...) on an emit-based event type and the server loses the subscription: it restarts, or the TTL lapses after a late refresh. The background _refresh resends events/subscribe with cursor=result.cursor (client.py:163). The server has no entry for the key, so subscribe calls _create (webhooks.py:126). EmitBuffer.read sees a cursor from another epoch and returns EventBatch(cursor=..., truncated=cursor is not None) with no events (_buffer.py:58), or a truncated batch when the entries aged out (_buffer.py:60). _create puts truncated=batch.truncated only in the SubscribeEventsResult (webhooks.py:169-174) and starts _run, which sends no gap body for that first batch; _read (webhooks.py:241-242) only covers later reads. Back in the client, result is used only for refresh_before and cursor. The handler behind WebhookReceiver never gets a WebhookGap, and the caller of subscribe only ever saw the first result. docs/advanced/events.md:283-285 says the server "replays from there"; no safeguard surfaces the gap.
Verification: Triggers when a background refresh reaches a server that no longer holds the subscription and the cursor is from another epoch or already evicted from the buffer. src/mcp/experimental/events/client.py:158-164: result.truncated is never read. The only gap POST is in _read (webhooks.py:241-242), never for the creation batch. The receiver's handler never gets a WebhookGap.
Finding ID: f-90f3cb42
| # 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 |
There was a problem hiding this comment.
🔴 Someone who captures one signed webhook delivery can make the receiver hand the same event to the handler again. The new dedupe key at receiver.py:104 includes subscription_id, which is read from the x-mcp-subscription-id header. The signature does not cover that header. Replaying the request within tolerance with a different header value misses _seen. Fix: build the dedupe key only from signed input, for example also record the signed webhook-id and drop repeats, while still delivering one event_id once per real subscription. Before this push the key was event_id alone, taken from the signed body.
Why this was flagged
An attacker who has observed one valid event POST (a proxy log, a compromised hop) resends it inside the 5 minute tolerance with the same webhook-id, webhook-timestamp, webhook-signature and body, but a different x-mcp-subscription-id value. verify at receiver.py:86 passes, because sign covers only f"{message_id}.{timestamp}." plus the body (signing.py:37). subscription_id comes from the unsigned header at receiver.py:88, so seen_key at receiver.py:104 is a new tuple and self._handler(delivery) runs again at receiver.py:107. Each new header value gives another handler call for the same event. On the commit reviewed before this push, _seen was keyed on delivery.event_id from the signed body, so the replay got a 204 without reaching the handler. The timestamp check at receiver.py:84 does not stop it within the tolerance window, and nothing records webhook-id.
Verification: Triggers when a captured, signed event POST is resent within tolerance with a different x-mcp-subscription-id header. receiver.py:104 builds seen_key = (subscription_id, delivery.event_id), so a replay with an altered header misses _seen and reaches self._handler(delivery). The key changed from delivery.event_id (taken from the signed body). No guard elsewhere tracks webhook-id.
Finding ID: f-4fa1284a
| if (principal, url) not in self._verified: | ||
| try: | ||
| await self._callbacks.verify(url, subscription_id, secret) |
There was a problem hiding this comment.
🔴 A caller that on_subscribe rejects can still make the server POST to a URL it picks and run the check function, once per events/subscribe call. _create verifies the callback (webhooks.py:144) and calls event_type.read (webhooks.py:151) before the hooks at webhooks.py:165, which the docs now name as the access check. Fix: reject a denied principal before any outbound request or upstream read on the webhook path, as _poll already does by running the hooks before read; if the hooks move ahead of verification, a later failure must also run their on_unsubscribe.
Why this was flagged
An authenticated principal that the server's on_subscribe hook denies, the pattern docs/advanced/events.md:158-160 and tests/experimental/events/test_webhooks.py:282 establish, calls events/subscribe through Events._subscribe (server.py:202). _create first awaits self._callbacks.verify(url, subscription_id, secret) (webhooks.py:144), a signed POST to the caller's URL, then event_type.read(EventQuery(arguments, params.cursor, principal, max_age)) (webhooks.py:151), and only then runs event_type.subscribe_hooks (webhooks.py:165-166), where FORBIDDEN is raised. The except BaseException at webhooks.py:127-128 then drops the _verified entry, so every repeat call sends another challenge POST and another check-function call for arguments the caller may not see. The base branch has no Events extension. The only earlier guard is principal is None at server.py:199, which an authenticated but denied caller passes.
Verification: In src/mcp/experimental/events/webhooks.py _create, line 144 sends a signed POST to the caller-supplied URL, line 151 runs the user's check function, and the hooks at lines 165-166 run only afterwards. Nothing earlier rejects a hook-denied principal. The file is new in this diff, so the base has no such path.
Finding ID: f-80218156
| started = anyio.current_time() | ||
| with anyio.fail_after(5): | ||
| quiet = await anext(watch) | ||
| busy = await anext(watch) | ||
| await watch.aclose() | ||
| elapsed = anyio.current_time() - started | ||
|
|
||
| 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 |
There was a problem hiding this comment.
🟡 (optional) This test asserts wall-clock time, which the test-quality rules forbid, and it can fail on Windows CI runners with Python 3.11 or 3.12. assert elapsed >= 0.02 at test_client.py:105 measures a single 20 ms anyio.sleep. asyncio runs a timer up to one clock resolution early, about 15.6 ms on those runners, so elapsed can be 0.0156. Fix: remove the started/elapsed measurement and prove the floor without reading a clock, since SKILL.md says "Never assert wall-clock time, even with huge margins".
Why this was flagged
The trigger is the test itself, on a loop whose clock resolution is coarse; Windows with Python 3.11 or 3.12 reports about 0.0156 s for time.monotonic, and AGENTS.md says CI runs 3.11-3.15 on ubuntu and windows. watch sleeps once for min_poll_interval of 20 ms between the two yielded batches (src/mcp/experimental/events/client.py:117-118). started is taken at tests/experimental/events/test_client.py:95 and assert elapsed >= 0.02 runs at line 105. asyncio's _run_once fires handles whose deadline is within time() + _clock_resolution, so with tick-quantised time the sleep can return when the loop clock has advanced one 15.6 ms tick. The assertion then fails although watch behaved correctly. Before this push the test had no timing assertion. .claude/skills/test-quality/SKILL.md:73-75 states "Never assert wall-clock time, even with huge margins" and that fail_after is "the only timing primitive allowed". The Windows resolution figure is from CPython behaviour, not verified in this checkout.
Verification: nit — triggered whenever the test runs (rule violation); the intermittent failure needs a Windows 3.11/3.12 runner with a coarse monotonic clock. .claude/skills/test-quality/SKILL.md:73-75 reads "Never assert wall-clock time, even with huge margins." Violation, in a file this diff adds (tests/experimental/events/test_client.py): :105 assert elapsed >= 0.02.
Finding ID: f-835544de
There was a problem hiding this comment.
2 issues found across 10 files (changes from recent commits).
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="src/mcp/experimental/events/webhooks.py">
<violation number="1" location="src/mcp/experimental/events/webhooks.py:186">
P2: `_forget_unused_verification` can evict the cache while another `_create` has already skipped verification but is still awaiting `event_type.read`. That leaves an active subscription uncached, so a later subscription unexpectedly repeats endpoint verification; synchronize eviction with in-flight setup.</violation>
</file>
<file name="docs/advanced/events.md">
<violation number="1" location="docs/advanced/events.md:284">
P2: This promises replay after a lost subscription, but `emit` event history is process-local: after a restart, the old cursor produces an empty truncated batch. Qualify replay as dependent on the event source retaining that cursor.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Turn on auto-fix | Re-trigger cubic
| 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)) |
There was a problem hiding this comment.
P2: _forget_unused_verification can evict the cache while another _create has already skipped verification but is still awaiting event_type.read. That leaves an active subscription uncached, so a later subscription unexpectedly repeats endpoint verification; synchronize eviction with in-flight setup.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At src/mcp/experimental/events/webhooks.py, line 186:
<comment>`_forget_unused_verification` can evict the cache while another `_create` has already skipped verification but is still awaiting `event_type.read`. That leaves an active subscription uncached, so a later subscription unexpectedly repeats endpoint verification; synchronize eviction with in-flight setup.</comment>
<file context>
@@ -160,6 +180,11 @@ def unsubscribe(self, event_type: EventType[Any, Any], principal: str, params: U
+ 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:
</file context>
| ``` | ||
|
|
||
| `subscribe` holds the subscription for the duration of the block. It refreshes it before it | ||
| expires, passing the latest cursor so a server that lost the subscription replays from there, |
There was a problem hiding this comment.
P2: This promises replay after a lost subscription, but emit event history is process-local: after a restart, the old cursor produces an empty truncated batch. Qualify replay as dependent on the event source retaining that cursor.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At docs/advanced/events.md, line 284:
<comment>This promises replay after a lost subscription, but `emit` event history is process-local: after a restart, the old cursor produces an empty truncated batch. Qualify replay as dependent on the event source retaining that cursor.</comment>
<file context>
@@ -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.
</file context>
| expires, passing the latest cursor so a server that lost the subscription replays from there, | |
| expires, passing the latest cursor so a server can replay from it when its event source still retains that cursor, |
|
|
||
| ## Offering events | ||
|
|
||
| ```python title="server.py" |
There was a problem hiding this comment.
could this be a tutorial file instead
| When the upstream can answer "what happened since this cursor", register a check function | ||
| instead of calling `emit`: | ||
|
|
||
| ```python title="server.py" |
There was a problem hiding this comment.
same here and all other code snippets
|
I'm building Resource Events, an extension on SEP-3415 that lets a client watch a collection of resources, and I'd like to build its reference implementation on this PR. Its event type doesn't replay: a client that misses an event recovers with a catch-up of its own. So the proposal asks that each delivery carry
Would a way to mark an emit-fed type as not replaying fit, so its deliveries and subscribe results carry |
Implements the poll and webhook delivery modes of SEP-3415 (MCP Events) under
mcp.experimental.events, on both the server and the client. Push (events/stream) is not included.Eventsextension servingevents/list,events/poll,events/subscribeandevents/unsubscribe. Event types are fed either byemit(bounded in-memory buffer) or by acheckfunction over an upstream cursor, withmatch/transformand leasedon_subscribe/on_unsubscribehooks.gap/terminatedcontrol bodies, and SSRF protection (https only, public addresses only, pinned IP, no redirects).watchandpollfor polling,subscribecontext manager that refreshes and unsubscribes, and aWebhookReceiverASGI app.Extension.lifespan(), entered around the server lifespan, to host background tasks.The new public API is experimental and may change until the SEP is accepted. Docs are in
docs/advanced/events.md.AI Disclaimer
This PR was developed with the assistance of either Claude or Codex. I've reviewed and verified the changes.