Relax Accept header requirement for JSON-only responses

src/mcp/server/streamable_http.py

@@ -306,20 +306,36 @@ def _check_content_type(self, request: Request) -> bool:
 
         return any(part == CONTENT_TYPE_JSON for part in content_type_parts)
 
+    async def _validate_accept_header(self, request: Request, scope: Scope, send: Send) -> bool:
+        """Validate Accept header based on response mode. Returns True if valid."""
+        has_json, has_sse = self._check_accept_headers(request)
+        if self.is_json_response_enabled:
+            # For JSON-only responses, only require application/json
+            if not has_json:
+                response = self._create_error_response(
+                    "Not Acceptable: Client must accept application/json",
+                    HTTPStatus.NOT_ACCEPTABLE,
+                )
+                await response(scope, request.receive, send)
+                return False
+        # For SSE responses, require both content types
+        elif not (has_json and has_sse):
+            response = self._create_error_response(
+                "Not Acceptable: Client must accept both application/json and text/event-stream",
+                HTTPStatus.NOT_ACCEPTABLE,
+            )
+            await response(scope, request.receive, send)
+            return False
+        return True
+
     async def _handle_post_request(self, scope: Scope, request: Request, receive: Receive, send: Send) -> None:
         """Handle POST requests containing JSON-RPC messages."""
         writer = self._read_stream_writer
         if writer is None:
             raise ValueError("No read stream writer available. Ensure connect() is called first.")
         try:
-            # Check Accept headers
-            has_json, has_sse = self._check_accept_headers(request)
-            if not (has_json and has_sse):
-                response = self._create_error_response(
-                    ("Not Acceptable: Client must accept both application/json and text/event-stream"),
-                    HTTPStatus.NOT_ACCEPTABLE,
-                )
-                await response(scope, receive, send)
+            # Validate Accept header
+            if not await self._validate_accept_header(request, scope, send):
                 return
 
             # Validate Content-Type

tests/shared/test_streamable_http.py

@@ -693,6 +693,51 @@ def test_json_response(json_response_server: None, json_server_url: str):
     assert response.headers.get("Content-Type") == "application/json"
 
 
+def test_json_response_accept_json_only(json_response_server: None, json_server_url: str):
+    """Test that json_response servers only require application/json in Accept header."""
+    mcp_url = f"{json_server_url}/mcp"
+    response = requests.post(
+        mcp_url,
+        headers={
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+        },
+        json=INIT_REQUEST,
+    )
+    assert response.status_code == 200
+    assert response.headers.get("Content-Type") == "application/json"
+
+
+def test_json_response_missing_accept_header(json_response_server: None, json_server_url: str):
+    """Test that json_response servers reject requests without Accept header."""
+    mcp_url = f"{json_server_url}/mcp"
+    response = requests.post(
+        mcp_url,
+        headers={
+            "Content-Type": "application/json",
+        },
+        json=INIT_REQUEST,
+    )
+    assert response.status_code == 406
+    assert "Not Acceptable" in response.text
+
+
+def test_json_response_incorrect_accept_header(json_response_server: None, json_server_url: str):
+    """Test that json_response servers reject requests with incorrect Accept header."""
+    mcp_url = f"{json_server_url}/mcp"
+    # Test with only text/event-stream (wrong for JSON server)
+    response = requests.post(
+        mcp_url,
+        headers={
+            "Accept": "text/event-stream",
+            "Content-Type": "application/json",
+        },
+        json=INIT_REQUEST,
+    )
+    assert response.status_code == 406
+    assert "Not Acceptable" in response.text
+
+
 def test_get_sse_stream(basic_server: None, basic_server_url: str):
     """Test establishing an SSE stream via GET request."""
     # First, we need to initialize a session