From cad54402a0b173e646b584da3763cfbe501f8a37 Mon Sep 17 00:00:00 2001
From: Max Isbey <224885523+maxisbey@users.noreply.github.com>
Date: Fri, 24 Jul 2026 17:15:40 +0000
Subject: [PATCH] Make custom notifications first-class: typed send, observable
 drops

Widen ClientSession.send_notification and ServerSession.send_notification
to accept any Notification subclass, mirroring send_request's open Request
arm, so extensions can emit their own methods without casts. Raise
dropped-notification logging from debug to a once-per-method warning that
names the remedy: on the client when no NotificationBinding observes the
method, on the server when a spec notification is not defined at the
negotiated version or a custom method has no registered handler (an
unserved spec notification stays at debug; repeat drops of a method log at
debug so a stream cannot flood the log). Document the message_handler
routing change and the binding channel in the migration guide, fix the
callbacks page's catch-all claim, and cover the vendor-notification round
trip with an interaction test, un-deferring its requirement.
---
 docs/advanced/extensions.md                   |  6 +-
 docs/advanced/low-level-server.md             | 27 ++++++++-
 docs/client/callbacks.md                      |  2 +-
 docs/migration.md                             | 30 ++++++++++
 docs/whats-new.md                             |  1 +
 src/mcp/client/session.py                     | 15 ++++-
 src/mcp/server/runner.py                      | 23 ++++++--
 src/mcp/server/session.py                     | 13 +++-
 tests/client/test_session.py                  | 55 ++++++++++++-----
 .../test_session_notification_bindings.py     | 17 ++++--
 tests/interaction/_requirements.py            |  9 +--
 .../interaction/mcpserver/test_extensions.py  | 59 ++++++++++++++++++-
 tests/server/test_runner.py                   | 41 +++++++++----
 13 files changed, 247 insertions(+), 51 deletions(-)

diff --git a/docs/advanced/extensions.md b/docs/advanced/extensions.md
index 715c5bbdc8..05d3035e82 100644
--- a/docs/advanced/extensions.md
+++ b/docs/advanced/extensions.md
@@ -214,7 +214,11 @@ def notifications(self) -> Sequence[NotificationBinding[Any]]:
 ```
 
 The handler receives validated params one at a time, in dispatch order. It observes; it cannot veto
-or reply.
+or reply. The emitting half needs no registration: the server side of your extension defines the
+notification as a `mcp_types.Notification` subclass and sends it with
+`ctx.session.send_notification(..., related_request_id=ctx.request_id)`, as shown in
+**[The low-level Server](low-level-server.md#a-method-of-your-own)**. A client without the binding
+drops the notification with a warning.
 
 Two quiet rules. Claims are active on 2026-07-28 connections only, and the capability
 ad follows them: on a legacy connection the claims dissolve and the identifier drops
diff --git a/docs/advanced/low-level-server.md b/docs/advanced/low-level-server.md
index dd5fb427a2..922b9aeac4 100644
--- a/docs/advanced/low-level-server.md
+++ b/docs/advanced/low-level-server.md
@@ -162,10 +162,33 @@ The constructor covers the methods MCP defines. `add_request_handler` covers eve
 --8<-- "docs_src/lowlevel/tutorial006.py"
 ```
 
-* The first argument is the method string. Notifications have a twin, `add_notification_handler`.
+* The first argument is the method string.
 * `params_type` is the model the incoming `params` are validated against **before** your handler runs, so custom methods *do* get the validation tools don't. Subclass `RequestParams` so the `_meta` field parses like every other method's.
 * The handler returns a `BaseModel`, a `dict`, or `None`. The SDK serialises it into the JSON-RPC result.
 
