Merge origin/main (in-process transport tests #2764/2765/2767)

Conflict in tests/shared/test_streamable_http.py: kept the gh-106749
heal at the resumption-test cancel site, took main's in-process
make_client/BASE_URL.

Author: maxisbey

AGENTS.md

@@ -65,6 +65,11 @@
   tests. Don't silence warnings from your own code; fix the underlying cause.
   Scoped `ignore::` entries for upstream libraries are acceptable in
   `pyproject.toml` with a comment explaining why.
+- New features from the 2026-07-28 spec must have a matching test in the
+  [conformance suite](https://github.com/modelcontextprotocol/conformance)
+  that passes against this SDK (CI runs it via
+  `.github/workflows/conformance.yml`). If no matching test exists, stop and
+  tell the user so they can raise an issue on the conformance repo.
 
 ### Coverage
 

src/mcp/server/streamable_http.py

@@ -207,7 +207,7 @@ def close_sse_stream(self, request_id: RequestId) -> None:
             send_stream.close()
             receive_stream.close()
 
-    def close_standalone_sse_stream(self) -> None:  # pragma: no cover
+    def close_standalone_sse_stream(self) -> None:
         """Close the standalone GET SSE stream, triggering client reconnection.
 
         This method closes the HTTP connection for the standalone GET stream used
@@ -221,8 +221,6 @@ def close_standalone_sse_stream(self) -> None:  # pragma: no cover
             This is a no-op if there is no active standalone SSE stream.
             Requires event_store to be configured for events to be stored during
             the disconnect.
-            Currently, client reconnection for standalone GET streams is NOT
-            implemented - this is a known gap (see test_standalone_get_stream_reconnection).
         """
         self.close_sse_stream(GET_STREAM_KEY)
 
@@ -245,7 +243,7 @@ def _create_session_message(
             async def close_stream_callback() -> None:
                 self.close_sse_stream(request_id)
 
-            async def close_standalone_stream_callback() -> None:  # pragma: no cover
+            async def close_standalone_stream_callback() -> None:
                 self.close_standalone_sse_stream()
 
             metadata = ServerMessageMetadata(
@@ -421,7 +419,7 @@ async def _validate_accept_header(self, request: Request, scope: Scope, send: Se
         has_json, has_sse = self._check_accept_headers(request)
         if self.is_json_response_enabled:
             # For JSON-only responses, only require application/json
-            if not has_json:  # pragma: no cover
+            if not has_json:
                 response = self._create_error_response(
                     "Not Acceptable: Client must accept application/json",
                     HTTPStatus.NOT_ACCEPTABLE,
@@ -672,7 +670,7 @@ async def _handle_get_request(self, request: Request, send: Send) -> None:
             await response(request.scope, request.receive, send)
             return
 
-        if not await self._validate_request_headers(request, send):  # pragma: no cover
+        if not await self._validate_request_headers(request, send):
             return
 
         # Handle resumability: check for Last-Event-ID header

tests/client/test_http_unicode.py

@@ -4,11 +4,10 @@
 (server→client and client→server) using the streamable HTTP transport.
 """
 
-import multiprocessing
-import socket
-from collections.abc import AsyncGenerator, Generator
+from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 
+import httpx
 import pytest
 from starlette.applications import Starlette
 from starlette.routing import Mount
@@ -19,7 +18,10 @@
 from mcp.server import Server, ServerRequestContext
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.types import TextContent, Tool
-from tests.test_helpers import wait_for_server
+from tests.interaction.transports import StreamingASGITransport
+
+# The in-process app is mounted at this origin purely so URLs are well-formed; nothing listens here.
+BASE_URL = "http://127.0.0.1:8000"
 
 # Test constants with various Unicode characters
 UNICODE_TEST_STRINGS = {
@@ -41,197 +43,131 @@
 }
 
 
-def run_unicode_server(port: int) -> None:  # pragma: no cover
-    """Run the Unicode test server in a separate process."""
-    import uvicorn
-
-    # Need to recreate the server setup in this process
-    async def handle_list_tools(
-        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
-    ) -> types.ListToolsResult:
-        return types.ListToolsResult(
-            tools=[
-                Tool(
-                    name="echo_unicode",
-                    description="🔤 Echo Unicode text - Hello 👋 World 🌍 - Testing 🧪 Unicode ✨",
-                    input_schema={
-                        "type": "object",
-                        "properties": {
-                            "text": {"type": "string", "description": "Text to echo back"},
-                        },
-                        "required": ["text"],
+async def handle_list_tools(
+    ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+) -> types.ListToolsResult:
+    return types.ListToolsResult(
+        tools=[
+            Tool(
+                name="echo_unicode",
+                description="🔤 Echo Unicode text - Hello 👋 World 🌍 - Testing 🧪 Unicode ✨",
+                input_schema={
+                    "type": "object",
+                    "properties": {
+                        "text": {"type": "string", "description": "Text to echo back"},
                     },
-                ),
-            ]
-        )
-
-    async def handle_call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> types.CallToolResult:
-        if params.name == "echo_unicode":
-            text = params.arguments.get("text", "") if params.arguments else ""
-            return types.CallToolResult(
-                content=[
-                    TextContent(
-                        type="text",
-                        text=f"Echo: {text}",
-                    )
-                ]
+                    "required": ["text"],
+                },
+            ),
+        ]
+    )
+
+
+async def handle_call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> types.CallToolResult:
+    assert params.name == "echo_unicode"
+    assert params.arguments is not None
+    return types.CallToolResult(content=[TextContent(type="text", text=f"Echo: {params.arguments['text']}")])
+
+
+async def handle_list_prompts(
+    ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+) -> types.ListPromptsResult:
+    return types.ListPromptsResult(
+        prompts=[
+            types.Prompt(
+                name="unicode_prompt",
+                description="Unicode prompt - Слой хранилища, где располагаются",
+                arguments=[],
             )
-        else:
-            raise ValueError(f"Unknown tool: {params.name}")
-
-    async def handle_list_prompts(
-        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
-    ) -> types.ListPromptsResult:
-        return types.ListPromptsResult(
-            prompts=[
-                types.Prompt(
-                    name="unicode_prompt",
-                    description="Unicode prompt - Слой хранилища, где располагаются",
-                    arguments=[],
-                )
-            ]
-        )
-
-    async def handle_get_prompt(
-        ctx: ServerRequestContext, params: types.GetPromptRequestParams
-    ) -> types.GetPromptResult:
-        if params.name == "unicode_prompt":
-            return types.GetPromptResult(
-                messages=[
-                    types.PromptMessage(
-                        role="user",
-                        content=types.TextContent(
-                            type="text",
-                            text="Hello世界🌍Привет안녕مرحباשלום",
-                        ),
-                    )
-                ]
+        ]
+    )
+
+
+async def handle_get_prompt(ctx: ServerRequestContext, params: types.GetPromptRequestParams) -> types.GetPromptResult:
+    assert params.name == "unicode_prompt"
+    return types.GetPromptResult(
+        messages=[
+            types.PromptMessage(
+                role="user",
+                content=types.TextContent(type="text", text="Hello世界🌍Привет안녕مرحباשלום"),
             )
-        raise ValueError(f"Unknown prompt: {params.name}")
+        ]
+    )
 
+
+@asynccontextmanager
+async def unicode_session() -> AsyncIterator[ClientSession]:
+    """Yield an initialized ClientSession speaking streamable HTTP (SSE responses) to the
+    Unicode test server, entirely in process."""
     server = Server(
         name="unicode_test_server",
         on_list_tools=handle_list_tools,
         on_call_tool=handle_call_tool,
         on_list_prompts=handle_list_prompts,
         on_get_prompt=handle_get_prompt,
     )
-
-    # Create the session manager
-    session_manager = StreamableHTTPSessionManager(
-        app=server,
-        json_response=False,  # Use SSE for testing
-    )
-
-    @asynccontextmanager
-    async def lifespan(app: Starlette) -> AsyncGenerator[None, None]:
-        async with session_manager.run():
-            yield
-
-    # Create an ASGI application
-    app = Starlette(
-        debug=True,
-        routes=[
-            Mount("/mcp", app=session_manager.handle_request),
-        ],
-        lifespan=lifespan,
-    )
-
-    # Run the server
-    config = uvicorn.Config(
-        app=app,
-        host="127.0.0.1",
-        port=port,
-        log_level="error",
-    )
-    uvicorn_server = uvicorn.Server(config)
-    uvicorn_server.run()
-
-
-@pytest.fixture
-def unicode_server_port() -> int:
-    """Find an available port for the Unicode test server."""
-    with socket.socket() as s:
-        s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
-
-
-@pytest.fixture
-def running_unicode_server(unicode_server_port: int) -> Generator[str, None, None]:
-    """Start a Unicode test server in a separate process."""
-    proc = multiprocessing.Process(target=run_unicode_server, kwargs={"port": unicode_server_port}, daemon=True)
-    proc.start()
-
-    # Wait for server to be ready
-    wait_for_server(unicode_server_port)
-
-    try:
-        yield f"http://127.0.0.1:{unicode_server_port}"
-    finally:
-        # Clean up - try graceful termination first
-        proc.terminate()
-        proc.join(timeout=2)
-        if proc.is_alive():  # pragma: no cover
-            proc.kill()
-            proc.join(timeout=1)
+    # SSE response mode, so Unicode rides the SSE event encoding rather than a plain JSON body.
+    session_manager = StreamableHTTPSessionManager(app=server, json_response=False)
+    app = Starlette(routes=[Mount("/mcp", app=session_manager.handle_request)])
+
+    async with (
+        session_manager.run(),
+        # follow_redirects matches the SDK's own client factory; Starlette's Mount 307-redirects
+        # the bare /mcp path to /mcp/.
+        httpx.AsyncClient(
+            transport=StreamingASGITransport(app), base_url=BASE_URL, follow_redirects=True
+        ) as http_client,
+        streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client) as (read_stream, write_stream),
+        ClientSession(read_stream, write_stream) as session,
+    ):
+        await session.initialize()
+        yield session
 
 
 @pytest.mark.anyio
-async def test_streamable_http_client_unicode_tool_call(running_unicode_server: str) -> None:
+async def test_streamable_http_client_unicode_tool_call() -> None:
     """Test that Unicode text is correctly handled in tool calls via streamable HTTP."""
-    base_url = running_unicode_server
-    endpoint_url = f"{base_url}/mcp"
-
-    async with streamable_http_client(endpoint_url) as (read_stream, write_stream):
-        async with ClientSession(read_stream, write_stream) as session:
-            await session.initialize()
-
-            # Test 1: List tools (server→client Unicode in descriptions)
-            tools = await session.list_tools()
-            assert len(tools.tools) == 1
+    async with unicode_session() as session:
+        # Test 1: List tools (server→client Unicode in descriptions)
+        tools = await session.list_tools()
+        assert len(tools.tools) == 1
 
-            # Check Unicode in tool descriptions
-            echo_tool = tools.tools[0]
-            assert echo_tool.name == "echo_unicode"
-            assert echo_tool.description is not None
-            assert "🔤" in echo_tool.description
-            assert "👋" in echo_tool.description
+        # Check Unicode in tool descriptions
+        echo_tool = tools.tools[0]
+        assert echo_tool.name == "echo_unicode"
+        assert echo_tool.description is not None
+        assert "🔤" in echo_tool.description
+        assert "👋" in echo_tool.description
 
-            # Test 2: Send Unicode text in tool call (client→server→client)
-            for test_name, test_string in UNICODE_TEST_STRINGS.items():
-                result = await session.call_tool("echo_unicode", arguments={"text": test_string})
+        # Test 2: Send Unicode text in tool call (client→server→client)
+        for test_name, test_string in UNICODE_TEST_STRINGS.items():
+            result = await session.call_tool("echo_unicode", arguments={"text": test_string})
 
-                # Verify server correctly received and echoed back Unicode
-                assert len(result.content) == 1
-                content = result.content[0]
-                assert content.type == "text"
-                assert f"Echo: {test_string}" == content.text, f"Failed for {test_name}"
+            # Verify server correctly received and echoed back Unicode
+            assert len(result.content) == 1
+            content = result.content[0]
+            assert content.type == "text"
+            assert f"Echo: {test_string}" == content.text, f"Failed for {test_name}"
 
 
 @pytest.mark.anyio
-async def test_streamable_http_client_unicode_prompts(running_unicode_server: str) -> None:
+async def test_streamable_http_client_unicode_prompts() -> None:
     """Test that Unicode text is correctly handled in prompts via streamable HTTP."""
-    base_url = running_unicode_server
-    endpoint_url = f"{base_url}/mcp"
-
-    async with streamable_http_client(endpoint_url) as (read_stream, write_stream):
-        async with ClientSession(read_stream, write_stream) as session:
-            await session.initialize()
-
-            # Test 1: List prompts (server→client Unicode in descriptions)
-            prompts = await session.list_prompts()
-            assert len(prompts.prompts) == 1
-
-            prompt = prompts.prompts[0]
-            assert prompt.name == "unicode_prompt"
-            assert prompt.description is not None
-            assert "Слой хранилища, где располагаются" in prompt.description
-
-            # Test 2: Get prompt with Unicode content (server→client)
-            result = await session.get_prompt("unicode_prompt", arguments={})
-            assert len(result.messages) == 1
-
-            message = result.messages[0]
-            assert message.role == "user"
-            assert message.content.type == "text"
-            assert message.content.text == "Hello世界🌍Привет안녕مرحباשלום"
+    async with unicode_session() as session:
+        # Test 1: List prompts (server→client Unicode in descriptions)
+        prompts = await session.list_prompts()
+        assert len(prompts.prompts) == 1
+
+        prompt = prompts.prompts[0]
+        assert prompt.name == "unicode_prompt"
+        assert prompt.description is not None
+        assert "Слой хранилища, где располагаются" in prompt.description
+
+        # Test 2: Get prompt with Unicode content (server→client)
+        result = await session.get_prompt("unicode_prompt", arguments={})
+        assert len(result.messages) == 1
+
+        message = result.messages[0]
+        assert message.role == "user"
+        assert message.content.type == "text"
+        assert message.content.text == "Hello世界🌍Привет안녕مرحباשלום"

tests/interaction/transports/__init__.py

@@ -0,0 +1,9 @@
+"""Transport-specific interaction tests, and the in-process streaming bridge they are built on.
+
+`StreamingASGITransport` is re-exported here as the sanctioned import point for test code
+outside this suite (the bridge module itself is suite-private).
+"""
+
+from tests.interaction.transports._bridge import StreamingASGITransport
+
+__all__ = ["StreamingASGITransport"]