test: regression test for initialize hang on unexpected content-type

tests/client/test_notification_response.py

@@ -6,17 +6,18 @@
 
 import json
 
-import httpx
+import httpx2
+import mcp_types as types
 import pytest
+from mcp_types import RootsListChangedNotification
 from starlette.applications import Starlette
 from starlette.requests import Request
 from starlette.responses import JSONResponse, Response
 from starlette.routing import Route
 
-from mcp import ClientSession, MCPError, types
+from mcp import ClientSession, MCPError
 from mcp.client.streamable_http import streamable_http_client
 from mcp.shared.session import RequestResponder
-from mcp.types import RootsListChangedNotification
 
 pytestmark = pytest.mark.anyio
 
@@ -71,6 +72,21 @@ async def handle_mcp_request(request: Request) -> Response:
     return Starlette(debug=True, routes=[Route("/mcp", handle_mcp_request, methods=["POST"])])
 
 
+def _create_plain_text_server_app() -> Starlette:
+    """Create a server that returns text/plain for all requests, including initialize.
+
+    This reproduces the scenario from issue #2432 where a misconfigured server
+    returns an unexpected content type on the initialize call, causing the client
+    to hang forever.
+    """
+
+    async def handle_mcp_request(request: Request) -> Response:
+        # Always return text/plain — never a valid MCP response.
+        return Response(content="this is not json", status_code=200, media_type="text/plain")
+
+    return Starlette(debug=True, routes=[Route("/mcp", handle_mcp_request, methods=["POST"])])
+
+
 async def test_non_compliant_notification_response() -> None:
     """Verify the client ignores unexpected responses to notifications.
 
@@ -88,7 +104,7 @@ async def message_handler(  # pragma: no cover
         if isinstance(message, Exception):
             returned_exception = message
 
-    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=_create_non_sdk_server_app())) as client:
+    async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=_create_non_sdk_server_app())) as client:
         async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream):
             async with ClientSession(read_stream, write_stream, message_handler=message_handler) as session:
                 await session.initialize()
@@ -107,7 +123,7 @@ async def test_unexpected_content_type_sends_jsonrpc_error() -> None:
     the client should send a JSONRPCError so the pending request resolves immediately
     instead of hanging until timeout.
     """
-    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=_create_unexpected_content_type_app())) as client:
+    async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=_create_unexpected_content_type_app())) as client:
         async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream):
             async with ClientSession(read_stream, write_stream) as session:  # pragma: no branch
                 await session.initialize()
@@ -116,6 +132,20 @@ async def test_unexpected_content_type_sends_jsonrpc_error() -> None:
                     await session.list_tools()
 
 
+async def test_initialize_does_not_hang_on_unexpected_content_type() -> None:
+    """Verify that initialize() raises MCPError immediately when server returns wrong content type.
+
+    Regression test for issue #2432: when a misconfigured server returns a content type
+    other than application/json or text/event-stream in response to the initialize request,
+    the client must raise MCPError right away instead of hanging forever.
+    """
+    async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=_create_plain_text_server_app())) as client:
+        async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream):
+            async with ClientSession(read_stream, write_stream) as session:  # pragma: no branch
+                with pytest.raises(MCPError, match="Unexpected content type: text/plain"):  # pragma: no branch
+                    await session.initialize()
+
+
 def _create_http_error_app(error_status: int, *, error_on_notifications: bool = False) -> Starlette:
     """Create a server that returns an HTTP error for non-init requests."""
 
@@ -141,9 +171,9 @@ async def test_http_error_status_sends_jsonrpc_error() -> None:
 
     When a server returns a non-2xx status code (e.g. 500), the client should
     send a JSONRPCError so the pending request resolves immediately instead of
-    raising an unhandled httpx.HTTPStatusError that causes the caller to hang.
+    raising an unhandled httpx2.HTTPStatusError that causes the caller to hang.
     """
-    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=_create_http_error_app(500))) as client:
+    async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=_create_http_error_app(500))) as client:
         async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream):
             async with ClientSession(read_stream, write_stream) as session:  # pragma: no branch
                 await session.initialize()
@@ -159,7 +189,7 @@ async def test_http_error_on_notification_does_not_hang() -> None:
     unblock, so the client should just return without sending a JSONRPCError.
     """
     app = _create_http_error_app(500, error_on_notifications=True)
