backport: harness H4 — build_streamable_http_app (no-auth), connect_over_streamable_http, client_via_http; gate: 3 smoke legs pass

Also:
- pyproject: add [tool.inline-snapshot] default-flags=["disable"] (matches main; without
  it -n0 runs use inline-snapshot active mode whose pydantic comparison mishandles
  extra="allow" models)
- conftest: suppress PytestUnraisableExceptionWarning/ResourceWarning — v1 streamable-HTTP
  server transport leaks memory streams on teardown (e.g. _handle_get_request only closes
  sse_stream_reader on the exception path); fixes are src/-side on main, out of scope here
- test_ping.py: convert to v1 decorator pattern using session-access pattern B (request_ctx
  contextvar) per the file→pattern assignment

Gate: tests/interaction/lowlevel/test_ping.py 6/6 pass (both tests × 3 transport legs).

Author: maxisbey

pyproject.toml

@@ -155,6 +155,11 @@ members = ["examples/clients/*", "examples/servers/*", "examples/snippets"]
 [tool.uv.sources]
 mcp = { workspace = true }
 
+[tool.inline-snapshot]
+# `snapshot(x)` becomes the identity (plain `==`); regenerate with `--inline-snapshot=fix`.
+default-flags = ["disable"]
+format-command = "ruff format --stdin-filename {filename}"
+
 [tool.pytest.ini_options]
 log_cli = true
 xfail_strict = true

tests/interaction/_connect.py

@@ -27,6 +27,7 @@
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider, TokenVerifier
 from mcp.server.auth.settings import AuthSettings
 from mcp.server.fastmcp import FastMCP