+Notifications have a twin in each direction. Inbound, `add_notification_handler(method, params_type, handler)` registers `async (ctx, params) -> None`. Spec-defined notification methods are validated against the negotiated version's tables before dispatch, and one that does not exist at that version is dropped with a warning. Custom methods skip the version gate; your `params_type` is their validation, and a custom notification nobody registered a handler for is dropped with a warning.
+
+Outbound, `ctx.session.send_notification(...)` accepts any `mcp_types.Notification` subclass, not just the spec-defined union, so the notifying half of a vendor protocol is one model away:
+
+```python
+class ReindexProgressParams(NotificationParams):
+    percent: float
+
+
+class ReindexProgress(Notification[ReindexProgressParams, Literal["notifications/reindex/progress"]]):
+    method: Literal["notifications/reindex/progress"] = "notifications/reindex/progress"
+    params: ReindexProgressParams
+
+
+async def reindex(ctx: ServerRequestContext, params: ReindexParams) -> None:
+    await ctx.session.send_notification(
+        ReindexProgress(params=ReindexProgressParams(percent=40.0)),
+        related_request_id=ctx.request_id,
+    )
+```
+
+Pass `related_request_id` so the notification rides the originating request's stream; the 2026-07-28 wire gives standalone notifications no channel outside `subscriptions/listen`. A python-sdk client observes vendor methods with a `NotificationBinding` (**[Extensions](extensions.md)**) and warns about ones it has no binding for.
+
 One honest caveat: the high-level `Client` only has verbs for the methods MCP defines, so there is no `client.reindex()`. A vendor method is for a peer that already knows it exists: a client you also ship, or another service of yours speaking JSON-RPC.
 
 One method you cannot claim:
@@ -196,7 +219,7 @@ Each of these is one idea you now have the vocabulary for; each has its own page
 * An exception in a handler is a `-32603` protocol error. A tool error the model can read is a `CallToolResult` with `is_error=True` that **you** return.
 * `_meta` on the result is addressed to the client application, not the model.
 * `Server[T]` is generic in what its lifespan yields; `ctx.lifespan_context` is a typed `T`.
-* `add_request_handler(method, params_type, handler)` serves any method. `initialize` is reserved.
+* `add_request_handler(method, params_type, handler)` serves any method. `initialize` is reserved. `add_notification_handler` and `ctx.session.send_notification` are the notification twins.
 * The capabilities a `Server` advertises are derived from which handlers you registered.
 
 `Client(server)` treated both servers identically because they *are* the same protocol, which is the whole point. The next layer down isn't a class at all: it's **[Middleware](middleware.md)**.
diff --git a/docs/client/callbacks.md b/docs/client/callbacks.md
index 6b4e934cf9..3339cd96dc 100644
--- a/docs/client/callbacks.md
+++ b/docs/client/callbacks.md
@@ -135,7 +135,7 @@ Two more. Neither declares anything.
 
 `logging_callback` receives every `notifications/message` a server sends, as `LoggingMessageNotificationParams` (`level`, `logger`, `data`). Protocol logging is itself deprecated by the 2026-07-28 spec (**[Logging](../handlers/logging.md)** has what to do instead), so this callback exists for the servers that still emit it.
 
-`message_handler` is the catch-all: every server notification reaches it (as well as its specific callback), and on a stream-backed transport so does every transport-level `Exception`. The one pattern worth knowing is `if isinstance(message, Exception): raise message`, so a broken connection fails loudly instead of vanishing.
+`message_handler` is the catch-all for spec traffic: every notification the negotiated protocol version defines reaches it (as well as its specific callback), and on a stream-backed transport so does every transport-level `Exception`. Vendor and extension methods are outside that set; they are delivered only to a matching `NotificationBinding` (**[Extensions](../advanced/extensions.md)**), and without one they are dropped with a warning. The one pattern worth knowing is `if isinstance(message, Exception): raise message`, so a broken connection fails loudly instead of vanishing.
 
 ## Recap
 
diff --git a/docs/migration.md b/docs/migration.md
index 2f2594c4c9..7246afb769 100644
--- a/docs/migration.md
+++ b/docs/migration.md
@@ -1653,6 +1653,36 @@ Behavior changes:
 
 `mcp.shared.session` is now a compatibility module: `ProgressFnT` is re-exported (its home is `mcp.shared.dispatcher`), and `RequestResponder` remains as a typing-only stub so `MessageHandlerFnT` annotations keep importing. `RequestResponder.respond()` no longer exists, and neither do the cancellation-tracking members (`cancel()`, the `cancelled` and `in_flight` properties, the `on_complete` constructor argument) or `BaseSession._in_flight`; inbound cancellation is handled by `JSONRPCDispatcher`.
 
