Skip to content
Open
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
4 changes: 3 additions & 1 deletion src/mcp/server/streamable_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -411,7 +411,9 @@ def _check_accept_headers(self, request: Request) -> tuple[bool, bool]:
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(",")]
# Media types are case-insensitive (RFC 9110, section 8.3.1), so normalize
# to lower case before comparing — consistent with _check_accept_headers.
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
24 changes: 24 additions & 0 deletions tests/shared/test_streamable_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -597,6 +597,30 @@ def test_streamable_http_transport_init_validation() -> None:
StreamableHTTPServerTransport(mcp_session_id="test\n")


def test_check_content_type_is_case_insensitive() -> None:
"""Content-Type media types are case-insensitive (RFC 9110, section 8.3.1).
A spec-valid request such as ``Content-Type: Application/JSON`` must be
accepted, consistent with ``_check_accept_headers`` (which already lowercases).
"""
transport = StreamableHTTPServerTransport(mcp_session_id=None)

def request_with(content_type: str) -> Request:
return Request({"type": "http", "headers": [(b"content-type", content_type.encode())]})

for value in (
"application/json",
"Application/JSON",
"APPLICATION/JSON",
"application/json; charset=utf-8",
"Application/Json; charset=utf-8",
):
assert transport._check_content_type(request_with(value)) is True, value

# A genuinely different media type is still rejected.
assert transport._check_content_type(request_with("text/plain")) is False


@pytest.mark.anyio
async def test_session_termination(basic_app: Starlette) -> None:
"""DELETE terminates the session, after which requests for it return 404."""
Expand Down
Loading