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.

Author: maxisbey

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,

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")

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