+### Vendor notifications never reach `message_handler`; observe them with a `NotificationBinding`
+
+In v1 every inbound server notification was validated against the closed `ServerNotification` union. Methods outside the union were dropped with a warning before any callback ran, and `notifications/tasks/status` was a union member, so it did reach `message_handler`. In v2 the notification tables are per protocol version and `TaskStatusNotification` is types-only: a method the negotiated version does not define is delivered only to a matching `NotificationBinding`, and without one it is dropped with a warning naming the method. `message_handler` keeps its typed contract and never sees these.
+
+Code that watched task status (or any vendor method) through `message_handler` registers a binding instead:
+
+```python
+from mcp.client import NotificationBinding
+from mcp_types import TaskStatusNotificationParams
+
+
+async def on_task_status(params: TaskStatusNotificationParams) -> None:
+    ...
+
+
+session = ClientSession(
+    read_stream,
+    write_stream,
+    notification_bindings=[
+        NotificationBinding(
+            method="notifications/tasks/status",
+            params_type=TaskStatusNotificationParams,
+            handler=on_task_status,
+        )
+    ],
+)
+```
+
+On the high-level `Client`, bindings are declared by a `ClientExtension`'s `notifications()`; see [Extensions](advanced/extensions.md). The sending direction also loses its `cast`: `send_notification` on both `ClientSession` and `ServerSession` accepts any `mcp_types.Notification` subclass, where v1 typed it to the closed unions.
+
 ### Experimental Tasks support removed
 
 Tasks ([SEP-1686](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1686)) have been removed from the MCP specification and are no longer part of this SDK. The `mcp.client.experimental`, `mcp.server.experimental`, `mcp.shared.experimental`, and `mcp.server.lowlevel.experimental` modules have been removed, along with the `experimental` properties on `ClientSession`, `ServerSession`, `Server`, and `ServerRequestContext`. The corresponding `Task*` types remain in `mcp_types` as types-only definitions, except the `TaskExecutionMode` alias, whose literal is now inlined on `ToolExecution.task_support`.
