From 72ddad4f8bd3968e41f3d2e753c7f49f07375469 Mon Sep 17 00:00:00 2001
From: Max Isbey <224885523+maxisbey@users.noreply.github.com>
Date: Mon, 27 Jul 2026 23:33:18 +0000
Subject: [PATCH 1/4] Gate log notifications on the per-request log-level
 opt-in at 2026-07-28

The 2026-07-28 spec makes log delivery a per-request opt-in: a server
must not send notifications/message for a request whose _meta lacks
io.modelcontextprotocol/logLevel, may send only at or above that
level when it is present, and never on any stream but the one carrying
the response. The SDK sent unconditionally on modern connections.

allowed_log_levels(protocol_version, meta) in server/connection.py is
the one gate: handshake versions may send every level as before, modern
versions the at-or-above subset of the request's opt-in, or nothing
without one. ServerSession fixes the set at construction from the
inbound request's _meta and routes modern logs onto the requesting
stream regardless of related_request_id; Context.log applies the same
gate; Connection.log, with no request to opt in, never sends at 2026.

Client(log_level=...) stamps the opt-in on every modern request (a
per-call _meta entry overrides it), so a logging_callback on a 2026
connection can receive messages.
---
 docs/client/callbacks.md                    |  2 +-
 docs/migration.md                           | 15 +++-
 src/mcp/client/client.py                    | 11 +++
 src/mcp/client/session.py                   | 14 +++-
 src/mcp/server/connection.py                | 53 ++++++++++++--
 src/mcp/server/context.py                   | 20 +++++-
 src/mcp/server/runner.py                    |  3 +-
 src/mcp/server/session.py                   | 48 +++++++++++--
 tests/client/test_session.py                | 16 +++++
 tests/interaction/_connect.py               | 10 +++
 tests/interaction/_requirements.py          | 31 +++++++++
 tests/interaction/lowlevel/test_logging.py  | 77 ++++++++++++++++++++-
 tests/interaction/mcpserver/test_context.py |  4 +-
 tests/interaction/mcpserver/test_tools.py   |  2 +-
 tests/server/test_connection.py             | 10 +++
 tests/server/test_server_context.py         | 30 ++++++++
 tests/server/test_session.py                | 49 +++++++++++++
 tests/server/test_stateless_mode.py         | 17 +++--
 18 files changed, 384 insertions(+), 28 deletions(-)

diff --git a/docs/client/callbacks.md b/docs/client/callbacks.md
index 53f6e563ee..48ef310eff 100644
--- a/docs/client/callbacks.md
+++ b/docs/client/callbacks.md
@@ -133,7 +133,7 @@ Pass them to `Client(...)` exactly like `elicitation_callback`.
 
 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.
+`logging_callback` receives the `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. On a 2026-era connection the callback alone gets you nothing, because 2026 servers send log messages only to requests that opt in: pass `log_level="info"` (or another level) to `Client(...)` to stamp that opt-in on every request and receive that level and above. Pre-2026 servers ignore it and keep their `logging/setLevel` behavior.
 
 `message_handler` is the catch-all: every server notification the session surfaces reaches it (as well as its specific callback), and on a stream-backed transport so does every transport-level `Exception`. Two never do: `notifications/cancelled` is applied by the SDK rather than surfaced, and a subscription acknowledgment for a live `listen()` stream is consumed by that stream. Annotate the parameter with `IncomingMessage` (`ServerNotification | Exception`, exported from `mcp.client`). The one pattern worth knowing is `if isinstance(message, Exception): raise message`, so a broken connection fails loudly instead of vanishing.
 
diff --git a/docs/migration.md b/docs/migration.md
index 7b5950e7cf..7f64354684 100644
--- a/docs/migration.md
+++ b/docs/migration.md
@@ -1689,7 +1689,7 @@ The parametrized `Context[MyLifespanState]` annotation currently works only on `
 
 `ServerSession` no longer subclasses `BaseSession`. It is now a small per-request proxy that exposes `send_request`, `send_notification`, the typed convenience helpers — `create_message`, `elicit` / `elicit_form` / `elicit_url`, `send_elicit_complete`, `list_roots`, `send_log_message`, `send_resource_updated`, `send_resource_list_changed` / `send_tool_list_changed` / `send_prompt_list_changed`, `send_ping`, `send_progress_notification`, and the new `report_progress` — plus `check_client_capability` and the read-only `client_params`, `client_capabilities`, `protocol_version`, and `can_send_request` properties. The receive loop, `initialize` handling, and per-request task isolation that previously lived in `ServerSession` have moved to `JSONRPCDispatcher` and `ServerRunner`.
 