+from mcp.server.fastmcp.server import StreamableHTTPASGIApp
 from mcp.server.sse import SseServerTransport
 from mcp.server.streamable_http import EventStore
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
@@ -114,6 +115,51 @@ async def connect_in_memory(
         yield session
 
 
+def build_streamable_http_app(
+    server: Server[Any] | FastMCP,
+    *,
+    stateless_http: bool = False,
+    json_response: bool = False,
+    event_store: EventStore | None = None,
+    retry_interval: int | None = None,
+    transport_security: TransportSecuritySettings | None = NO_DNS_REBINDING_PROTECTION,
+    auth: AuthSettings | None = None,
+    token_verifier: TokenVerifier | None = None,
+    auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
+) -> tuple[Starlette, StreamableHTTPSessionManager]:
+    """Assemble a streamable-HTTP Starlette app for either server flavour.
+
+    v1's lowlevel `Server` has no `streamable_http_app()`; this follows
+    `FastMCP.streamable_http_app()` (`mcp/server/fastmcp/server.py`) so behaviour matches what a
+    v1 user would get from `FastMCP(..., **knobs).streamable_http_app()`. Returns the live
+    `StreamableHTTPSessionManager` alongside the app so the caller can enter `manager.run()`
+    (the in-process bridge does not drive Starlette lifespan) and so tests can reach
+    `manager._server_instances`.
+
+    `/mcp` is mounted via `Route(path, endpoint=<class instance>)` with no `methods=`, exactly
+    as FastMCP does — Starlette treats a class-instance endpoint as raw ASGI and matches all
+    verbs, which is what the transport requires.
+    """
+    manager = StreamableHTTPSessionManager(
+        app=_lowlevel(server),
+        event_store=event_store,
+        json_response=json_response,
+        stateless=stateless_http,
+        security_settings=transport_security,
+        retry_interval=retry_interval,
+    )
+    asgi = StreamableHTTPASGIApp(manager)
+
+    routes: list[Route] = []
+    # Auth routing (middleware, AS routes, RequireAuthMiddleware wrap, PRM routes) is added in
+    # H5; until then `auth` / `token_verifier` / `auth_server_provider` are accepted but ignored
+    # so callers that don't pass them work today.
+    assert auth is None and token_verifier is None and auth_server_provider is None, "auth branch lands in H5"
+    routes.append(Route("/mcp", endpoint=asgi))
+
+    return Starlette(routes=routes), manager
+
+
 @asynccontextmanager
 async def connect_over_streamable_http(
     server: Server[Any] | FastMCP,
@@ -137,28 +183,31 @@ async def connect_over_streamable_http(
     server modes, and the resumability tests pass an `event_store` (with `retry_interval=0` so
     the client's reconnection wait is a no-op).
     """
-    app = server.streamable_http_app(
+    app, manager = build_streamable_http_app(
+        server,
         stateless_http=stateless_http,
         json_response=json_response,
         event_store=event_store,
         retry_interval=retry_interval,
-        transport_security=NO_DNS_REBINDING_PROTECTION,
     )
     async with (
-        server.session_manager.run(),
+        manager.run(),
         httpx.AsyncClient(transport=StreamingASGITransport(app), base_url=BASE_URL) as http_client,
-        Client(  # noqa: F821  -- body rewritten in H4
-            streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client),
+        streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client) as (read, write, _get_session_id),
+        ClientSession(
+            read,
+            write,
             read_timeout_seconds=read_timeout_seconds,
             sampling_callback=sampling_callback,
             list_roots_callback=list_roots_callback,
             logging_callback=logging_callback,
             message_handler=message_handler,
             client_info=client_info,
             elicitation_callback=elicitation_callback,
-        ) as client,
+        ) as session,
     ):
-        yield client
+        await session.initialize()
+        yield session
 
 
 @asynccontextmanager
@@ -219,18 +268,22 @@ async def client_via_http(
 ) -> AsyncIterator[ClientSession]:
     """Connect a `ClientSession` over an already-mounted streamable HTTP app.
 
-    Use with `mounted_app(...)` so several `Client`s share the one session manager, or so a
-    client-driven assertion can sit alongside raw-httpx assertions in the same test. The
-    underlying `httpx.AsyncClient` is left open when the `Client` exits.
+    Use with `mounted_app(...)` so several `ClientSession`s share the one session manager, or
+    so a client-driven assertion can sit alongside raw-httpx assertions in the same test. The
+    underlying `httpx.AsyncClient` is left open when the session exits.
     """
-    transport = streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client)
-    async with Client(  # noqa: F821  -- body rewritten in H4
-        transport,
-        logging_callback=logging_callback,
-        message_handler=message_handler,
-        elicitation_callback=elicitation_callback,
-    ) as client:
-        yield client
+    async with (
+        streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client) as (read, write, _get_session_id),
+        ClientSession(
+            read,
+            write,
+            logging_callback=logging_callback,
+            message_handler=message_handler,
+            elicitation_callback=elicitation_callback,
+        ) as session,
+    ):
+        await session.initialize()
+        yield session
 
 
 def parse_sse_messages(events: Iterable[ServerSentEvent]) -> list[JSONRPCMessage]:

tests/interaction/conftest.py

@@ -9,6 +9,14 @@ def pytest_configure(config: pytest.Config) -> None:
     config.addinivalue_line(
         "markers", "requirement(id): tag a test as covering an entry in tests/interaction/_requirements.py"
     )
+    # v1's streamable-HTTP server transport leaks a handful of anyio memory streams on teardown
+    # (e.g. `_handle_get_request` only closes `sse_stream_reader` on the exception path; the
+    # session manager's per-session task-group cancel can race the per-request cleanup). v1's own
+    # tests run the transport in a separate process and so never observe these `__del__`-time
+    # ResourceWarnings; running in-process via the streaming bridge does. The fixes live in `src/`
+    # on `main` and are out of scope for this tests-only backport, so suppress here.
+    config.addinivalue_line("filterwarnings", "ignore::pytest.PytestUnraisableExceptionWarning")
+    config.addinivalue_line("filterwarnings", "ignore::ResourceWarning")
 
 
 _FACTORIES: dict[str, Connect] = {

tests/interaction/lowlevel/test_ping.py

@@ -1,10 +1,20 @@
-"""Ping interactions against the low-level Server, driven through the public Client API."""
+"""Ping interactions against the low-level Server, driven through the public ClientSession API.
+
+This file reaches the server session via the module-level `request_ctx` contextvar (pattern B
+from the v1 backport's session-access spread). That contextvar is the mechanism behind
+`Server.request_context`; reading it directly is a public-module-level name a v1 user can
+import, and exercising it here covers the contextvar path the eventual v2 compatibility shims
+must preserve.
+"""
+
+from typing import Any
 
 import pytest
 from inline_snapshot import snapshot
 
 from mcp import types
-from mcp.server import Server, ServerRequestContext
+from mcp.server import Server
+from mcp.server.lowlevel.server import request_ctx
 from mcp.types import CallToolResult, EmptyResult, TextContent
 from tests.interaction._connect import Connect
 from tests.interaction._requirements import requirement
@@ -16,7 +26,7 @@
 @requirement("ping:client-to-server")
 async def test_client_ping_returns_empty_result(connect: Connect) -> None:
     """A client ping is answered with an empty result, even by a server with no handlers."""
-    server = Server("silent")
+    server: Server[None] = Server("silent")
 
     async with connect(server) as client:
         result = await client.send_ping()
@@ -32,21 +42,18 @@ async def test_server_ping_returns_empty_result(connect: Connect) -> None:
     The tool returns the type of the ping response, proving the round trip completed inside
     the handler before the tool result was produced.
     """
+    server: Server[None] = Server("pinger")
 
-    async def list_tools(
-        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
-    ) -> types.ListToolsResult:
-        return types.ListToolsResult(
-            tools=[types.Tool(name="ping_back", description="Ping the client.", inputSchema={"type": "object"})]
-        )
+    @server.list_tools()
+    async def list_tools() -> list[types.Tool]:
+        return [types.Tool(name="ping_back", description="Ping the client.", inputSchema={"type": "object"})]
 
-    async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
-        assert params.name == "ping_back"
-        pong = await ctx.session.send_ping()
+    @server.call_tool()
+    async def call_tool(name: str, arguments: dict[str, Any]) -> CallToolResult:
+        assert name == "ping_back"
+        pong = await request_ctx.get().session.send_ping()
         return CallToolResult(content=[TextContent(type="text", text=type(pong).__name__)])
 
-    server = Server("pinger", on_list_tools=list_tools, on_call_tool=call_tool)
-
     async with connect(server) as client:
         result = await client.call_tool("ping_back", {})