S2 wave 0: conformance pin, classifier, discover handler, helpers, fixture tool, migration note

- F1: conformance harness pin -> github:modelcontextprotocol/conformance#fe3a3103;
  reconcile both expected-failures baselines (auth/scope-step-up moves to 2026-only per #337)
- D2: extract to_jsonrpc_response + aclose_shielded helpers in runner.py;
  add handler_exception_to_error_data in shared/jsonrpc_dispatcher.py (single exception->ErrorData source)
- H1: new src/mcp/shared/inbound.py — classify_inbound_request() + ERROR_CODE_HTTP_STATUS table
  (pure classifier; no I/O, no mcp.server imports; method-at-version via CLIENT_REQUESTS map)
- H2: server/discover auto-derived handler at lowlevel.Server.__init__;
  get_capabilities() params now optional; new server_info property
- H4: test_missing_capability fixture tool in everything-server (raises MISSING_REQUIRED_CLIENT_CAPABILITY)
- C1: docs/migration.md entry for stateless_http lifespan cardinality change

Part of the 2026-07-28 server-stateless conformance work (#2891).

Author: maxisbey

docs/migration.md

@@ -487,6 +487,18 @@ app = Starlette(routes=[Mount("/", app=mcp.streamable_http_app(json_response=Tru
 
 If you were mutating these via `mcp.settings` after construction (e.g., `mcp.settings.port = 9000`), pass them to `run()` / `sse_app()` / `streamable_http_app()` instead — these fields no longer exist on `Settings`. The `debug` and `log_level` parameters remain on the constructor.
 
+### `stateless_http=True`: lifespan now entered once at startup
+
+When serving streamable HTTP with `stateless_http=True`, the server's `lifespan` context manager is now entered once when `StreamableHTTPSessionManager.run()` starts, and the resulting state is shared across all requests. Previously each incoming request entered (and exited) `lifespan` independently.
+
+Lifespans that set up process-wide state (connection pools, caches, background tasks) are unaffected — they now run once instead of on every request. If your lifespan was acquiring per-connection resources, move that into the handler and register cleanup on the per-connection `exit_stack`:
+
+```python
+async def handle_call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
+    db = await ctx.connection.exit_stack.enter_async_context(open_db())
+    ...
+```
+
 ### `MCPServer.get_context()` removed
 
 `MCPServer.get_context()` has been removed. Context is now injected by the framework and passed explicitly — there is no ambient ContextVar to read from.

examples/servers/everything-server/mcp_everything_server/server.py

@@ -14,6 +14,7 @@
 from mcp.server.mcpserver import Context, MCPServer
 from mcp.server.mcpserver.prompts.base import UserMessage
 from mcp.server.streamable_http import EventCallback, EventMessage, EventStore
+from mcp.shared.exceptions import MCPError
 from mcp.types import (
     AudioContent,
     Completion,
@@ -32,6 +33,7 @@
     TextResourceContents,
     UnsubscribeRequestParams,
 )
+from mcp.types.jsonrpc import MISSING_REQUIRED_CLIENT_CAPABILITY
 from pydantic import BaseModel, Field
 
 logger = logging.getLogger(__name__)
@@ -311,6 +313,26 @@ def test_error_handling() -> str:
     raise RuntimeError("This tool intentionally returns an error for testing")
 
 
+@mcp.tool()
+async def test_missing_capability(ctx: Context) -> str:
+    """Tests that a handler-raised MISSING_REQUIRED_CLIENT_CAPABILITY surfaces as a top-level JSON-RPC error.
+
+    Requires the client to declare the ``sampling`` capability. When absent, raises
+    `MCPError` (which the tool dispatch re-raises rather than wrapping in
+    ``CallToolResult.isError``) so the conformance harness observes a protocol-level
+    error response with ``data.requiredCapabilities``.
+    """
+    client_params = ctx.session.client_params
+    sampling_declared = client_params is not None and client_params.capabilities.sampling is not None
+    if not sampling_declared:
+        raise MCPError(
+            code=MISSING_REQUIRED_CLIENT_CAPABILITY,
+            message="This tool requires the client 'sampling' capability",
+            data={"requiredCapabilities": ["sampling"]},
+        )
+    return "Client declared sampling capability; proceeding."
+
+
 @mcp.tool()
 async def test_reconnection(ctx: Context) -> str:
     """Tests SSE polling by closing stream mid-call (SEP-1699)"""

src/mcp/server/lowlevel/server.py

@@ -66,6 +66,7 @@ async def main():
 from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
 from mcp.shared.message import SessionMessage
 from mcp.shared.transport_context import TransportContext
+from mcp.shared.version import MODERN_PROTOCOL_VERSIONS
 
 logger = logging.getLogger(__name__)
 
@@ -117,6 +118,15 @@ async def _ping_handler(ctx: ServerRequestContext[Any], params: types.RequestPar
     return types.EmptyResult()
 
 
+def _package_version(package: str) -> str:
+    try:
+        return importlib_version(package)
+    except Exception:  # pragma: no cover
+        pass
+
+    return "unknown"  # pragma: no cover
+
+
 class Server(Generic[LifespanResultT]):
     def __init__(
         self,
@@ -226,6 +236,7 @@ def __init__(
 
         _spec_requests: list[tuple[str, type[BaseModel], RequestHandler[LifespanResultT, Any] | None]] = [
             ("ping", types.RequestParams, on_ping),
+            ("server/discover", types.RequestParams, self._handle_discover),
             ("prompts/list", types.PaginatedRequestParams, on_list_prompts),
             ("prompts/get", types.GetPromptRequestParams, on_get_prompt),
             ("resources/list", types.PaginatedRequestParams, on_list_resources),
@@ -309,18 +320,9 @@ def create_initialization_options(
         experimental_capabilities: dict[str, dict[str, Any]] | None = None,
     ) -> InitializationOptions:
         """Create initialization options from this server instance."""
-
-        def pkg_version(package: str) -> str:
-            try:
-                return importlib_version(package)
-            except Exception:  # pragma: no cover
-                pass
-
-            return "unknown"  # pragma: no cover
-
         return InitializationOptions(
             server_name=self.name,
-            server_version=self.version if self.version else pkg_version("mcp"),
+            server_version=self.version if self.version else _package_version("mcp"),
             title=self.title,
             description=self.description,
             capabilities=self.get_capabilities(
@@ -334,10 +336,11 @@ def pkg_version(package: str) -> str:
 
     def get_capabilities(
         self,
-        notification_options: NotificationOptions,
-        experimental_capabilities: dict[str, dict[str, Any]],
+        notification_options: NotificationOptions | None = None,
+        experimental_capabilities: dict[str, dict[str, Any]] | None = None,
     ) -> types.ServerCapabilities:
         """Convert existing handlers to a ServerCapabilities object."""
+        notification_options = notification_options or NotificationOptions()
         prompts_capability = None
         resources_capability = None
         tools_capability = None
@@ -377,6 +380,40 @@ def get_capabilities(
         )
         return capabilities
 
+    @property
+    def server_info(self) -> types.Implementation:
+        """The `serverInfo` block describing this implementation.
+
+        Derived from the constructor's identity fields. `version` falls back to
+        the installed `mcp` package version when not supplied explicitly.
+        """
+        return types.Implementation(
+            name=self.name,
+            version=self.version if self.version else _package_version("mcp"),
+            title=self.title,
+            description=self.description,
+            website_url=self.website_url,
+            icons=self.icons,
+        )
+
+    async def _handle_discover(
+        self, ctx: ServerRequestContext[LifespanResultT], params: types.RequestParams | None
+    ) -> types.DiscoverResult:
+        """Default `server/discover` handler.
+
+        Auto-derived from server state at call time, so capabilities reflect
+        whatever has been registered (constructor `on_*` kwargs and later
+        `add_request_handler` calls). Operators can replace it wholesale via
+        `add_request_handler("server/discover", ...)`. Reachability for legacy
+        peers is decided at the boundary (`types.methods`), not here.
+        """
+        return types.DiscoverResult(
+            supported_versions=list(MODERN_PROTOCOL_VERSIONS),
+            capabilities=self.get_capabilities(),
+            server_info=self.server_info,
+            instructions=self.instructions,
+        )
+
     @property
     def session_manager(self) -> StreamableHTTPSessionManager:
         """Get the StreamableHTTP session manager.

src/mcp/server/runner.py

@@ -16,7 +16,7 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Mapping
+from collections.abc import Awaitable, Mapping
 from dataclasses import dataclass, field
 from functools import partial, reduce
 from typing import TYPE_CHECKING, Any, Generic, cast
@@ -33,6 +33,7 @@
 from mcp.shared._otel import extract_trace_context, otel_span
 from mcp.shared.dispatcher import DispatchContext, Dispatcher, DispatchMiddleware, OnRequest
 from mcp.shared.exceptions import MCPError
+from mcp.shared.jsonrpc_dispatcher import handler_exception_to_error_data
 from mcp.shared.message import MessageMetadata, ServerMessageMetadata
 from mcp.shared.transport_context import TransportContext
 from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
@@ -46,6 +47,9 @@
     Implementation,
     InitializeRequestParams,
     InitializeResult,
+    JSONRPCError,
+    JSONRPCResponse,
+    RequestId,
     RequestParams,
     RequestParamsMeta,
 )
@@ -54,7 +58,7 @@
 if TYPE_CHECKING:
     from mcp.server.lowlevel.server import Server
 
-__all__ = ["CallNext", "ServerMiddleware", "ServerRunner", "otel_middleware"]
+__all__ = ["CallNext", "ServerMiddleware", "ServerRunner", "aclose_shielded", "otel_middleware", "to_jsonrpc_response"]
 
 logger = logging.getLogger(__name__)
 
@@ -64,8 +68,8 @@
 _INIT_EXEMPT: frozenset[str] = frozenset({"ping"})
 
 _EXIT_STACK_CLOSE_TIMEOUT: float = 5
-"""Bound for the shielded exit-stack unwind in `run()`; a hung cleanup
-callback must not wedge shutdown."""
+"""Bound for `aclose_shielded`'s exit-stack unwind; a hung cleanup callback
+must not wedge shutdown."""
 
 
 def _extract_meta(params: Mapping[str, Any] | None) -> RequestParamsMeta | None:
@@ -169,6 +173,47 @@ def _dump_result(result: Any) -> dict[str, Any]:
     raise TypeError(f"handler returned {type(result).__name__}; expected BaseModel, dict, or None")
 
 
+async def aclose_shielded(connection: Connection) -> None:
+    """Unwind ``connection.exit_stack`` under a shielded, bounded scope.
+
+    Called from a driver's ``finally``: the shield lets per-connection cleanup
+    callbacks run even when the driver itself is being cancelled, the
+    `_EXIT_STACK_CLOSE_TIMEOUT` bound stops a hung callback wedging shutdown,
+    and a raising callback is logged-and-swallowed so it never masks the
+    driver's own exception.
+    """
+    with anyio.move_on_after(_EXIT_STACK_CLOSE_TIMEOUT, shield=True) as scope:
+        try:
+            await connection.exit_stack.aclose()
+        except Exception:
+            logger.exception("connection exit_stack cleanup raised")
+    if scope.cancelled_caught:
+        logger.warning(
+            "connection exit_stack cleanup exceeded %s seconds; abandoning remaining callbacks",
+            _EXIT_STACK_CLOSE_TIMEOUT,
+        )
+
+
+async def to_jsonrpc_response(request_id: RequestId, coro: Awaitable[dict[str, Any]]) -> JSONRPCResponse | JSONRPCError:
+    """Await ``coro`` and wrap its outcome as the JSON-RPC reply for ``request_id``.
+
+    The exception-to-wire boundary for the request-per-call drivers
+    (`serve_one`, the modern HTTP entry). `MCPError` and `ValidationError`
+    map via the shared `handler_exception_to_error_data` ladder; any other
+    exception is logged and surfaced as `INTERNAL_ERROR` so handler internals
+    never reach the wire.
+    """
+    try:
+        result = await coro
+    except Exception as exc:
+        error = handler_exception_to_error_data(exc)
+        if error is None:
+            logger.exception("request handler raised")
+            error = ErrorData(code=INTERNAL_ERROR, message="Internal server error")
+        return JSONRPCError(jsonrpc="2.0", id=request_id, error=error)
+    return JSONRPCResponse(jsonrpc="2.0", id=request_id, result=result)
+
+
 @dataclass
 class ServerRunner(Generic[LifespanT]):
     """Per-connection orchestrator. One instance per client connection."""
@@ -205,26 +250,14 @@ async def run(self, *, task_status: anyio.abc.TaskStatus[None] = anyio.TASK_STAT
         to `dispatcher.run()`. `task_status.started()` is forwarded so callers
         can `await tg.start(runner.run)` and resume once the dispatcher is
         ready to accept requests. Once the dispatcher exits,
-        `connection.exit_stack` is unwound (shielded from outer cancellation,
-        bounded by `_EXIT_STACK_CLOSE_TIMEOUT`) so any per-connection cleanup
-        registered by handlers or middleware gets a chance to run without a
-        misbehaving callback hanging shutdown indefinitely.
+        `connection.exit_stack` is unwound via `aclose_shielded` so any
+        per-connection cleanup registered by handlers or middleware gets a
+        chance to run.
         """
         try:
             await self.dispatcher.run(self._compose_on_request(), self._on_notify, task_status=task_status)
         finally:
-            with anyio.move_on_after(_EXIT_STACK_CLOSE_TIMEOUT, shield=True) as scope:
-                try:
-                    await self.connection.exit_stack.aclose()
-                except Exception:
-                    # Raising here would mask dispatcher.run()'s exception and
-                    # crash stdio servers on normal disconnect.
-                    logger.exception("connection exit_stack cleanup raised")
-            if scope.cancelled_caught:
-                logger.warning(
-                    "connection exit_stack cleanup exceeded %s seconds; abandoning remaining callbacks",
-                    _EXIT_STACK_CLOSE_TIMEOUT,
-                )
+            await aclose_shielded(self.connection)
 
     def _compose_on_request(self) -> OnRequest:
         """Wrap `_on_request` in `dispatch_middleware`, outermost-first.