diff --git a/docs/whats-new.md b/docs/whats-new.md
index 3f1188f8fc..e2699df375 100644
--- a/docs/whats-new.md
+++ b/docs/whats-new.md
@@ -201,6 +201,7 @@ At 2026-07-28 the standalone HTTP GET stream and `resources/subscribe` are repla
 * **Requests are routable without parsing bodies.** Modern HTTP requests carry `Mcp-Method` (and, for the three tool-ish calls, `Mcp-Name`); a tool input-schema property annotated with `x-mcp-header` is mirrored into an `Mcp-Param-*` header and cross-checked by the server ([SEP-2243](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2243)). Gateways and rate limiters can route on headers alone; the **[Migration Guide](migration.md#servers-validate-mcp-param-headers-against-the-request-body-sep-2243)** has the rules.
 * **Results carry cache hints.** List and read results declare `ttlMs` and `cacheScope` ([SEP-2549](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2549)); you set them per method with `cache_hints=`, and `Client` honors them with a built-in response cache. A server that sends no hints (every pre-2026 server) sees identical, uncached traffic. **[Caching hints](client/caching.md)**.
 * **Extensions are first class.** Servers and clients declare optional capability bundles under reverse-DNS identifiers ([SEP-2133](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2133)); the built-in `Apps` extension (MCP Apps) is the reference. **[Extensions](advanced/extensions.md)** and **[MCP Apps](advanced/apps.md)**.
+* **Vendor notifications are typed end to end.** `send_notification` on both sessions accepts any `mcp_types.Notification` subclass, and a client observes methods outside the negotiated version's tables with a `NotificationBinding` (on `ClientSession` directly, or declared by a `ClientExtension`); a notification nothing observes is dropped with a warning, not silently. **[Extensions](advanced/extensions.md)**.
 * **Error codes got standardized.** A missing resource is `-32602` with the URI in `error.data`, and the new spec-reserved codes appear as `-32020` (header mismatch), `-32021` (missing required capability), and `-32022` (unsupported protocol version). **[Troubleshooting](troubleshooting.md)** is keyed by the exact messages.
 * **Authorization got harder to hold wrong.** The client validates the `iss` returned with the authorization code ([RFC 9207](https://datatracker.ietf.org/doc/html/rfc9207); your `callback_handler` now returns an `AuthorizationCodeResult`), sends `application_type` when it registers, and never replays credentials against a different authorization server. New in the enterprise corner: the [SEP-990](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/990) identity-assertion flow. The **[Migration Guide](migration.md)** lists every OAuth change; **[OAuth for clients](client/oauth-clients.md)** and **[Identity assertion](client/identity-assertion.md)** are the pages.
 * **Every server is traceable.** OpenTelemetry ships on by default as middleware: every request gets a server span, at no cost until the process configures an exporter. When both ends run the SDK, the client also propagates W3C trace context in `_meta`, so the traces join up. **[OpenTelemetry](run/opentelemetry.md)**.
diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py
index 36cb78df5d..d23f4cbba5 100644
--- a/src/mcp/client/session.py
+++ b/src/mcp/client/session.py
@@ -363,6 +363,7 @@ def __init__(
         self._extensions = dict(extensions) if extensions is not None else None
         self._result_claims = _index_claims(result_claims, extensions)
         self._notification_bindings = _index_bindings(notification_bindings)
+        self._warned_notification_drops: set[str] = set()
         self._active_claims: dict[str, ResultClaim[Any]] = {}
         self._call_tool_adapter = _CallToolResultAdapter
         self._binding_queues: dict[
@@ -527,9 +528,14 @@ async def send_request(
             return result_type.validate_python(raw, by_name=False)
         return result_type.model_validate(raw, by_name=False)
 
-    async def send_notification(self, notification: types.ClientNotification) -> None:
+    async def send_notification(self, notification: types.ClientNotification | types.Notification[Any, Any]) -> None:
         """Send a one-way notification. Usable before entering the context manager.
 
+        Spec notifications are the `types.ClientNotification` union members; any
+        other `types.Notification` subclass is sent as-is, so extensions can emit
+        their own methods (a python-sdk server routes those to handlers registered
+        with `Server.add_notification_handler`).
+
         Fire-and-forget: after the connection has closed, the notification is
         dropped with a debug log instead of raising.
         """
@@ -1334,7 +1340,12 @@ async def _on_notify(
             # Only methods unknown to the negotiated version's core tables reach the bindings.
             binding = self._notification_bindings.get(method)
             if binding is None:
-                logger.debug("dropped %r: not defined at %s", method, version)
+                # The first drop of a method warns; repeats log at debug so a stream cannot flood the log.
+                level = logging.WARNING if method not in self._warned_notification_drops else logging.DEBUG
+                self._warned_notification_drops.add(method)
+                logger.log(
+                    level, "dropped %r: not defined at %s and no notification binding is registered", method, version
+                )
                 return
             try:
                 bound_params = binding.params_type.model_validate(params or {})
diff --git a/src/mcp/server/runner.py b/src/mcp/server/runner.py
index 6f9f7a8f74..4d7fecb3d7 100644
--- a/src/mcp/server/runner.py
+++ b/src/mcp/server/runner.py
@@ -17,7 +17,7 @@
 import logging
 from collections.abc import AsyncIterator, Awaitable, Mapping
 from contextlib import asynccontextmanager
-from dataclasses import KW_ONLY, dataclass, replace
+from dataclasses import KW_ONLY, dataclass, field, replace
 from functools import cached_property, partial
 from typing import TYPE_CHECKING, Any, Generic, cast
 
@@ -164,6 +164,14 @@ class ServerRunner(Generic[LifespanT]):
     init_options: InitializationOptions | None = None
     """`InitializeResult` payload. Defaults to `server.create_initialization_options()`."""
 
+    _warned_notification_drops: set[str] = field(default_factory=lambda: set(), init=False, repr=False)
+
+    def _log_dropped_notification(self, method: str, message: str, *args: object) -> None:
+        """The first drop of a method warns; repeats log at debug so a stream cannot flood the log."""
+        level = logging.WARNING if method not in self._warned_notification_drops else logging.DEBUG
+        self._warned_notification_drops.add(method)
+        logger.log(level, message, method, *args)
+
     @cached_property
     def on_request(self) -> OnRequest:
         return self._on_request
@@ -251,11 +259,12 @@ async def _on_notify(
 
         async def _inner(ctx: ServerRequestContext[LifespanT, Any]) -> None:
             method, params = ctx.method, ctx.params
-            if method in _methods.SPEC_CLIENT_NOTIFICATION_METHODS:
+            is_spec = method in _methods.SPEC_CLIENT_NOTIFICATION_METHODS
+            if is_spec:
                 try:
                     _methods.validate_client_notification(method, version, params)
                 except KeyError:
-                    logger.debug("dropped %r: not defined at %s", method, version)
+                    self._log_dropped_notification(method, "dropped %r: not defined at %s", version)
                     return
                 except ValidationError:
                     logger.warning("dropped %r: malformed params", method)
@@ -270,7 +279,13 @@ async def _inner(ctx: ServerRequestContext[LifespanT, Any]) -> None:
                 return
             entry = self.server.get_notification_handler(method)
             if entry is None:
-                logger.debug("no handler for notification %s", method)
+                if is_spec:
+                    # Not serving a spec notification is ordinary (most servers
+                    # ignore roots/list_changed); dropping a custom method the
+                    # peer went out of its way to send deserves a warning.
+                    logger.debug("no handler for notification %s", method)
+                else:
+                    self._log_dropped_notification(method, "dropped %r: no notification handler is registered")
                 return
             # Same absent-params contract as requests.
             try:
diff --git a/src/mcp/server/session.py b/src/mcp/server/session.py
index 69ad5ecadb..b30fb10dd2 100644
--- a/src/mcp/server/session.py
+++ b/src/mcp/server/session.py
@@ -102,10 +102,19 @@ async def send_request(
 
     async def send_notification(
         self,
-        notification: types.ServerNotification,
+        notification: types.ServerNotification | types.Notification[Any, Any],
         related_request_id: types.RequestId | None = None,
     ) -> None:
-        """Send a typed server-to-client notification."""
+        """Send a typed server-to-client notification.
+
+        Spec notifications are the `types.ServerNotification` union members; any
+        other `types.Notification` subclass is sent as-is, so extensions can emit
+        their own methods (a python-sdk client observes those by registering a
+        `NotificationBinding`). The 2026-07-28 revision gives standalone
+        notifications no channel outside `subscriptions/listen`, so on modern
+        connections pass `related_request_id` to ride the originating request's
+        stream.
+        """
         channel = self._request_outbound if related_request_id is not None else self._connection.outbound
         data = notification.model_dump(by_alias=True, mode="json", exclude_none=True)
         await channel.notify(data["method"], data.get("params"))
diff --git a/tests/client/test_session.py b/tests/client/test_session.py
index 9c935ef18a..364a61370e 100644
--- a/tests/client/test_session.py
+++ b/tests/client/test_session.py
@@ -2,7 +2,7 @@
 
 from collections.abc import AsyncIterator, Mapping
 from contextlib import AsyncExitStack, asynccontextmanager
-from typing import Any, cast
+from typing import Any, Literal, cast
 
 import anyio
 import anyio.abc
@@ -867,7 +867,7 @@ async def test_on_notify_drops_a_server_notification_absent_at_the_negotiated_ve
     caplog: pytest.LogCaptureFixture,
 ):
     """`notifications/elicitation/complete` does not exist at 2025-06-18: it is
-    debug-log-dropped without reaching `message_handler`."""
+    warn-dropped without reaching `message_handler`."""
     seen: list[object] = []
     delivered = anyio.Event()
 
@@ -875,23 +875,25 @@ async def handler(msg: object) -> None:
         seen.append(msg)
         delivered.set()
 
-    with caplog.at_level("DEBUG", logger="client"):
-        async with raw_client_session(message_handler=handler) as (session, to_client, _):
-            _set_negotiated_version(session, "2025-06-18")
-            await to_client.send(
-                SessionMessage(
-                    JSONRPCNotification(
-                        jsonrpc="2.0", method="notifications/elicitation/complete", params={"elicitationId": "e1"}
-                    )
+    async with raw_client_session(message_handler=handler) as (session, to_client, _):
+        _set_negotiated_version(session, "2025-06-18")
+        await to_client.send(
+            SessionMessage(
+                JSONRPCNotification(
+                    jsonrpc="2.0", method="notifications/elicitation/complete", params={"elicitationId": "e1"}
                 )
             )
-            await to_client.send(
-                SessionMessage(JSONRPCNotification(jsonrpc="2.0", method="notifications/tools/list_changed"))
-            )
-            await delivered.wait()
+        )
+        await to_client.send(
+            SessionMessage(JSONRPCNotification(jsonrpc="2.0", method="notifications/tools/list_changed"))
+        )
+        await delivered.wait()
     assert len(seen) == 1
     assert isinstance(seen[0], types.ToolListChangedNotification)
-    assert "dropped 'notifications/elicitation/complete': not defined at 2025-06-18" in caplog.text
+    assert (
+        "dropped 'notifications/elicitation/complete': not defined at 2025-06-18 "
+        "and no notification binding is registered"
+    ) in caplog.text
 
 
 @pytest.mark.anyio
@@ -1473,6 +1475,29 @@ async def test_send_notification_after_close_is_dropped_silently():
             s.close()
 
 
+class _ReceiptEventParams(types.NotificationParams):
+    seq: int
+
+
+class _ReceiptEvent(types.Notification[_ReceiptEventParams, Literal["notifications/com.example/receipt"]]):
+    method: Literal["notifications/com.example/receipt"] = "notifications/com.example/receipt"
+    params: _ReceiptEventParams
+
+
+@pytest.mark.anyio
+async def test_send_notification_accepts_a_custom_notification_model():
+    """A `types.Notification` subclass outside the spec union serializes onto the wire
+    as-is, so extensions can emit their own methods without casts."""
+    async with raw_client_session() as (session, _, recv_from_client):
+        await session.send_notification(_ReceiptEvent(params=_ReceiptEventParams(seq=7)))
+        with anyio.fail_after(5):
+            sent = await recv_from_client.receive()
+    notification = sent.message
+    assert isinstance(notification, JSONRPCNotification)
+    assert notification.method == "notifications/com.example/receipt"
+    assert notification.params == {"seq": 7}
+
+
 # --- discover() ladder ---
 
 
diff --git a/tests/client/test_session_notification_bindings.py b/tests/client/test_session_notification_bindings.py
index 45e09998b8..956ed52b77 100644
--- a/tests/client/test_session_notification_bindings.py
+++ b/tests/client/test_session_notification_bindings.py
@@ -182,22 +182,27 @@ async def on_event(params: _EventParams) -> None:
 
 
 @pytest.mark.anyio
-async def test_unbound_vendor_notification_keeps_the_debug_drop(caplog: pytest.LogCaptureFixture) -> None:
-    """SDK-defined: a vendor method with no binding keeps the debug-log-and-drop behaviour."""
-    caplog.set_level(logging.DEBUG, logger="client")
-
+async def test_unbound_vendor_notification_is_dropped_with_a_warning(caplog: pytest.LogCaptureFixture) -> None:
+    """SDK-defined: a vendor method with no binding is dropped with a warning naming the
+    remedy; repeats of the same method drop to debug so a stream cannot flood the log."""
     client_side, server_side = create_direct_dispatcher_pair()
     binding = NotificationBinding(method=_VENDOR_METHOD, params_type=_EventParams, handler=_noop_handler)
     session = ClientSession(dispatcher=client_side, notification_bindings=[binding])
-    with anyio.fail_after(5):
+    with caplog.at_level(logging.DEBUG, logger="client"), anyio.fail_after(5):
         async with anyio.create_task_group() as tg:
             await tg.start(server_side.run, _server_on_request, _server_on_notify)
             async with session:
                 _adopt_modern(session)
                 await server_side.notify("notifications/vendor/unbound", {"seq": 1})
+                await server_side.notify("notifications/vendor/unbound", {"seq": 2})
             server_side.close()
 
-    assert f"dropped 'notifications/vendor/unbound': not defined at {LATEST_MODERN_VERSION}" in caplog.text
+    expected = (
+        f"dropped 'notifications/vendor/unbound': not defined at {LATEST_MODERN_VERSION} "
+        "and no notification binding is registered"
+    )
+    levels = [r.levelno for r in caplog.records if r.getMessage() == expected]
+    assert levels == [logging.WARNING, logging.DEBUG]
 
 
 @pytest.mark.anyio
diff --git a/tests/interaction/_requirements.py b/tests/interaction/_requirements.py
index a9fa7ea87f..6df0b4c033 100644
--- a/tests/interaction/_requirements.py
+++ b/tests/interaction/_requirements.py
@@ -2535,10 +2535,11 @@ def __post_init__(self) -> None:
             "against the binding's params type and delivered to its handler serially, in dispatch order."
         ),
         added_in="2026-07-28",
-        deferred=(
-            "Covered at session tier by tests/client/test_session_notification_bindings.py: no public "
-            "server-side surface emits vendor-method notifications (ServerNotification is a closed union), "
-            "and HTTP-modern arrival additionally needs the subscriptions/listen client runtime."
+        note=(
+            "The server emits with related_request_id so the notification rides the originating "
+            "request's stream; a standalone vendor notification has no modern channel outside "
+            "subscriptions/listen. Queue bounds and shadowing edge cases stay pinned at session tier "
+            "in tests/client/test_session_notification_bindings.py."
         ),
     ),
     # ═══════════════════════════════════════════════════════════════════════════
diff --git a/tests/interaction/mcpserver/test_extensions.py b/tests/interaction/mcpserver/test_extensions.py
index 129324538f..1a7b17724b 100644
--- a/tests/interaction/mcpserver/test_extensions.py
+++ b/tests/interaction/mcpserver/test_extensions.py
@@ -4,6 +4,7 @@
 from collections.abc import Awaitable, Callable, Sequence
 from typing import Any, Literal
 
+import anyio
 import mcp_types as types
 import pytest
 from inline_snapshot import snapshot
@@ -11,7 +12,7 @@
 from pydantic import ValidationError
 
 from mcp import MCPError
-from mcp.client import ClaimContext, ClientExtension, ResultClaim, advertise
+from mcp.client import ClaimContext, ClientExtension, NotificationBinding, ResultClaim, advertise
 from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
 from mcp.server.extension import Extension
 from mcp.server.mcpserver import Context, MCPServer, require_client_extension
@@ -175,3 +176,59 @@ def declared(ctx: Context) -> list[str]:
         result = await client.call_tool("declared", {})
 
     assert result.structured_content == {"result": [_FLAGS]}
+
+
+class ReceiptIssuedParams(types.NotificationParams):
+    seq: int
+
+
+class ReceiptIssued(types.Notification[ReceiptIssuedParams, Literal["notifications/receipts"]]):
+    """Vendor notification the Receipts extension observes; off-spec, so only a binding sees it."""
+
+    method: Literal["notifications/receipts"] = "notifications/receipts"
+    params: ReceiptIssuedParams
+
+
+class ReceiptFeed(ClientExtension):
+    """Client half: observes the vendor `notifications/receipts` stream."""
+
+    identifier = _RECEIPTS
+
+    def __init__(self, handler: Callable[[ReceiptIssuedParams], Awaitable[None]]) -> None:
+        self._handler = handler
+
+    def notifications(self) -> Sequence[NotificationBinding[Any]]:
+        return [
+            NotificationBinding(method="notifications/receipts", params_type=ReceiptIssuedParams, handler=self._handler)
+        ]
+
+
+@requirement("extensions:client:notification-binding-delivery")
+async def test_vendor_notifications_reach_the_owning_extensions_binding_in_order(connect: Connect) -> None:
+    """A tool emits vendor notifications through `ctx.session.send_notification`; the declaring
+    client's binding receives the validated params serially, in dispatch order."""
+    received: list[int] = []
+    done = anyio.Event()
+
+    async def on_receipt(params: ReceiptIssuedParams) -> None:
+        received.append(params.seq)
+        if params.seq == 3:
+            done.set()
+
+    server = MCPServer("issuer")
+
+    @server.tool()
+    async def buy_many(ctx: Context) -> str:
+        """Issue three receipts."""
+        for seq in (1, 2, 3):
+            await ctx.session.send_notification(
+                ReceiptIssued(params=ReceiptIssuedParams(seq=seq)), related_request_id=ctx.request_id
+            )
+        return "issued"
+
+    async with connect(server, extensions=[ReceiptFeed(on_receipt)]) as client:
+        await client.call_tool("buy_many", {})
+        with anyio.fail_after(5):
+            await done.wait()
+
+    assert received == [1, 2, 3]
diff --git a/tests/server/test_runner.py b/tests/server/test_runner.py
index eb212dafdb..ccf931f2a9 100644
--- a/tests/server/test_runner.py
+++ b/tests/server/test_runner.py
@@ -8,6 +8,7 @@
 """
 
 import contextvars
+import logging
 from collections.abc import AsyncIterator, Mapping
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field, replace
@@ -375,12 +376,12 @@ async def on_barrier(ctx: Ctx, params: NotificationParams) -> None:
     # A custom (non-spec) method bypasses the version gate, so it reaches its
     # handler regardless of which spec notifications exist at the pinned version.
     server.add_notification_handler("custom/barrier", NotificationParams, on_barrier)
-    with caplog.at_level("DEBUG", logger="mcp.server.runner"):
-        async with connected_runner(server) as (client, runner):
-            runner.connection.protocol_version = "2026-07-28"
-            await client.notify("notifications/roots/list_changed", None)
-            await client.notify("custom/barrier", None)
-            await barrier.wait()
+    async with connected_runner(server) as (client, runner):
+        runner.connection.protocol_version = "2026-07-28"
+        await client.notify("notifications/roots/list_changed", None)
+        await client.notify("custom/barrier", None)
+        await barrier.wait()
+    # Warn-level: the peer violated the negotiated version, so the drop must be visible.
     assert "dropped 'notifications/roots/list_changed': not defined at 2026-07-28" in caplog.text
 
 
@@ -576,20 +577,34 @@ async def drop_params(ctx: Ctx, call_next: Any) -> Any:
 
 
 @pytest.mark.anyio
-async def test_runner_on_notify_drops_before_init_and_unknown_methods(server: SrvT):
+async def test_runner_on_notify_drops_before_init_and_unknown_methods(server: SrvT, caplog: pytest.LogCaptureFixture):
     seen: list[Any] = []
 
     async def on_roots(ctx: Ctx, params: NotificationParams | None) -> None:
         seen.append(params)
 
     server.add_notification_handler("notifications/roots/list_changed", NotificationParams, on_roots)
-    async with connected_runner(server, initialized=False) as (client, _):
-        await client.notify("notifications/roots/list_changed", None)  # before init: dropped
-        await client.notify("notifications/initialized", None)
-        await client.notify("notifications/unknown", None)  # no handler: dropped
-        await client.notify("notifications/roots/list_changed", None)  # post-init: delivered
-        await anyio.wait_all_tasks_blocked()
+    with caplog.at_level(logging.DEBUG, logger="mcp.server.runner"):
+        async with connected_runner(server, initialized=False) as (client, _):
+            await client.notify("notifications/roots/list_changed", None)  # before init: dropped
+            await client.notify("notifications/initialized", None)
+            await client.notify("notifications/unknown", None)  # no handler: dropped
+            await client.notify("notifications/unknown", None)  # repeat: logged at debug, not warned again
+            await client.notify("notifications/roots/list_changed", None)  # post-init: delivered
+            await anyio.wait_all_tasks_blocked()
     assert seen == [NotificationParams()]  # only the post-init one reached the handler
+    # An unserved custom method warns once (repeats drop to debug); an unserved
+    # spec method (initialized) stays at debug.
+    drop_levels = [
+        r.levelno
+        for r in caplog.records
+        if r.getMessage() == "dropped 'notifications/unknown': no notification handler is registered"
+    ]
+    assert drop_levels == [logging.WARNING, logging.DEBUG]
+    spec_levels = [
+        r.levelno for r in caplog.records if r.getMessage() == "no handler for notification notifications/initialized"
+    ]
+    assert spec_levels == [logging.DEBUG]
 
 
 @pytest.mark.anyio