-The helpers keep their v1 signatures, so calls through `ctx.session` are source-compatible: `send_notification(notification, related_request_id=None)`, `send_log_message(level, data, logger=None, related_request_id=None)` (now [SEP-2577-deprecated](#roots-sampling-and-logging-methods-deprecated-sep-2577)), `send_progress_notification(progress_token, progress, total=None, message=None, related_request_id=None)`, `related_request_id=` on `elicit_form` / `elicit_url` / `send_elicit_complete`, and `metadata=ServerMessageMetadata(related_request_id=...)` on `send_request` (used by `create_message`). As in v1, a present `related_request_id` routes the message onto that request's own stream (the POST response in streamable HTTP) and an absent one uses the connection's standalone stream. Two adjustments: `send_resource_updated(uri)` accepts `str | AnyUrl`, and `send_notification` takes the notification model itself — the `types.ServerNotification(...)` wrapper is gone with the other `RootModel` unions (`await session.send_notification(types.ResourceListChangedNotification())`; see [Replace `RootModel` by union types with `TypeAdapter` validation](#replace-rootmodel-by-union-types-with-typeadapter-validation)).
+The helpers keep their v1 signatures, so calls through `ctx.session` are source-compatible: `send_notification(notification, related_request_id=None)`, `send_log_message(level, data, logger=None, related_request_id=None)` (now [SEP-2577-deprecated](#roots-sampling-and-logging-methods-deprecated-sep-2577)), `send_progress_notification(progress_token, progress, total=None, message=None, related_request_id=None)`, `related_request_id=` on `elicit_form` / `elicit_url` / `send_elicit_complete`, and `metadata=ServerMessageMetadata(related_request_id=...)` on `send_request` (used by `create_message`). As in v1, a present `related_request_id` routes the message onto that request's own stream (the POST response in streamable HTTP) and an absent one uses the connection's standalone stream — the one 2026-era exception being `send_log_message`, whose delivery is gated and request-scoped by the spec there (see [Log messages are delivered only to requests that opt in](#log-messages-are-delivered-only-to-requests-that-opt-in)). Two adjustments: `send_resource_updated(uri)` accepts `str | AnyUrl`, and `send_notification` takes the notification model itself — the `types.ServerNotification(...)` wrapper is gone with the other `RootModel` unions (`await session.send_notification(types.ResourceListChangedNotification())`; see [Replace `RootModel` by union types with `TypeAdapter` validation](#replace-rootmodel-by-union-types-with-typeadapter-validation)).
 
 Behavior changes:
 
@@ -2819,6 +2819,19 @@ async def book_table(date: str, answer: Annotated[Confirmation, Resolve(ask_to_c
 
 The client's same `elicitation_callback` answers both; the resolver lets the server *return* the question instead of pushing it.
 
+### Log messages are delivered only to requests that opt in
+
+At 2026-07-28 the deprecated logging capability changes shape: `logging/setLevel` is gone, and log delivery becomes a per-request opt-in. A server MUST NOT send `notifications/message` for a request whose `_meta` lacks `io.modelcontextprotocol/logLevel`, and when the key is present it sends only entries at or above that level, on that request's own stream. So on a 2026-era connection the request-scoped log calls — `ctx.info(...)` and friends on `MCPServer`'s `Context`, `ctx.session.send_log_message(...)`, `Context.log(...)` — are silently dropped (debug-logged) unless the request opted in, and dropped when they fall below the requested level; `Connection.log(...)`, which has no request to opt in, never sends there. Nothing changes on 2025-11-25 and earlier connections.
+
+The most visible consequence is the in-process `Client(server)`, which negotiates 2026-07-28 by default: a `logging_callback` that used to receive every message now receives nothing until the client opts in. `Client` grows a `log_level` argument for exactly this, stamped as the reserved `_meta` key on every modern request:
+
+```python
+async with Client(server, logging_callback=on_log, log_level="info") as client:
+    await client.call_tool("chatty", {})  # info and above reach `on_log`
+```
+
+`log_level=None` (the default) means no opt-in — a `logging_callback` alone is not one — and a single request can override the client-wide default by supplying the key in its own `meta=` (e.g. `meta={LOG_LEVEL_META_KEY: "debug"}` from `mcp_types`). The opt-in is what the spec calls for on 2026-era servers generally, not just this SDK's. Because 2026 log delivery is request-scoped by construction, `related_request_id` on `send_log_message` no longer selects the standalone stream there: whatever is delivered rides the requesting stream.
+
 ### Servers validate `Mcp-Param-*` headers against the request body ([SEP-2243](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2243))
 
 On the 2026-07-28 Streamable HTTP path, a `tools/call` whose tool declares `x-mcp-header` annotations is validated before dispatch — each annotated argument and its mirroring `Mcp-Param-*` header must be present together and agree (after base64-sentinel decoding; integers compare numerically), or absent together. A violation is rejected with HTTP 400 and JSON-RPC error `-32020` (`HeaderMismatch`), as the spec requires. A client that sends an annotated argument *without* its header — for example one that never listed the tool — is therefore rejected instead of silently served; the spec's recovery is to re-list and retry. On the client side, `ClientSession.call_tool` emits these headers automatically for annotated arguments of any tool it has listed; list the tool first, and note that pre-2026 connections and non-HTTP transports never emit them.
diff --git a/src/mcp/client/client.py b/src/mcp/client/client.py
index aaf7c83b0b..ed7c40f123 100644
--- a/src/mcp/client/client.py
+++ b/src/mcp/client/client.py
@@ -312,6 +312,16 @@ async def main():
     logging_callback: LoggingFnT | None = None
     """Callback for handling logging notifications."""
 
+    log_level: LoggingLevel | None = None
+    """The log level to opt in to on 2026-07-28+ connections (deprecated logging feature, SEP-2577).
+
+    Modern (2026-07-28+) servers send `notifications/message` only for requests that opt in by
+    carrying `io.modelcontextprotocol/logLevel` in `_meta`, and only at or above that level. Setting
+    this stamps that opt-in on every request; `None` (the default) means no opt-in, so no log
+    messages arrive - a `logging_callback` alone is not an opt-in. No effect on handshake-era
+    connections, where the deprecated `logging/setLevel` request governs delivery instead. A
+    per-request `_meta` entry with the same key overrides this default."""
+
     # TODO(Marcelo): Why do we have both "callback" and "handler"?
     message_handler: MessageHandlerFnT | None = None
     """Callback for handling raw messages."""
@@ -425,6 +435,7 @@ async def _build_session(self, exit_stack: AsyncExitStack) -> ClientSession:
             sampling_capabilities=self.sampling_capabilities,
             list_roots_callback=self.list_roots_callback,
             logging_callback=self.logging_callback,
+            log_level=self.log_level,
             message_handler=message_handler,
             client_info=self.client_info,
             elicitation_callback=self.elicitation_callback,
diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py
index 80539eaeee..895339ca18 100644
--- a/src/mcp/client/session.py
+++ b/src/mcp/client/session.py
@@ -19,6 +19,7 @@
     CLIENT_INFO_META_KEY,
     CONNECTION_CLOSED,
     INTERNAL_ERROR,
+    LOG_LEVEL_META_KEY,
     METHOD_NOT_FOUND,
     PROTOCOL_VERSION_META_KEY,
     SERVER_INFO_META_KEY,
@@ -121,6 +122,8 @@ def _make_modern_stamp(
     client_info: dict[str, Any],
     capabilities: dict[str, Any],
     resolve_param_headers: Callable[[str, Mapping[str, Any]], dict[str, str]],
+    *,
+    log_level: types.LoggingLevel | None = None,
 ) -> Callable[[dict[str, Any], CallOptions], None]:
     def stamp(data: dict[str, Any], opts: CallOptions) -> None:
         params = data.setdefault("params", {})
@@ -128,6 +131,11 @@ def stamp(data: dict[str, Any], opts: CallOptions) -> None:
         meta[PROTOCOL_VERSION_META_KEY] = protocol_version
         meta[CLIENT_INFO_META_KEY] = client_info
         meta[CLIENT_CAPABILITIES_META_KEY] = capabilities
+        # The per-request log-delivery opt-in (2026 logging is opt-in per
+        # request). A default the caller can override on any single call by
+        # supplying the key in that request's `_meta`, hence setdefault.
+        if log_level is not None:
+            meta.setdefault(LOG_LEVEL_META_KEY, log_level)
         # `cancel_on_abandon` stays at the dispatcher default (True): the
         # courtesy `notifications/cancelled` is the abandon signal. On the
         # stream transports it is the 2026 wire's cancellation spelling; the
@@ -372,6 +380,7 @@ def __init__(
         message_handler: MessageHandlerFnT | None = None,
         client_info: types.Implementation | None = None,
         *,
+        log_level: types.LoggingLevel | None = None,
         sampling_capabilities: types.SamplingCapability | None = None,
         extensions: dict[str, dict[str, Any]] | None = None,
         result_claims: Mapping[str, Sequence[ResultClaim[Any]]] | None = None,
@@ -393,6 +402,7 @@ def __init__(
         self._elicitation_callback = elicitation_callback or _default_elicitation_callback
         self._list_roots_callback = list_roots_callback or _default_list_roots_callback
         self._logging_callback = logging_callback or _default_logging_callback
+        self._log_level: types.LoggingLevel | None = log_level
         self._message_handler = message_handler or _default_message_handler
         self._tool_output_schemas: dict[str, dict[str, Any] | None] = {}
         # Compiled output-schema validators, derived from `_tool_output_schemas` and owned by
@@ -645,7 +655,9 @@ def adopt(self, result: types.InitializeResult | types.DiscoverResult) -> None:
             version = mutual[-1]
             client_info = self._client_info.model_dump(by_alias=True, mode="json", exclude_none=True)
             capabilities = self._build_capabilities(version).model_dump(by_alias=True, mode="json", exclude_none=True)
-            self._stamp = _make_modern_stamp(version, client_info, capabilities, self._resolve_param_headers)
+            self._stamp = _make_modern_stamp(
+                version, client_info, capabilities, self._resolve_param_headers, log_level=self._log_level
+            )
             self._discover_result = result
             self._discover_server_info = _parse_server_info_stamp(result)
             self._initialize_result = None
diff --git a/src/mcp/server/connection.py b/src/mcp/server/connection.py
index ca05928dff..e1fb1383d9 100644
--- a/src/mcp/server/connection.py
+++ b/src/mcp/server/connection.py
@@ -22,10 +22,11 @@
 import logging
 from collections.abc import Mapping
 from contextlib import AsyncExitStack
-from typing import Any, TypeVar, overload
+from typing import Any, Final, TypeVar, get_args, overload
 
 import anyio
 from mcp_types import (
+    LOG_LEVEL_META_KEY,
     ClientCapabilities,
     CreateMessageRequest,
     CreateMessageResult,
@@ -41,7 +42,7 @@
     Request,
 )
 from mcp_types import methods as _methods
-from mcp_types.version import LATEST_HANDSHAKE_VERSION
+from mcp_types.version import LATEST_HANDSHAKE_VERSION, MODERN_PROTOCOL_VERSIONS
 from pydantic import BaseModel, ValidationError
 from typing_extensions import deprecated
 
@@ -49,9 +50,43 @@
 from mcp.shared.exceptions import MCPDeprecationWarning, NoBackChannelError
 from mcp.shared.peer import Meta, dump_params
 
-__all__ = ["Connection"]
+__all__ = ["Connection", "allowed_log_levels"]
 
 logger = logging.getLogger(__name__)
+# `Connection.log`'s `logger` parameter (public API, the spec's logger-name
+# field) shadows the module logger inside that method; this alias keeps the
+# module logger reachable there.
+_logger = logger
+
+_LOG_LEVELS: Final[tuple[LoggingLevel, ...]] = get_args(LoggingLevel)
+"""Severity-ascending, from the `LoggingLevel` literal's declaration order (the
+RFC 5424 scale) - the literal is the single source of the ordering."""
+
+_ALL_LOG_LEVELS: Final[frozenset[LoggingLevel]] = frozenset(_LOG_LEVELS)
+
+
+def allowed_log_levels(protocol_version: str, meta: Mapping[str, Any] | None) -> frozenset[LoggingLevel]:
+    """The `notifications/message` levels deliverable for one inbound request.
+
+    2026-07-28+ makes log delivery a per-request opt-in (server/utilities/
+    logging): the client sets the reserved `io.modelcontextprotocol/logLevel`
+    `_meta` key, absent means no levels - the server MUST NOT send - and
+    present means that level and above. An unrecognized value reads as absent;
+    spec methods already reject a malformed value at surface validation
+    before any handler runs, so that arm only serves custom methods, where
+    dropping is the safe direction. Connection-scoped emitters pass
+    `meta=None`: `logging/setLevel` is gone at 2026 and log delivery is
+    request-scoped only, so they deliver nothing. Handshake versions keep
+    their `logging/setLevel`-era semantics: every level may be sent, filtering
+    is the application's `logging/setLevel` handler's job as before.
+    """
+    if protocol_version not in MODERN_PROTOCOL_VERSIONS:
+        return _ALL_LOG_LEVELS
+    requested = (meta or {}).get(LOG_LEVEL_META_KEY)
+    if requested not in _LOG_LEVELS:
+        return frozenset()
+    return frozenset(_LOG_LEVELS[_LOG_LEVELS.index(requested) :])
+
 
 ResultT = TypeVar("ResultT", bound=BaseModel)
 
@@ -373,7 +408,17 @@ async def ping(self, *, meta: Meta | None = None, opts: CallOptions | None = Non
 
     @deprecated("The logging capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
     async def log(self, level: LoggingLevel, data: Any, logger: str | None = None, *, meta: Meta | None = None) -> None:
-        """Send a `notifications/message` log entry on the standalone stream. Best-effort."""
+        """Send a `notifications/message` log entry on the standalone stream. Best-effort.
+
+        On 2026-07-28+ connections this never sends: log delivery is a
+        per-request opt-in that rides the requesting stream (`ctx.log`,
+        `ctx.session.send_log_message`), and the standalone stream is
+        forbidden from carrying `notifications/message`, so the entry is
+        debug-logged and dropped.
+        """
+        if level not in allowed_log_levels(self.protocol_version, None):
+            _logger.debug("dropped notifications/message: no connection-wide log delivery at %s", self.protocol_version)
+            return
         params: dict[str, Any] = {"level": level, "data": data}
         if logger is not None:
             params["logger"] = logger
diff --git a/src/mcp/server/context.py b/src/mcp/server/context.py
index 903b7ef6f8..bfcb9c9ca4 100644
--- a/src/mcp/server/context.py
+++ b/src/mcp/server/context.py
@@ -1,3 +1,4 @@
+import logging
 from collections.abc import Awaitable, Callable, Mapping
 from dataclasses import dataclass
 from typing import Any, Generic, Protocol
@@ -6,7 +7,7 @@
 from pydantic import BaseModel
 from typing_extensions import TypeVar, deprecated
 
-from mcp.server.connection import Connection
+from mcp.server.connection import Connection, allowed_log_levels
 from mcp.server.session import ServerSession
 from mcp.shared.context import BaseContext
 from mcp.shared.dispatcher import DispatchContext
@@ -15,6 +16,12 @@
 from mcp.shared.peer import Meta
 from mcp.shared.transport_context import TransportContext
 
+logger = logging.getLogger(__name__)
+# `Context.log`'s `logger` parameter (public API, the spec's logger-name
+# field) shadows the module logger inside that method; this alias keeps it
+# reachable there.
+_logger = logger
+
 # Invariant: parametrizes a mutable dataclass field; dict default matches the default lifespan.
 LifespanContextT = TypeVar("LifespanContextT", default=dict[str, Any])
 RequestT = TypeVar("RequestT", default=Any)
@@ -67,6 +74,9 @@ def __init__(
         super().__init__(dctx, meta=meta)
         self._lifespan = lifespan
         self._connection = connection
+        # Same per-request log gate as `ServerSession`: fixed at construction
+        # from this request's `_meta` log-level opt-in and the connection's era.
+        self._allowed_log_levels = allowed_log_levels(connection.protocol_version, meta)
 
     @property
     def lifespan(self) -> LifespanT_co:
@@ -102,7 +112,15 @@ async def log(self, level: LoggingLevel, data: Any, logger: str | None = None, *
         Uses this request's back-channel (so the entry rides the request's SSE
         stream in streamable HTTP), not the standalone stream - use
         `ctx.connection.log(...)` for that.
+
+        On 2026-07-28+ delivery is a per-request opt-in: nothing is sent
+        unless this request's `_meta` carried the reserved log-level key, and
+        entries below the requested level are dropped (debug-logged).
+        Handshake versions send unconditionally, as before.
         """
+        if level not in self._allowed_log_levels:
+            _logger.debug("dropped notifications/message at %r: not opted in at that level on this request", level)
+            return
         params: dict[str, Any] = {"level": level, "data": data}
         if logger is not None:
             params["logger"] = logger
diff --git a/src/mcp/server/runner.py b/src/mcp/server/runner.py
index 6f9f7a8f74..b799a00be6 100644
--- a/src/mcp/server/runner.py
+++ b/src/mcp/server/runner.py
@@ -321,7 +321,8 @@ def _make_context(
         # Per-request session: `dctx` is the request-scoped channel (auto-threads
         # its own request_id on streamable HTTP); the standalone channel is read
         # off `connection.outbound`. `related_request_id` on the public API selects.
-        session = ServerSession(dctx, self.connection)
+        # `meta` carries this request's log-level opt-in for the session's log gate.
+        session = ServerSession(dctx, self.connection, request_meta=meta)
         return ServerRequestContext(
             session=session,
             lifespan_context=self.lifespan_state,
diff --git a/src/mcp/server/session.py b/src/mcp/server/session.py
index 69ad5ecadb..bb446415e8 100644
--- a/src/mcp/server/session.py
+++ b/src/mcp/server/session.py
@@ -6,14 +6,16 @@
 `send_log_message`, ...) to call back to the client.
 """
 
+import logging
 from typing import Any, TypeVar, overload
 
 import mcp_types as types
 from mcp_types import methods as _methods
+from mcp_types.version import MODERN_PROTOCOL_VERSIONS
 from pydantic import AnyUrl, BaseModel
 from typing_extensions import deprecated
 
-from mcp.server.connection import Connection
+from mcp.server.connection import Connection, allowed_log_levels
 from mcp.server.validation import validate_sampling_tools, validate_tool_use_result_messages, wants_sampling_tools
 from mcp.shared.dispatcher import CallOptions, DispatchContext, ProgressFnT
 from mcp.shared.exceptions import MCPDeprecationWarning
@@ -21,6 +23,12 @@
 
 __all__ = ["ServerSession"]
 
+logger = logging.getLogger(__name__)
+# `send_log_message`'s `logger` parameter (public API, the spec's logger-name
+# field) shadows the module logger inside that method; this alias keeps it
+# reachable there.
+_logger = logger
+
 ResultT = TypeVar("ResultT", bound=BaseModel)
 
 
@@ -36,9 +44,22 @@ class ServerSession:
     never crosses the `Outbound` Protocol.
     """
 
-    def __init__(self, request_outbound: DispatchContext[Any], connection: Connection) -> None:
+    def __init__(
+        self,
+        request_outbound: DispatchContext[Any],
+        connection: Connection,
+        *,
+        request_meta: types.RequestParamsMeta | None = None,
+    ) -> None:
         self._request_outbound = request_outbound
         self._connection = connection
+        # The per-request log-delivery contract, fixed at construction: on
+        # 2026-07-28+ the inbound request's `_meta` log-level opt-in decides
+        # which `notifications/message` levels may be sent for this request
+        # (and they ride this request's stream only); on handshake versions
+        # every level may be sent (`logging/setLevel`-era semantics).
+        self._log_is_request_scoped = connection.protocol_version in MODERN_PROTOCOL_VERSIONS
+        self._allowed_log_levels = allowed_log_levels(connection.protocol_version, request_meta)
 
     @property
     def client_params(self) -> types.InitializeRequestParams | None:
@@ -106,7 +127,10 @@ async def send_notification(
         related_request_id: types.RequestId | None = None,
     ) -> None:
         """Send a typed server-to-client notification."""
-        channel = self._request_outbound if related_request_id is not None else self._connection.outbound
+        await self._notify(notification, request_scoped=related_request_id is not None)
+
+    async def _notify(self, notification: types.ServerNotification, *, request_scoped: bool) -> None:
+        channel = self._request_outbound if request_scoped else self._connection.outbound
         data = notification.model_dump(by_alias=True, mode="json", exclude_none=True)
         await channel.notify(data["method"], data.get("params"))
 
@@ -122,8 +146,20 @@ async def send_log_message(
         logger: str | None = None,
         related_request_id: types.RequestId | None = None,
     ) -> None:
-        """Send a log message notification."""
-        await self.send_notification(
+        """Send a log message notification.
+
+        On 2026-07-28+ delivery is a per-request opt-in: nothing is sent
+        unless this request's `_meta` carried the reserved log-level key, and
+        entries below the requested level are dropped (debug-logged). What is
+        sent rides this request's stream regardless of `related_request_id` -
+        the spec forbids `notifications/message` on any stream but the one
+        carrying the response. Handshake versions send unconditionally on the
+        channel `related_request_id` selects, as before.
+        """
+        if level not in self._allowed_log_levels:
+            _logger.debug("dropped notifications/message at %r: not opted in at that level on this request", level)
+            return
+        await self._notify(
             types.LoggingMessageNotification(
                 params=types.LoggingMessageNotificationParams(
                     level=level,
@@ -131,7 +167,7 @@ async def send_log_message(
                     logger=logger,
                 ),
             ),
-            related_request_id,
+            request_scoped=self._log_is_request_scoped or related_request_id is not None,
         )
 
     async def send_resource_updated(self, uri: str | AnyUrl) -> None:
diff --git a/tests/client/test_session.py b/tests/client/test_session.py
index 12fab4c8c9..6e5e0a14d8 100644
--- a/tests/client/test_session.py
+++ b/tests/client/test_session.py
@@ -13,6 +13,7 @@
     CONNECTION_CLOSED,
     INTERNAL_ERROR,
     INVALID_PARAMS,
+    LOG_LEVEL_META_KEY,
     METHOD_NOT_FOUND,
     PROTOCOL_VERSION_META_KEY,
     REQUEST_TIMEOUT,
@@ -1548,6 +1549,21 @@ async def test_discover_adopts_the_returned_result_and_installs_the_modern_stamp
     assert ping_params["_meta"][PROTOCOL_VERSION_META_KEY] == "2026-07-28"
 
 
+@pytest.mark.anyio
+async def test_log_level_opt_in_is_stamped_on_modern_requests_and_overridable_per_call() -> None:
+    """SDK-defined: `log_level` stamps the reserved log-level `_meta` key on every modern
+    request, and a request supplying that key in its own `_meta` overrides the default."""
+    dispatcher = _ScriptedDispatcher(_discover_result_dict(), {}, {})
+    with anyio.fail_after(5):
+        async with ClientSession(dispatcher=dispatcher, log_level="warning") as session:
+            await session.discover()
+            await session.send_ping()
+            await session.send_ping(meta={LOG_LEVEL_META_KEY: "debug"})
+    default_meta, override_meta = (params["_meta"] for _, params in dispatcher.calls[-2:] if params is not None)
+    assert default_meta[LOG_LEVEL_META_KEY] == "warning"
+    assert override_meta[LOG_LEVEL_META_KEY] == "debug"
+
+
 @pytest.mark.anyio
 async def test_discover_retries_once_on_unsupported_version_then_adopts() -> None:
     """Spec SHOULD: a -32022 reply that names a mutually-supported version
diff --git a/tests/interaction/_connect.py b/tests/interaction/_connect.py
index 651dd6a8f9..6a8094e2d6 100644
--- a/tests/interaction/_connect.py
+++ b/tests/interaction/_connect.py
@@ -21,6 +21,7 @@
     JSONRPCMessage,
     JSONRPCRequest,
     JSONRPCResponse,
+    LoggingLevel,
     jsonrpc_message_adapter,
 )
 from mcp_types.version import LATEST_HANDSHAKE_VERSION, MODERN_PROTOCOL_VERSIONS
@@ -68,6 +69,7 @@ def __call__(
         sampling_callback: SamplingFnT | None = None,
         list_roots_callback: ListRootsFnT | None = None,
         logging_callback: LoggingFnT | None = None,
+        log_level: LoggingLevel | None = None,
         message_handler: MessageHandlerFnT | None = None,
         client_info: Implementation | None = None,
         elicitation_callback: ElicitationFnT | None = None,
@@ -84,6 +86,7 @@ async def connect_in_memory(
     sampling_callback: SamplingFnT | None = None,
     list_roots_callback: ListRootsFnT | None = None,
     logging_callback: LoggingFnT | None = None,
+    log_level: LoggingLevel | None = None,
     message_handler: MessageHandlerFnT | None = None,
     client_info: Implementation | None = None,
     elicitation_callback: ElicitationFnT | None = None,
@@ -103,6 +106,7 @@ async def connect_in_memory(
         sampling_callback=sampling_callback,
         list_roots_callback=list_roots_callback,
         logging_callback=logging_callback,
+        log_level=log_level,
         message_handler=message_handler,
         client_info=client_info,
         elicitation_callback=elicitation_callback,
@@ -123,6 +127,7 @@ async def connect_over_streamable_http(
     sampling_callback: SamplingFnT | None = None,
     list_roots_callback: ListRootsFnT | None = None,
     logging_callback: LoggingFnT | None = None,
+    log_level: LoggingLevel | None = None,
     message_handler: MessageHandlerFnT | None = None,
     client_info: Implementation | None = None,
     elicitation_callback: ElicitationFnT | None = None,
@@ -158,6 +163,7 @@ async def connect_over_streamable_http(
             sampling_callback=sampling_callback,
             list_roots_callback=list_roots_callback,
             logging_callback=logging_callback,
+            log_level=log_level,
             message_handler=message_handler,
             client_info=client_info,
             elicitation_callback=elicitation_callback,
@@ -232,6 +238,7 @@ async def client_via_http(
     http_client: httpx2.AsyncClient,
     *,
     logging_callback: LoggingFnT | None = None,
+    log_level: LoggingLevel | None = None,
     message_handler: MessageHandlerFnT | None = None,
     elicitation_callback: ElicitationFnT | None = None,
 ) -> AsyncIterator[Client]:
@@ -248,6 +255,7 @@ async def client_via_http(
         # closing DELETE); the modern flow is sessionless and would silently change the subject.
         mode="legacy",
         logging_callback=logging_callback,
+        log_level=log_level,
         message_handler=message_handler,
         elicitation_callback=elicitation_callback,
     ) as client:
@@ -360,6 +368,7 @@ async def connect_over_sse(
     sampling_callback: SamplingFnT | None = None,
     list_roots_callback: ListRootsFnT | None = None,
     logging_callback: LoggingFnT | None = None,
+    log_level: LoggingLevel | None = None,
     message_handler: MessageHandlerFnT | None = None,
     client_info: Implementation | None = None,
     elicitation_callback: ElicitationFnT | None = None,
@@ -394,6 +403,7 @@ def httpx_client_factory(
         sampling_callback=sampling_callback,
         list_roots_callback=list_roots_callback,
         logging_callback=logging_callback,
+        log_level=log_level,
         message_handler=message_handler,
         client_info=client_info,
         elicitation_callback=elicitation_callback,
diff --git a/tests/interaction/_requirements.py b/tests/interaction/_requirements.py
index a26f7d9983..567c4b1ade 100644
--- a/tests/interaction/_requirements.py
+++ b/tests/interaction/_requirements.py
@@ -1522,6 +1522,7 @@ def __post_init__(self) -> None:
             ),
         ),
         removed_in="2026-07-28",
+        superseded_by="logging:per-request:threshold",
         note=(
             "removed in 2026-07-28 (SEP-2575); logging/setLevel removed, replaced by per-request "
             "io.modelcontextprotocol/logLevel in _meta."
@@ -1531,6 +1532,7 @@ def __post_init__(self) -> None:
         source=f"{SPEC_BASE_URL}/server/utilities/logging#setting-log-level",
         behavior="logging/setLevel delivers the requested level to the server's handler and returns an empty result.",
         removed_in="2026-07-28",
+        superseded_by="logging:per-request:opt-in",
         note=(
             "removed in 2026-07-28 (SEP-2575); logging/setLevel removed, replaced by per-request "
             "io.modelcontextprotocol/logLevel in _meta."
@@ -1540,11 +1542,40 @@ def __post_init__(self) -> None:
         source=f"{SPEC_BASE_URL}/server/utilities/logging#error-handling",
         behavior="logging/setLevel with an invalid level value returns JSON-RPC error -32602 (Invalid params).",
         removed_in="2026-07-28",
+        superseded_by="logging:per-request:invalid-level",
         note=(
             "removed in 2026-07-28 (SEP-2575); logging/setLevel removed, replaced by per-request "
             "io.modelcontextprotocol/logLevel in _meta."
         ),
     ),
+    "logging:per-request:opt-in": Requirement(
+        source=f"{SPEC_2026_BASE_URL}/server/utilities/logging#per-request-log-level",
+        behavior=(
+            "The server does not send log message notifications for a request unless the request opts in by "
+            "carrying io.modelcontextprotocol/logLevel in _meta; a handler's log calls on an un-opted "
+            "request are dropped, not delivered on another stream."
+        ),
+        added_in="2026-07-28",
+        supersedes=("logging:set-level",),
+    ),
+    "logging:per-request:threshold": Requirement(
+        source=f"{SPEC_2026_BASE_URL}/server/utilities/logging#per-request-log-level",
+        behavior=(
+            "A request that opts in receives log message notifications only at or above the level named in "
+            "its io.modelcontextprotocol/logLevel; entries below the level are dropped."
+        ),
+        added_in="2026-07-28",
+        supersedes=("logging:message:filtered",),
+    ),
+    "logging:per-request:invalid-level": Requirement(
+        source=f"{SPEC_2026_BASE_URL}/server/utilities/logging#error-handling",
+        behavior=(
+            "A request whose io.modelcontextprotocol/logLevel is not a recognized log level is rejected with "
+            "JSON-RPC error -32602 (Invalid params)."
+        ),
+        added_in="2026-07-28",
+        supersedes=("logging:set-level:invalid-level",),
+    ),
     # ═══════════════════════════════════════════════════════════════════════════
     # Sampling (server → client)
     # ═══════════════════════════════════════════════════════════════════════════
diff --git a/tests/interaction/lowlevel/test_logging.py b/tests/interaction/lowlevel/test_logging.py
index 7f2f89fd1a..cae5eec522 100644
--- a/tests/interaction/lowlevel/test_logging.py
+++ b/tests/interaction/lowlevel/test_logging.py
@@ -5,11 +5,21 @@
 streamable HTTP, so plain-list collection is deterministic on every transport leg.
 """
 
+from unittest.mock import AsyncMock
+
 import mcp_types as types
 import pytest
 from inline_snapshot import snapshot
-from mcp_types import CallToolResult, EmptyResult, LoggingMessageNotificationParams, TextContent
+from mcp_types import (
+    INVALID_PARAMS,
+    LOG_LEVEL_META_KEY,
+    CallToolResult,
+    EmptyResult,
+    LoggingMessageNotificationParams,
+    TextContent,
+)
 
+from mcp import MCPError
 from mcp.server import Server, ServerRequestContext
 from tests._stamp import Unstamp
 from tests.interaction._connect import Connect
@@ -81,7 +91,7 @@ async def set_logging_level(ctx: ServerRequestContext, params: types.SetLevelReq
         "logger", on_list_tools=list_tools, on_call_tool=call_tool, on_set_logging_level=set_logging_level
     )
 
-    async with connect(server, logging_callback=collect) as client:
+    async with connect(server, logging_callback=collect, log_level="debug") as client:
         result = await client.call_tool("chatty", {})
 
     assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="done")]))
@@ -122,7 +132,68 @@ async def set_logging_level(ctx: ServerRequestContext, params: types.SetLevelReq
         "logger", on_list_tools=list_tools, on_call_tool=call_tool, on_set_logging_level=set_logging_level
     )
 
-    async with connect(server, logging_callback=collect) as client:
+    async with connect(server, logging_callback=collect, log_level="debug") as client:
         await client.call_tool("siren", {})
 
     assert [params.level for params in received] == list(ALL_LEVELS)
+
+
+def _siren_server() -> Server:
+    """A server whose `siren` tool logs one message at each of the eight severity levels.
+
+    The messages are sent without `related_request_id`: on 2026-07-28+ log delivery is
+    request-scoped by construction, so they still ride the requesting stream on every leg.
+    """
+
+    async def list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        return types.ListToolsResult(tools=[types.Tool(name="siren", input_schema={"type": "object"})])
+
+    async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
+        assert params.name == "siren"
+        for level in ALL_LEVELS:
+            await ctx.session.send_log_message(level=level, data=f"a {level} message")  # pyright: ignore[reportDeprecated]
+        return CallToolResult(content=[TextContent(text="logged")])
+
+    return Server("logger", on_list_tools=list_tools, on_call_tool=call_tool)
+
+
+@requirement("logging:per-request:opt-in")
+async def test_no_log_messages_are_sent_for_a_request_without_a_log_level(connect: Connect) -> None:
+    """A request whose _meta carries no io.modelcontextprotocol/logLevel gets no log messages at all.
+
+    The handler emits at every severity, but the client never opts in, so the server must send
+    nothing: the log calls are dropped rather than delivered on some other stream.
+    """
+    logging_callback = AsyncMock()
+
+    async with connect(_siren_server(), logging_callback=logging_callback) as client:
+        result = await client.call_tool("siren", {})
+
+    assert isinstance(result.content[0], TextContent) and result.content[0].text == "logged"
+    logging_callback.assert_not_awaited()
+
+
+@requirement("logging:per-request:threshold")
+async def test_log_messages_below_the_requested_level_are_dropped(connect: Connect) -> None:
+    """A request opting in at `warning` receives only warning and above, in order."""
+    received: list[LoggingMessageNotificationParams] = []
+
+    async def collect(params: LoggingMessageNotificationParams) -> None:
+        received.append(params)
+
+    async with connect(_siren_server(), logging_callback=collect, log_level="warning") as client:
+        await client.call_tool("siren", {})
+
+    assert [params.level for params in received] == ["warning", "error", "critical", "alert", "emergency"]
+
+
+@requirement("logging:per-request:invalid-level")
+async def test_a_request_with_an_unrecognized_log_level_is_rejected(connect: Connect) -> None:
+    """A request whose _meta names an unrecognized log level is rejected with -32602 before the handler runs."""
+    async with connect(_siren_server()) as client:
+        with pytest.raises(MCPError) as exc_info:
+            await client.call_tool("siren", {}, meta={LOG_LEVEL_META_KEY: "verbose"})
+
+    assert exc_info.value.error.code == INVALID_PARAMS
diff --git a/tests/interaction/mcpserver/test_context.py b/tests/interaction/mcpserver/test_context.py
index 0f07a128b7..2b979b8936 100644
--- a/tests/interaction/mcpserver/test_context.py
+++ b/tests/interaction/mcpserver/test_context.py
@@ -50,7 +50,7 @@ async def narrate(ctx: Context) -> str:
     async def collect(params: LoggingMessageNotificationParams) -> None:
         received.append(params)
 
-    async with connect(mcp, logging_callback=collect) as client:
+    async with connect(mcp, logging_callback=collect, log_level="debug") as client:
         result = await client.call_tool("narrate", {})
         advertised_logging = client.server_capabilities.logging
 
@@ -142,7 +142,7 @@ async def mill(ctx: Context) -> str:
     async def collect(message: IncomingMessage) -> None:
         received.append(message)
 
-    async with connect(mcp, message_handler=collect) as client:
+    async with connect(mcp, message_handler=collect, log_level="debug") as client:
         result = await client.call_tool("mill", {})
 
     assert unstamped(result) == snapshot(
diff --git a/tests/interaction/mcpserver/test_tools.py b/tests/interaction/mcpserver/test_tools.py
index 21a7163360..a6418ac9c5 100644
--- a/tests/interaction/mcpserver/test_tools.py
+++ b/tests/interaction/mcpserver/test_tools.py
@@ -431,7 +431,7 @@ async def grow(ctx: Context) -> str:
     async def collect(message: IncomingMessage) -> None:
         received.append(message)
 
-    async with connect(mcp, message_handler=collect) as client:
+    async with connect(mcp, message_handler=collect, log_level="debug") as client:
         before = await client.list_tools()
         await client.call_tool("grow", {})
         after = await client.list_tools()
diff --git a/tests/server/test_connection.py b/tests/server/test_connection.py
index 31caf87bf2..683473d620 100644
--- a/tests/server/test_connection.py
+++ b/tests/server/test_connection.py
@@ -283,6 +283,16 @@ async def test_connection_log_sends_logging_message_notification():
     assert params["logger"] == "my.logger"
 
 
+@pytest.mark.anyio
+async def test_connection_log_sends_nothing_on_a_modern_connection():
+    """2026 log delivery is a per-request opt-in on the requesting stream; the
+    connection-scoped standalone entry has no request to opt in, so it never sends."""
+    out = StubOutbound()
+    conn = Connection.from_envelope(LATEST_MODERN_VERSION, None, None, outbound=out)
+    await conn.log("emergency", "unheard")  # pyright: ignore[reportDeprecated]
+    assert out.notifications == []
+
+
 @pytest.mark.anyio
 async def test_connection_log_with_meta_includes_meta_in_params():
     out = StubOutbound()
diff --git a/tests/server/test_server_context.py b/tests/server/test_server_context.py
index 9a9eaa3d97..9907e18013 100644
--- a/tests/server/test_server_context.py
+++ b/tests/server/test_server_context.py
@@ -11,6 +11,8 @@
 
 import anyio
 import pytest
+from mcp_types import LOG_LEVEL_META_KEY
+from mcp_types.version import LATEST_MODERN_VERSION
 
 from mcp.server.connection import Connection
 from mcp.server.context import Context
@@ -73,6 +75,34 @@ async def server_on_request(dctx: DCtx, method: str, params: Mapping[str, Any] |
         assert params is not None and params["level"] == "debug" and params["data"] == "hello"
 
 
+@pytest.mark.anyio
+async def test_context_log_is_gated_by_the_request_log_level_at_2026():
+    """On a 2026 connection an un-opted request delivers nothing; opting in at
+    `warning` delivers `warning`+ and drops what falls below."""
+    crec = Recorder()
+    _, c_notify = echo_handlers(crec)
+
+    async def server_on_request(dctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:
+        modern = Connection.from_envelope(LATEST_MODERN_VERSION, None, None, outbound=dctx)
+        silent: Context[_Lifespan] = Context(dctx, lifespan=_Lifespan("app"), connection=modern)
+        await silent.log("emergency", "dropped: no opt-in")  # pyright: ignore[reportDeprecated]
+        opted: Context[_Lifespan] = Context(
+            dctx, lifespan=_Lifespan("app"), connection=modern, meta={LOG_LEVEL_META_KEY: "warning"}
+        )
+        await opted.log("info", "dropped: below level")  # pyright: ignore[reportDeprecated]
+        await opted.log("warning", "delivered")  # pyright: ignore[reportDeprecated]
+        return {}
+
+    async with running_pair(direct_pair, server_on_request=server_on_request, client_on_notify=c_notify) as (
+        client,
+        *_,
+    ):
+        with anyio.fail_after(5):
+            await client.send_raw_request("t", None)
+            await crec.notified.wait()
+        assert [p["data"] for _, p in crec.notifications if p is not None] == ["delivered"]
+
+
 @pytest.mark.anyio
 async def test_context_log_includes_logger_and_meta_when_supplied():
     crec = Recorder()
diff --git a/tests/server/test_session.py b/tests/server/test_session.py
index 49e3b4615c..9d039fe59b 100644
--- a/tests/server/test_session.py
+++ b/tests/server/test_session.py
@@ -12,6 +12,7 @@
 import mcp_types as types
 import pytest
 from mcp_types import (
+    LOG_LEVEL_META_KEY,
     ClientCapabilities,
     Implementation,
     SamplingCapability,
@@ -157,6 +158,54 @@ async def test_send_notification_routes_by_related_request_id():
     assert [m for m, _ in request_ch.notifications] == ["notifications/progress"]
 
 
+def _modern_session(
+    request_ch: StubOutbound, standalone_ch: StubOutbound, *, request_meta: types.RequestParamsMeta | None = None
+) -> ServerSession:
+    """A 2026-era session with distinct channels, carrying the inbound request's `_meta`."""
+    conn = Connection.from_envelope(LATEST_MODERN_VERSION, None, None, outbound=standalone_ch)
+    return ServerSession(request_ch, conn, request_meta=request_meta)
+
+
+@pytest.mark.anyio
+async def test_send_log_message_drops_everything_without_a_log_level_opt_in_at_2026():
+    """No `_meta` log-level opt-in on a 2026 request means no `notifications/message` at all."""
+    request_ch, standalone_ch = StubOutbound(), StubOutbound()
+    session = _modern_session(request_ch, standalone_ch)
+    await session.send_log_message("emergency", "on fire", related_request_id="req-1")  # pyright: ignore[reportDeprecated]
+    assert request_ch.notifications == [] and standalone_ch.notifications == []
+
+
+@pytest.mark.anyio
+async def test_send_log_message_drops_levels_below_the_requested_one_at_2026():
+    request_ch, standalone_ch = StubOutbound(), StubOutbound()
+    session = _modern_session(request_ch, standalone_ch, request_meta={LOG_LEVEL_META_KEY: "warning"})
+    await session.send_log_message("info", "quiet")  # pyright: ignore[reportDeprecated]
+    await session.send_log_message("error", "loud")  # pyright: ignore[reportDeprecated]
+    assert [p["level"] for _, p in request_ch.notifications if p is not None] == ["error"]
+
+
+@pytest.mark.anyio
+async def test_send_log_message_is_request_scoped_at_2026_even_without_related_request_id():
+    """The spec forbids 2026 log delivery on any stream but the requesting one, so
+    `related_request_id` no longer selects the standalone channel there."""
+    request_ch, standalone_ch = StubOutbound(), StubOutbound()
+    session = _modern_session(request_ch, standalone_ch, request_meta={LOG_LEVEL_META_KEY: "debug"})
+    await session.send_log_message("info", "hello")  # pyright: ignore[reportDeprecated]
+    assert [m for m, _ in request_ch.notifications] == ["notifications/message"]
+    assert standalone_ch.notifications == []
+
+
+@pytest.mark.anyio
+async def test_send_log_message_on_a_handshake_version_still_routes_by_related_request_id():
+    """Handshake versions keep the pre-2026 semantics: every level sends, channel by `related_request_id`."""
+    request_ch, standalone_ch = StubOutbound(), StubOutbound()
+    session = _two_channel_session(request_ch, standalone_ch)
+    await session.send_log_message("debug", "loose")  # pyright: ignore[reportDeprecated]
+    await session.send_log_message("debug", "tied", related_request_id="req-1")  # pyright: ignore[reportDeprecated]
+    assert [m for m, _ in standalone_ch.notifications] == ["notifications/message"]
+    assert [m for m, _ in request_ch.notifications] == ["notifications/message"]
+
+
 @pytest.mark.anyio
 async def test_report_progress_delegates_to_the_request_dispatch_context():
     """`report_progress` calls the per-request `DispatchContext.progress` seam, never the
diff --git a/tests/server/test_stateless_mode.py b/tests/server/test_stateless_mode.py
index 1124d69b71..6b9785bc3a 100644
--- a/tests/server/test_stateless_mode.py
+++ b/tests/server/test_stateless_mode.py
@@ -12,7 +12,7 @@
 
 import mcp_types as types
 import pytest
-from mcp_types import LATEST_PROTOCOL_VERSION
+from mcp_types import LATEST_PROTOCOL_VERSION, LOG_LEVEL_META_KEY
 
 from mcp.server.connection import Connection
 from mcp.server.session import ServerSession
@@ -53,13 +53,15 @@ async def progress(self, progress: float, total: float | None = None, message: s
         raise NotImplementedError  # pragma: no cover
 
 
-def _no_channel_session(request_ch: StubOutbound | None = None) -> tuple[ServerSession, StubOutbound]:
+def _no_channel_session(
+    request_ch: StubOutbound | None = None, *, request_meta: types.RequestParamsMeta | None = None
+) -> tuple[ServerSession, StubOutbound]:
     """A session whose standalone channel is the connection's no-channel
     sentinel; the request channel is a working stub."""
     conn = Connection.from_envelope(LATEST_PROTOCOL_VERSION, None, None)
     assert conn.has_standalone_channel is False
     request = request_ch if request_ch is not None else StubOutbound()
-    return ServerSession(request, conn), request
+    return ServerSession(request, conn, request_meta=request_meta), request
 
 
 @pytest.fixture
@@ -141,10 +143,11 @@ async def test_elicit_form_with_related_id_rides_the_request_channel():
 
 
 @pytest.mark.anyio
-async def test_send_log_message_with_related_id_rides_the_request_channel():
-    """SDK-defined: the deprecated ``send_log_message`` notification with a related id
-    rides the per-request channel, so it is delivered even with no standalone back-channel."""
-    session, request_ch = _no_channel_session()
+async def test_send_log_message_rides_the_request_channel_when_opted_in():
+    """SDK-defined: the deprecated `send_log_message` notification rides the per-request
+    channel on a 2026 connection (log delivery is request-scoped by spec), so it is delivered
+    even with no standalone back-channel - once the request opted in via its `_meta`."""
+    session, request_ch = _no_channel_session(request_meta={LOG_LEVEL_META_KEY: "debug"})
     await session.send_log_message(  # pyright: ignore[reportDeprecated]
         level="info", data="hello", logger="test", related_request_id=3
     )

From 101e8813d84e5e6e6a62e3e9355ce33ef2bf9986 Mon Sep 17 00:00:00 2001
From: Max Isbey <224885523+maxisbey@users.noreply.github.com>
Date: Mon, 27 Jul 2026 23:41:10 +0000
Subject: [PATCH 2/4] Opt the streaming story client into log delivery

The streaming story's client asserts that three log messages arrive
during the countdown, but never opted in, so on 2026-07-28 arms it
depended on the server hand-sending the notification around the
per-request log gate. Pass log_level="info" so the story models the
spec-correct opt-in, and rewrite the server comment that promised a
non-deprecated logging helper - protocol logging is deprecated
(SEP-2577), so the raw send is what keeps the example warning-free.
---
 examples/stories/streaming/client.py | 4 +++-
 examples/stories/streaming/server.py | 7 ++++---
 2 files changed, 7 insertions(+), 4 deletions(-)

diff --git a/examples/stories/streaming/client.py b/examples/stories/streaming/client.py
index e584b4c1ef..a858e758c8 100644
--- a/examples/stories/streaming/client.py
+++ b/examples/stories/streaming/client.py
@@ -15,7 +15,9 @@ async def main(target: Target, *, mode: str = "auto") -> None:
     async def on_log(params: LoggingMessageNotificationParams) -> None:
         logs.append(params)
 
-    async with Client(target, mode=mode, logging_callback=on_log) as client:
+    # `log_level` is the 2026-07-28 per-request opt-in: without it a modern server
+    # sends no log notifications at all (pre-2026 servers ignore it and send anyway).
+    async with Client(target, mode=mode, logging_callback=on_log, log_level="info") as client:
         # ── progress + logging: a short countdown delivers exactly `steps` of each, in order ──
         updates: list[tuple[float, float | None, str | None]] = []
 
diff --git a/examples/stories/streaming/server.py b/examples/stories/streaming/server.py
index ced59878d7..1782a1c44a 100644
--- a/examples/stories/streaming/server.py
+++ b/examples/stories/streaming/server.py
@@ -16,9 +16,10 @@ async def countdown(steps: int, ctx: Context) -> dict[str, int]:
         try:
             for i in range(1, steps + 1):
                 await ctx.report_progress(float(i), float(steps), f"step {i}/{steps}")
-                # No non-deprecated logging helper on Context yet, so send the raw
-                # notification. `related_request_id` keeps it on this request's response
-                # stream (matters over streamable HTTP).
+                # Protocol logging is deprecated (SEP-2577), so the raw notification
+                # keeps this warning-free. On 2026-07-28+ the client only receives it
+                # because it opts in with `log_level=`; `related_request_id` keeps it on
+                # this request's response stream (matters over streamable HTTP).
                 await ctx.request_context.session.send_notification(
                     types.LoggingMessageNotification(
                         params=types.LoggingMessageNotificationParams(

From 84724a4805aa3254bbc530c411a386988369cde7 Mon Sep 17 00:00:00 2001
From: Max Isbey <224885523+maxisbey@users.noreply.github.com>
Date: Tue, 28 Jul 2026 00:30:24 +0000
Subject: [PATCH 3/4] Keep the log-level gate helper out of connection.py
 __all__

---
 src/mcp/server/connection.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/mcp/server/connection.py b/src/mcp/server/connection.py
index e1fb1383d9..a0913fe14e 100644
--- a/src/mcp/server/connection.py
+++ b/src/mcp/server/connection.py
@@ -50,7 +50,7 @@
 from mcp.shared.exceptions import MCPDeprecationWarning, NoBackChannelError
 from mcp.shared.peer import Meta, dump_params
 
-__all__ = ["Connection", "allowed_log_levels"]
+__all__ = ["Connection"]
 
 logger = logging.getLogger(__name__)
 # `Connection.log`'s `logger` parameter (public API, the spec's logger-name

From 2636cb9a2c41e3e490239c570d0b356fa0e89282 Mon Sep 17 00:00:00 2001
From: Max Isbey <224885523+maxisbey@users.noreply.github.com>
Date: Tue, 28 Jul 2026 01:10:05 +0000
Subject: [PATCH 4/4] Keep the log-delivery opt-in request-only and drop the
 mock from the logging test

A notification carries no request to opt in and no response stream to carry
a log entry, so its _meta no longer opens the per-request log gate:
_make_context passes the log-level opt-in to ServerSession only for
requests. Regression-tested over the dual-era stream loop.

Merge the opt-in and threshold interaction tests into one two-phase test
sharing a real closure callback instead of an AsyncMock.
---
 src/mcp/server/runner.py                   |  6 ++--
 tests/interaction/lowlevel/test_logging.py | 35 +++++++++-------------
 tests/server/test_runner.py                | 21 +++++++++++++
 3 files changed, 39 insertions(+), 23 deletions(-)

diff --git a/src/mcp/server/runner.py b/src/mcp/server/runner.py
index b799a00be6..26e8efbe57 100644
--- a/src/mcp/server/runner.py
+++ b/src/mcp/server/runner.py
@@ -321,8 +321,10 @@ def _make_context(
         # Per-request session: `dctx` is the request-scoped channel (auto-threads
         # its own request_id on streamable HTTP); the standalone channel is read
         # off `connection.outbound`. `related_request_id` on the public API selects.
-        # `meta` carries this request's log-level opt-in for the session's log gate.
-        session = ServerSession(dctx, self.connection, request_meta=meta)
+        # `meta` carries a request's log-level opt-in for the session's log gate. A
+        # notification has no request to opt in (and no response stream to carry
+        # the log entry), so its `_meta` never opens the gate.
+        session = ServerSession(dctx, self.connection, request_meta=meta if dctx.request_id is not None else None)
         return ServerRequestContext(
             session=session,
             lifespan_context=self.lifespan_state,
diff --git a/tests/interaction/lowlevel/test_logging.py b/tests/interaction/lowlevel/test_logging.py
index cae5eec522..f827650e73 100644
--- a/tests/interaction/lowlevel/test_logging.py
+++ b/tests/interaction/lowlevel/test_logging.py
@@ -5,8 +5,6 @@
 streamable HTTP, so plain-list collection is deterministic on every transport leg.
 """
 
-from unittest.mock import AsyncMock
-
 import mcp_types as types
 import pytest
 from inline_snapshot import snapshot
@@ -160,33 +158,28 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara
 
 
 @requirement("logging:per-request:opt-in")
-async def test_no_log_messages_are_sent_for_a_request_without_a_log_level(connect: Connect) -> None:
-    """A request whose _meta carries no io.modelcontextprotocol/logLevel gets no log messages at all.
+@requirement("logging:per-request:threshold")
+async def test_log_delivery_follows_the_per_request_log_level(connect: Connect) -> None:
+    """Without io.modelcontextprotocol/logLevel in _meta a request gets no log messages;
+    with it, only entries at or above the requested level are delivered, in order.
 
-    The handler emits at every severity, but the client never opts in, so the server must send
-    nothing: the log calls are dropped rather than delivered on some other stream.
+    The handler emits at every severity in both phases: the un-opted request receives
+    nothing (the log calls are dropped, not delivered on some other stream), and the request
+    opting in at `warning` receives warning and above.
     """
-    logging_callback = AsyncMock()
+    received: list[types.LoggingLevel] = []
 
-    async with connect(_siren_server(), logging_callback=logging_callback) as client:
-        result = await client.call_tool("siren", {})
+    async def collect(params: LoggingMessageNotificationParams) -> None:
+        received.append(params.level)
 
+    async with connect(_siren_server(), logging_callback=collect) as client:
+        result = await client.call_tool("siren", {})
     assert isinstance(result.content[0], TextContent) and result.content[0].text == "logged"
-    logging_callback.assert_not_awaited()
-
-
-@requirement("logging:per-request:threshold")
-async def test_log_messages_below_the_requested_level_are_dropped(connect: Connect) -> None:
-    """A request opting in at `warning` receives only warning and above, in order."""
-    received: list[LoggingMessageNotificationParams] = []
-
-    async def collect(params: LoggingMessageNotificationParams) -> None:
-        received.append(params)
+    assert received == []
 
     async with connect(_siren_server(), logging_callback=collect, log_level="warning") as client:
         await client.call_tool("siren", {})
-
-    assert [params.level for params in received] == ["warning", "error", "critical", "alert", "emergency"]
+    assert received == ["warning", "error", "critical", "alert", "emergency"]
 
 
 @requirement("logging:per-request:invalid-level")
diff --git a/tests/server/test_runner.py b/tests/server/test_runner.py
index eb212dafdb..9781fcc9ee 100644
--- a/tests/server/test_runner.py
+++ b/tests/server/test_runner.py
@@ -25,6 +25,7 @@
     INVALID_PARAMS,
     INVALID_REQUEST,
     LATEST_PROTOCOL_VERSION,
+    LOG_LEVEL_META_KEY,
     METHOD_NOT_FOUND,
     PROTOCOL_VERSION_META_KEY,
     SERVER_INFO_META_KEY,
@@ -1759,6 +1760,26 @@ async def on_custom(ctx: Ctx, params: NotificationParams | None) -> None:
         await handled.wait()
 
 
+@pytest.mark.anyio
+async def test_dual_era_loop_a_notifications_meta_never_opens_the_log_gate(server: SrvT):
+    """The 2026 log-delivery opt-in is a request's `_meta`: an inbound notification
+    has no request to opt in, so a handler logging in response to one - even
+    with the log-level key in the notification's own `_meta` - sends nothing."""
+    logged = anyio.Event()
+
+    async def log_it(ctx: Ctx, params: NotificationParams | None) -> None:
+        await ctx.session.send_log_message("emergency", "no request opted in")  # pyright: ignore[reportDeprecated]
+        logged.set()
+
+    server.add_notification_handler("notifications/custom", NotificationParams, log_it)
+    async with dual_era_client(server) as (client, recorder):
+        await client.send_raw_request("tools/list", _modern_params())
+        meta = {**_modern_envelope(), LOG_LEVEL_META_KEY: "debug"}
+        await client.notify("notifications/custom", {"_meta": meta})
+        await logged.wait()
+    assert [method for method, _ in recorder.notifications] == []
+
+
 @pytest.mark.anyio
 async def test_dual_era_loop_modern_server_notifications_ride_the_pipe(server: SrvT):
     """A modern handler's standalone notification reaches the client over the