-    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app)) as client:
+    async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app)) as client:
         async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream):
             async with ClientSession(read_stream, write_stream) as session:  # pragma: no branch
                 await session.initialize()
@@ -194,10 +224,76 @@ async def test_invalid_json_response_sends_jsonrpc_error() -> None:
     should send a JSONRPCError so the pending request resolves immediately
     instead of hanging until timeout.
     """
-    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=_create_invalid_json_response_app())) as client:
+    async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=_create_invalid_json_response_app())) as client:
         async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream):
             async with ClientSession(read_stream, write_stream) as session:  # pragma: no branch
                 await session.initialize()
 
                 with pytest.raises(MCPError, match="Failed to parse JSON response"):  # pragma: no branch
                     await session.list_tools()
+
+
+def _create_non_2xx_json_body_app(status: int, body: bytes) -> Starlette:
+    """Server that returns a fixed non-2xx status + ``application/json`` body for non-init requests.
+
+    The initialize response carries an ``mcp-session-id`` so the client treats subsequent
+    requests as part of an established session (needed for the 404 → session-terminated mapping).
+    """
+
+    async def handle_mcp_request(request: Request) -> Response:
+        data = json.loads(await request.body())
+        if data.get("method") == "initialize":
+            return JSONResponse(
+                {"jsonrpc": "2.0", "id": data["id"], "result": INIT_RESPONSE},
+                headers={"mcp-session-id": "test-session"},
+            )
+        if "id" not in data:
+            return Response(status_code=202)
+        return Response(content=body, status_code=status, media_type="application/json")
+
+    return Starlette(debug=True, routes=[Route("/mcp", handle_mcp_request, methods=["POST"])])
+
+
+async def test_client_surfaces_jsonrpc_error_from_non_2xx_body_with_correlated_id() -> None:
+    """SDK-defined: a JSON-RPC error in a non-2xx body is surfaced verbatim even when the
+    server set ``id: null`` — the client rewraps it under the pending request's id, so
+    the awaiting call resolves with the server's error code instead of the generic fallback."""
+    body = json.dumps(
+        {"jsonrpc": "2.0", "id": None, "error": {"code": types.METHOD_NOT_FOUND, "message": "nope"}}
+    ).encode()
+    app = _create_non_2xx_json_body_app(400, body)
+    async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app)) as client:
+        async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream):
+            async with ClientSession(read_stream, write_stream) as session:  # pragma: no branch
+                await session.initialize()
+                with pytest.raises(MCPError) as exc:
+                    await session.list_tools()
+                assert exc.value.error.code == types.METHOD_NOT_FOUND
+
+
+async def test_client_falls_back_to_generic_error_when_non_2xx_body_is_a_jsonrpc_result() -> None:
+    """SDK-defined: a non-2xx response whose JSON body parses as a JSON-RPC *result* (not an
+    error) falls through to the generic ``INTERNAL_ERROR`` fallback rather than being
+    treated as the request's reply."""
+    app = _create_non_2xx_json_body_app(400, b'{"jsonrpc":"2.0","id":1,"result":{}}')
+    async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app)) as client:
+        async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream):
+            async with ClientSession(read_stream, write_stream) as session:  # pragma: no branch
+                await session.initialize()
+                with pytest.raises(MCPError) as exc:
+                    await session.list_tools()
+                assert exc.value.error.code == types.INTERNAL_ERROR
+
+
+async def test_client_falls_back_to_session_terminated_when_404_body_is_malformed_json() -> None:
+    """SDK-defined: an unparseable ``application/json`` body on a 404 response is swallowed
+    and the status-derived ``INVALID_REQUEST`` (session-terminated) fallback resolves the
+    pending request — the parse failure never propagates."""
+    app = _create_non_2xx_json_body_app(404, b"not valid json{{{")
+    async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app)) as client:
+        async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream):
+            async with ClientSession(read_stream, write_stream) as session:  # pragma: no branch
+                await session.initialize()
+                with pytest.raises(MCPError) as exc:
+                    await session.list_tools()
+                assert exc.value.error.code == types.INVALID_REQUEST