fix: return HTTP 400 PARSE_ERROR for non-UTF-8 POST bodies

src/mcp/server/streamable_http.py

@@ -491,8 +491,14 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
             body = await request.body()
 
             try:
-                raw_message = json.loads(body)
-            except json.JSONDecodeError as e:
+                # Decode explicitly rather than passing bytes to json.loads(): the latter
+                # auto-detects UTF-8/16/32 (RFC 4627 sec. 3), so a UTF-16/32 body that is
+                # otherwise valid JSON would parse successfully even though the MCP spec
+                # requires UTF-8.
+                raw_message = json.loads(body.decode("utf-8"))
+            except ValueError as e:
+                # Both json.JSONDecodeError (bad syntax) and UnicodeDecodeError (body bytes
+                # that are not valid UTF-8) subclass ValueError; both are client errors.
                 response = self._create_error_response(f"Parse error: {str(e)}", HTTPStatus.BAD_REQUEST, PARSE_ERROR)
                 await response(scope, receive, send)
                 return

tests/server/test_streamable_http_manager.py

@@ -1,6 +1,7 @@
 """Tests for StreamableHTTPSessionManager."""
 
 import json
+import logging
 from collections.abc import Iterator
 from typing import Any
 from unittest.mock import AsyncMock, patch
@@ -19,7 +20,7 @@
     RequestBodyLimitMiddleware,
     StreamableHTTPSessionManager,
 )
-from mcp.types import INVALID_REQUEST
+from mcp.types import INVALID_REQUEST, PARSE_ERROR
 
 
 @pytest.mark.anyio
@@ -442,6 +443,70 @@ async def mock_receive():
         assert error_data["error"]["message"] == "Session not found"
 
 
+@pytest.mark.parametrize(
+    "encoding",
+    ["cp1252", "utf-16", "utf-32"],
+)
+@pytest.mark.anyio
+async def test_non_utf8_body_returns_parse_error(caplog: pytest.LogCaptureFixture, encoding: str):
+    """A POST body that is not valid UTF-8 is a client error, not a server error.
+
+    For cp1252, json.loads() raises UnicodeDecodeError rather than json.JSONDecodeError,
+    so it used to escape the parse-error branch and surface as HTTP 500 with an
+    ERROR-level traceback.
+
+    For utf-16/utf-32, json.loads() auto-detects the encoding (per RFC 4627 sec. 3) and
+    parses the body successfully even though the MCP spec requires UTF-8, so it used to
+    bypass error handling entirely.
+    """
+    app = Server("test-non-utf8-body")
+    manager = StreamableHTTPSessionManager(app=app, stateless=True)
+
+    async with manager.run():
+        sent_messages: list[Message] = []
+        response_body = b""
+
+        async def mock_send(message: Message):
+            nonlocal response_body
+            sent_messages.append(message)
+            if message["type"] == "http.response.body":
+                response_body += message.get("body", b"")
+
+        scope: Scope = {
+            "type": "http",
+            "method": "POST",
+            "path": "/mcp",
+            "headers": [
+                (b"content-type", b"application/json"),
+                (b"accept", b"application/json, text/event-stream"),
+            ],
+        }
+
+        # Valid JSON syntax, but encoded as something other than UTF-8. For cp1252 the
+        # em dash (byte 0x97) makes the body invalid UTF-8; for utf-16/utf-32 the body
+        # is well-formed under that encoding and would otherwise be auto-detected and
+        # accepted by json.loads().
+        body = '{"jsonrpc": "2.0", "id": 1, "method": "x — y"}'.encode(encoding)
+
+        async def mock_receive():
+            return {"type": "http.request", "body": body, "more_body": False}
+
+        with caplog.at_level(logging.DEBUG):
+            await manager.handle_request(scope, mock_receive, mock_send)
+
+    response_start = next((msg for msg in sent_messages if msg["type"] == "http.response.start"), None)
+    assert response_start is not None, "Should have sent a response"
+    assert response_start["status"] == 400
+
+    error_data = json.loads(response_body)
+    assert error_data["jsonrpc"] == "2.0"
+    assert error_data["error"]["code"] == PARSE_ERROR
+    assert error_data["error"]["message"].startswith("Parse error:")
+
+    error_records = [record for record in caplog.records if record.levelno >= logging.ERROR]
+    assert error_records == [], f"Unparseable body should not log at ERROR: {[r.getMessage() for r in error_records]}"
+
+
 @pytest.mark.anyio
 async def test_idle_session_is_reaped():
     """After idle timeout fires, the session returns 404."""