Apply review fixes to subscriptions/listen serving

- Pass the server-scoped subscription bus into the fallback Contexts built
  by programmatic call_tool/read_resource/get_prompt, so ctx.notify_* works
  there instead of raising.
- Reject subscriptions/listen with 406 when the Accept header lacks
  text/event-stream: the response is always an SSE stream, so JSON-response
  mode must not serve a content type the client never accepted.
- Release a stream's subscription slot at backlog overflow time; the
  stream's own cleanup can be wedged in a transport write that closing the
  buffer cannot wake.
- End InMemorySubscriptionBus.publish with a checkpoint so a same-task
  publish burst lets listen streams drain between events instead of
  overflowing a healthy stream's buffer.
- Describe the graceful-close result as a deliberate end rather than a
  "don't re-listen" signal: a stream ended at the overflow cap sends the
  same result and the client should re-listen.
- Move a function-body import to the top of the file and fix a stale
  comment referencing a removed property.

Author: maxisbey

docs/advanced/subscriptions.md

@@ -77,7 +77,7 @@ Down on the low-level `Server` there is no pre-wired anything — and the same p
 
 * You own the bus, so you publish to it directly: `await bus.publish(ResourceUpdated(uri=...))`. Put it wherever your handlers can reach it — module scope here, the lifespan in a bigger app.
 * `ListenHandler(bus)` is the same handler `MCPServer` registers; `on_subscriptions_listen=` is an ordinary handler slot. Don't want the SDK's semantics? Write your own handler for the slot — the spec obligations come with it.
-* `ListenHandler.close()` gracefully ends every open stream: each one receives the listen request's result as its final frame, the spec's signal that the server ended the subscription deliberately and the client shouldn't re-listen. Without it, streams end when the client disconnects.
+* `ListenHandler.close()` gracefully ends every open stream: each one receives the listen request's result as its final frame, the spec's signal that the server ended the subscription deliberately — a clean end, as opposed to the abrupt drop a client may treat as a cue to reconnect. Without it, streams end when the client disconnects.
 
 ## Recap
 

src/mcp/server/_streamable_http_modern.py

