From bc83212d54a54683a614c43e6b5a23ee85a02b40 Mon Sep 17 00:00:00 2001 From: CRYPTONIKAV <29835619+CRYPTONIKAV@users.noreply.github.com> Date: Sun, 11 Oct 2026 02:39:30 -0400 Subject: [PATCH] fix(server): accept mixed-case JSON Content-Type Media types are case-insensitive (RFC 9110 section 8.3.1). Adds interaction test with mixed-case variants plus direct unit tests of _check_content_type. --- src/mcp/server/streamable_http.py | 2 +- .../transports/test_hosting_http.py | 19 +++++++++ .../test_streamable_http_content_type.py | 40 +++++++++++++++++++ 3 files changed, 60 insertions(+), 1 deletion(-) create mode 100644 tests/server/test_streamable_http_content_type.py diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index 416dd9e2b4..ca1b4b9b4e 100644 --- a/src/mcp/server/streamable_http.py +++ b/src/mcp/server/streamable_http.py @@ -529,7 +529,7 @@ async def _handle_request(self, scope: Scope, receive: Receive, send: Send) -> N def _check_content_type(self, request: Request) -> bool: """Check if the request has the correct Content-Type.""" content_type = request.headers.get("content-type", "") - content_type_parts = [part.strip() for part in content_type.split(";")[0].split(",")] + content_type_parts = [part.strip().lower() for part in content_type.split(";")[0].split(",")] return any(part == CONTENT_TYPE_JSON for part in content_type_parts) diff --git a/tests/interaction/transports/test_hosting_http.py b/tests/interaction/transports/test_hosting_http.py index ff9ac2ed05..ccd4290b84 100644 --- a/tests/interaction/transports/test_hosting_http.py +++ b/tests/interaction/transports/test_hosting_http.py @@ -131,6 +131,25 @@ async def test_non_json_content_type_is_rejected() -> None: assert (response.status_code, response.text) == snapshot((400, "Invalid Content-Type header")) +@requirement("hosting:http:content-type-415") +async def test_mixed_case_json_content_type_is_accepted() -> None: + """Media types are case-insensitive (RFC 9110 §8.3.1): a mixed-case JSON + Content-Type must pass the transport check like its siblings do.""" + async with mounted_app(_server()) as (http, _): + for value in ( + "application/json", + "Application/JSON", + "APPLICATION/JSON", + "Application/JSON; Charset=UTF-8", + ): + response = await http.post("/mcp", json=initialize_body(), headers=base_headers() | {"content-type": value}) + assert response.status_code == 200, value + refused = await http.post( + "/mcp", content=b"", headers=base_headers() | {"content-type": "text/plain"} + ) + assert refused.status_code == 400 + + @requirement("hosting:http:parse-error-400") @requirement("hosting:http:batch") async def test_malformed_and_batched_bodies_return_400() -> None: diff --git a/tests/server/test_streamable_http_content_type.py b/tests/server/test_streamable_http_content_type.py new file mode 100644 index 0000000000..8751ba5900 --- /dev/null +++ b/tests/server/test_streamable_http_content_type.py @@ -0,0 +1,40 @@ +"""Direct unit tests for StreamableHTTPServerTransport._check_content_type. + +Covers the transport-level 415 path (pragma: no cover through the public +entry, since the security middleware rejects non-JSON first) for +modelcontextprotocol/python-sdk#3670. Media types are case-insensitive +(RFC 9110 §8.3.1). +""" + +import pytest +from starlette.requests import Request + +from mcp.server.streamable_http import StreamableHTTPServerTransport + + +def make_transport() -> StreamableHTTPServerTransport: + # _check_content_type touches no instance state; bypass __init__. + return StreamableHTTPServerTransport.__new__(StreamableHTTPServerTransport) + + +def make_request(content_type: str) -> Request: + return Request({"type": "http", "headers": [(b"content-type", content_type.encode())]}) + + +@pytest.mark.parametrize( + "value", + [ + "application/json", + "Application/JSON", + "APPLICATION/JSON", + "Application/JSON; Charset=UTF-8", + "application/json; charset=utf-8", + ], +) +def test_check_content_type_accepts_json_case_insensitive(value: str) -> None: + assert make_transport()._check_content_type(make_request(value)) is True + + +@pytest.mark.parametrize("value", ["", "text/plain", "application/json-patch+json", "text/html; charset=utf-8"]) +def test_check_content_type_rejects_non_json(value: str) -> None: + assert make_transport()._check_content_type(make_request(value)) is False