Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/mcp/server/streamable_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
19 changes: 19 additions & 0 deletions tests/interaction/transports/test_hosting_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"<not-json/>", 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:
Expand Down
40 changes: 40 additions & 0 deletions tests/server/test_streamable_http_content_type.py
Original file line number Diff line number Diff line change
@@ -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
Loading