@@ -383,6 +383,13 @@ async def handle_modern_request(
         await _write(rej, scope, receive, send)
         return
 
+    if req.method == "subscriptions/listen" and not has_sse:
+        # A listen response IS a notification stream, never JSON (the
+        # json_response carve-out below), so this one method requires the
+        # SSE accept even in JSON-response mode; SSE mode gated it above.
+        await Response(status_code=406)(scope, receive, send)
+        return
+
     duplicated = find_duplicated_routing_header(request.headers.items())
     if duplicated is not None:
         # The raw carrier is the only place duplicates are visible; the classifier sees a folded mapping.

src/mcp/server/mcpserver/server.py

@@ -488,7 +488,7 @@ async def call_tool(
     ) -> CallToolResult | InputRequiredResult:
         """Call a tool by name with arguments."""
         if context is None:
-            context = Context(mcp_server=self)
+            context = Context(mcp_server=self, subscriptions=self._subscriptions)
         return await self._tool_manager.call_tool(name, arguments, context, convert_result=True)
 
     async def list_resources(self) -> list[MCPResource]:
@@ -540,7 +540,7 @@ async def read_resource(
             ResourceError: If template creation or resource reading fails.
         """
         if context is None:
-            context = Context(mcp_server=self)
+            context = Context(mcp_server=self, subscriptions=self._subscriptions)
         resource = await self._resource_manager.get_resource(uri, context)
         if isinstance(resource, InputRequiredResult):
             return resource
@@ -1260,7 +1260,7 @@ async def get_prompt(
         carrying the echoed opaque state.
         """
         if context is None:
-            context = Context(mcp_server=self)
+            context = Context(mcp_server=self, subscriptions=self._subscriptions)
         try:
             prompt = self._prompt_manager.get_prompt(name)
             if not prompt:

src/mcp/server/subscriptions.py

@@ -28,6 +28,7 @@
 from typing import Any, Protocol
 
 import anyio
+import anyio.lowlevel
 import anyio.streams.memory
 from mcp_types import (
     INTERNAL_ERROR,
@@ -118,13 +119,16 @@ async def publish(self, event: ServerEvent) -> None:
         """Deliver `event` to every subscribed listener.
 
         A raising listener is logged and skipped: one bad listener must not
-        starve the others or fail the publishing handler.
+        starve the others or fail the publishing handler. Ends with a
+        checkpoint so a burst of publishes from one task lets listen streams
+        drain between events instead of overflowing their buffers unread.
         """
         for listener in list(self._listeners.values()):
             try:
                 listener(event)
             except Exception:  # fan-out boundary: isolate listeners from each other
                 logger.exception("subscription listener raised; continuing")
+        await anyio.lowlevel.checkpoint()
 
     def subscribe(self, listener: Callable[[ServerEvent], None]) -> Callable[[], None]:
         """Register `listener` and return an idempotent unsubscribe callable."""
@@ -235,6 +239,10 @@ def deliver(event: ServerEvent) -> None:
                     pass
                 except anyio.WouldBlock:
                     logger.warning("listen stream %r backlog full; ending the stream", subscription_id)
+                    # Release the subscription slot now: the handler's own
+                    # cleanup can be wedged in a transport write that closing
+                    # this buffer cannot wake (a client that stopped reading).
+                    self._streams.discard(send)
                     send.close()
 
         # Subscribe before sending the ack so an event published while the
@@ -267,8 +275,9 @@ def close(self) -> None:
         Each stream then drains its buffered events and sends its
         `SubscriptionsListenResult` (stamped with the subscription id) as the
         final frame from its own handler task - the spec's graceful closure
-        flow, signalling clients not to re-listen. This method only initiates
-        that; it does not wait for the streams to finish flushing.
+        flow, telling clients the stream ended deliberately rather than
+        dropping. This method only initiates that; it does not wait for the
+        streams to finish flushing.
         """
         for stream in list(self._streams):
             stream.close()

tests/docs_src/test_subscriptions.py

@@ -128,7 +128,7 @@ async def listen() -> None:
             assert isinstance(updated, types.ResourceUpdatedNotification)
             assert updated.params.uri == "note://todo"
 
-            # `mcp.subscriptions` / the bus is also the publish surface outside a
+            # The bus you constructed is also the publish surface outside a
             # request; an unrequested kind never reaches this stream.
             await tutorial002.bus.publish(ToolsListChanged())
             await client.call_tool("edit_note", {"name": "todo", "text": "done"})

tests/server/lowlevel/test_server_discover.py

@@ -12,7 +12,7 @@
 import pytest
 from mcp_types.version import MODERN_PROTOCOL_VERSIONS
 
-from mcp.server import Server, ServerRequestContext
+from mcp.server import NotificationOptions, Server, ServerRequestContext
 
 
 # `Server._handle_discover` reads only `ctx.protocol_version` (capabilities are
@@ -216,7 +216,5 @@ async def list_tools(
     legacy = server.get_capabilities()
     assert legacy.tools is not None and legacy.tools.list_changed is False
 
-    from mcp.server import NotificationOptions
-
     opted_in = server.get_capabilities(NotificationOptions(tools_changed=True))
     assert opted_in.tools is not None and opted_in.tools.list_changed is True

tests/server/mcpserver/test_server.py

@@ -2279,6 +2279,37 @@ async def touch(ctx: Context) -> str:
     assert seen == [ToolsListChanged(), PromptsListChanged(), ResourcesListChanged(), ResourceUpdated(uri="r://x")]
 
 
+async def test_programmatic_entry_points_carry_the_subscription_bus() -> None:
+    """`ctx.notify_*` works when tools, resources, and prompts are invoked
+    programmatically (no wire request): the server-scoped bus rides along in
+    the fallback Context."""
+    bus = InMemorySubscriptionBus()
+    mcp = MCPServer(subscriptions=bus)
+    seen: list[ServerEvent] = []
+    bus.subscribe(seen.append)
+
+    @mcp.tool()
+    async def touch_tools(ctx: Context) -> str:
+        await ctx.notify_tools_changed()
+        return "ok"
+
+    @mcp.resource("res://{name}")
+    async def thing(name: str, ctx: Context) -> str:
+        await ctx.notify_resources_changed()
+        return "data"
+
+    @mcp.prompt()
+    async def ask(ctx: Context) -> str:
+        await ctx.notify_prompts_changed()
+        return "question"
+
+    await mcp.call_tool("touch_tools", {})
+    await mcp.read_resource("res://thing")
+    await mcp.get_prompt("ask")
+
+    assert seen == [ToolsListChanged(), ResourcesListChanged(), PromptsListChanged()]
+
+
 def test_context_mcp_server_outside_request_raises() -> None:
     with pytest.raises(ValueError, match="outside of a request"):
         _ = Context().mcp_server

tests/server/test_streamable_http_modern.py

@@ -1025,14 +1025,9 @@ def subscribe(self, listener: Callable[[ServerEvent], None]) -> Callable[[], Non
         return unsubscribe
 
 
-async def test_json_response_mode_still_streams_subscriptions_listen() -> None:
-    """SDK-defined (TypeScript/Go parity): a listen response IS a notification
-    stream, so `json_response=True` does not apply to it - the request takes
-    the SSE path, acks first, and ends with the stamped result on close()."""
-    bus = _OpenSignalBus()
-    handler = ListenHandler(bus)
-    server = Server("test", on_subscriptions_listen=handler)
-    body = {
+def _listen_body() -> dict[str, Any]:
+    """A minimal valid 2026-07-28 `subscriptions/listen` request body."""
+    return {
         "jsonrpc": "2.0",
         "id": 9,
         "method": "subscriptions/listen",
@@ -1046,6 +1041,26 @@ async def test_json_response_mode_still_streams_subscriptions_listen() -> None:
         },
     }
 
+
+async def test_subscriptions_listen_requires_the_sse_accept_even_in_json_mode() -> None:
+    """SDK-defined: a listen response is always SSE, so a request whose Accept
+    lacks `text/event-stream` is rejected with 406 rather than served a content
+    type it never accepted - JSON-response mode included."""
+    server = Server("test", on_subscriptions_listen=ListenHandler(InMemorySubscriptionBus()))
+    async with _asgi_client(server, json_response=True, accept="application/json") as http:
+        response = await http.post("/mcp", json=_listen_body(), headers={MCP_METHOD_HEADER: "subscriptions/listen"})
+    assert response.status_code == 406
+
+
+async def test_json_response_mode_still_streams_subscriptions_listen() -> None:
+    """SDK-defined (TypeScript/Go parity): a listen response IS a notification
+    stream, so `json_response=True` does not apply to it - the request takes
+    the SSE path, acks first, and ends with the stamped result on close()."""
+    bus = _OpenSignalBus()
+    handler = ListenHandler(bus)
+    server = Server("test", on_subscriptions_listen=handler)
+    body = _listen_body()
+
     responses: list[httpx.Response] = []
     async with _asgi_client(server, json_response=True) as http:
         async with anyio.create_task_group() as tg:

tests/server/test_subscriptions.py

@@ -338,14 +338,37 @@ async def run() -> None:
         handler.close()
 
 
+class _GatedSession(_RecordingSession):
+    """Lets the ack through, then wedges event sends until released - a client
+    that stopped reading the transport."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.wedged = anyio.Event()
+        self.release = anyio.Event()
+
+    async def send_notification(
+        self, notification: ServerNotification, related_request_id: RequestId | None = None
+    ) -> None:
+        if self.sent:  # the ack is the first frame; only event sends wedge
+            self.wedged.set()
+            await self.release.wait()
+        await super().send_notification(notification, related_request_id)
+
+
 @pytest.mark.anyio
-async def test_backlog_overflow_ends_the_stream() -> None:
+async def test_backlog_overflow_ends_the_stream_and_frees_its_slot() -> None:
     """SDK-defined: a stream whose client stopped reading is ended at
-    `max_buffered_events` rather than buffering forever; the client re-listens."""
+    `max_buffered_events` rather than buffering forever. The subscription slot
+    frees at overflow time - the stream's own cleanup may be wedged in a
+    transport write nothing can wake - and the backlog still drains into the
+    stamped graceful result once that write completes."""
     bus = InMemorySubscriptionBus()
-    handler = ListenHandler(bus, max_buffered_events=1)
-    session = _RecordingSession()
+    handler = ListenHandler(bus, max_subscriptions=1, max_buffered_events=1)
+    session = _GatedSession()
     results: list[SubscriptionsListenResult] = []
+    late_session = _RecordingSession()
+    late_results: list[SubscriptionsListenResult] = []
 
     async with anyio.create_task_group() as tg:
 
@@ -355,11 +378,49 @@ async def run() -> None:
         tg.start_soon(run)
         await session.wait_for(1)
 
-        # Two publishes before the handler task resumes: the first fills the
-        # one-slot buffer, the second overflows and ends the stream.
-        await bus.publish(ToolsListChanged())
-        await bus.publish(ToolsListChanged())
+        await bus.publish(ToolsListChanged())  # consumed, then wedged mid-send
+        with anyio.fail_after(5):
+            await session.wedged.wait()
+        await bus.publish(ToolsListChanged())  # fills the one-slot buffer
+        await bus.publish(ToolsListChanged())  # overflows: the stream is ended
+
+        async def run_late() -> None:
+            late_results.append(await handler(_ctx(late_session, request_id=8), _params(tools_list_changed=True)))
+
+        # The ended stream's slot is free immediately - a new listen does not
+        # wait for the wedged write to die with its connection.
+        tg.start_soon(run_late)
+        await late_session.wait_for(1)
+
+        session.release.set()
+        handler.close()
 
     delivered = [notification for notification, _ in session.sent[1:]]
-    assert len(delivered) == 1  # the buffered event still drained
-    assert results[0].meta is not None  # the stream ended with the stamped result
+    assert len(delivered) == 2  # the wedged event and the buffered one still drained
+    assert results[0].meta == {SUBSCRIPTION_ID_META_KEY: 7}
+    assert late_results[0].meta == {SUBSCRIPTION_ID_META_KEY: 8}
+
+
+@pytest.mark.anyio
+async def test_same_task_publish_burst_does_not_overflow_a_healthy_stream() -> None:
+    """SDK-defined: `publish` ends with a checkpoint, so a burst of events from
+    one task (no yields of its own) lets a reading stream drain between
+    publishes instead of deterministically overflowing the buffer."""
+    bus = InMemorySubscriptionBus()
+    handler = ListenHandler(bus, max_buffered_events=99)
+    session = _RecordingSession()
+
+    async with anyio.create_task_group() as tg:
+
+        async def run() -> None:
+            await handler(_ctx(session), _params(tools_list_changed=True))
+
+        tg.start_soon(run)
+        await session.wait_for(1)
+
+        for _ in range(100):
+            await bus.publish(ToolsListChanged())
+        await session.wait_for(101)
+        handler.close()
+
+    assert len(session.sent) == 101  # the ack plus every event in the burst