fix(http12,mcp): set SSE content-type and honor configured MCP message path

dubbo-plugin/dubbo-mcp/src/main/java/org/apache/dubbo/mcp/transport/DubboMcpSseTransportProvider.java

@@ -17,18 +17,22 @@
 package org.apache.dubbo.mcp.transport;
 
 import org.apache.dubbo.cache.support.expiring.ExpiringMap;
+import org.apache.dubbo.common.config.Configuration;
+import org.apache.dubbo.common.config.ConfigurationUtils;
 import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
 import org.apache.dubbo.common.logger.LoggerFactory;
 import org.apache.dubbo.common.stream.StreamObserver;
 import org.apache.dubbo.common.utils.IOUtils;
 import org.apache.dubbo.common.utils.StringUtils;
+import org.apache.dubbo.mcp.McpConstant;
 import org.apache.dubbo.remoting.http12.HttpMethods;
 import org.apache.dubbo.remoting.http12.HttpRequest;
 import org.apache.dubbo.remoting.http12.HttpResponse;
 import org.apache.dubbo.remoting.http12.HttpResult;
 import org.apache.dubbo.remoting.http12.HttpStatus;
 import org.apache.dubbo.remoting.http12.message.ServerSentEvent;
 import org.apache.dubbo.rpc.RpcContext;
+import org.apache.dubbo.rpc.model.ApplicationModel;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
@@ -159,7 +163,9 @@ private void handleSseConnection(StreamObserver<ServerSentEvent<String>> respons
                 new DubboMcpSessionTransport(responseObserver, objectMapper);
         McpServerSession mcpServerSession = sessionFactory.create(dubboMcpSessionTransport);
         sessions.put(mcpServerSession.getId(), mcpServerSession);
-        sendEvent(responseObserver, ENDPOINT_EVENT_TYPE, "/mcp/message" + "?sessionId=" + mcpServerSession.getId());
+        Configuration conf = ConfigurationUtils.getGlobalConfiguration(ApplicationModel.defaultModel());
+        String messagePath = conf.getString(McpConstant.SETTINGS_MCP_PATHS_MESSAGE, "/mcp/message");
+        sendEvent(responseObserver, ENDPOINT_EVENT_TYPE, messagePath + "?sessionId=" + mcpServerSession.getId());
     }
 
     private void refreshSessionExpire(McpServerSession session) {

dubbo-plugin/dubbo-mcp/src/test/java/org/apache/dubbo/mcp/transport/DubboMcpSseTransportProviderTest.java

@@ -36,6 +36,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.MockedStatic;
@@ -99,7 +100,11 @@ void handleRequestHandlesGetRequest() {
         transportProvider.handleRequest(responseObserver);
 
         verify(httpRequest, times(1)).method();
-        verify(responseObserver, times(1)).onNext(any(ServerSentEvent.class));
+        ArgumentCaptor<ServerSentEvent> captor = ArgumentCaptor.forClass(ServerSentEvent.class);
+        verify(responseObserver, times(1)).onNext(captor.capture());
+        ServerSentEvent evt = captor.getValue();
+        Assertions.assertEquals("endpoint", evt.getEvent());
+        Assertions.assertTrue(((String) evt.getData()).startsWith("/mcp/message?sessionId="));
     }
 
     @Test

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/ServerSentEventEncoder.java

@@ -93,12 +93,13 @@ private static void appendField(StringBuilder sb, String name, Object value) {
 
     @Override
     public String contentType() {
-        return httpMessageEncoder.contentType();
+        // An idea:sse use text/event-stream regardless of the underlying data encoder...
+        return MediaType.TEXT_EVENT_STREAM.getName();
     }
 
     @Override
     public MediaType mediaType() {
-        return httpMessageEncoder.mediaType();
+        return MediaType.TEXT_EVENT_STREAM;
     }
 
     @Override

dubbo-remoting/dubbo-remoting-http12/src/test/java/org/apache/dubbo/remoting/http12/message/ServerSentEventEncoderTest.java

@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.remoting.http12.message;
+
+import org.apache.dubbo.remoting.http12.exception.EncodeException;
+
+import java.io.ByteArrayOutputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class ServerSentEventEncoderTest {
+
+    static class DummyJsonEncoder implements HttpMessageEncoder {
+        @Override
+        public void encode(OutputStream outputStream, Object data, java.nio.charset.Charset charset)
+                throws EncodeException {
+            try {
+                if (data instanceof byte[]) {
+                    outputStream.write((byte[]) data);
+                } else {
+                    outputStream.write(String.valueOf(data).getBytes(charset));
+                }
+            } catch (Exception e) {
+                throw new EncodeException("encode error", e);
+            }
+        }
+
+        @Override
+        public MediaType mediaType() {
+            return MediaType.APPLICATION_JSON;
+        }
+
+        @Override
+        public boolean supports(String mediaType) {
+            return true;
+        }
+    }
+
+    @Test
+    void shouldUseTextEventStreamContentType() {
+        ServerSentEventEncoder sse = new ServerSentEventEncoder(new DummyJsonEncoder());
+        Assertions.assertEquals(MediaType.TEXT_EVENT_STREAM, sse.mediaType());
+        Assertions.assertEquals(MediaType.TEXT_EVENT_STREAM.getName(), sse.contentType());
+    }
+
+    @Test
+    void shouldEncodeServerSentEventFormat() {
+        ServerSentEventEncoder sse = new ServerSentEventEncoder(new DummyJsonEncoder());
+        ByteArrayOutputStream bos = new ByteArrayOutputStream();
+        sse.encode(
+                bos,
+                ServerSentEvent.builder()
+                        .event("message")
+                        .data("{\"a\":1}")
+                        .id("1")
+                        .build(),
+                StandardCharsets.UTF_8);
+        String text = bos.toString(StandardCharsets.UTF_8);
+        Assertions.assertTrue(text.contains("id:1\n"));
+        Assertions.assertTrue(text.contains("event:message\n"));
+        Assertions.assertTrue(text.contains("data:{\\\"a\\\":1}\n"));
+        Assertions.assertTrue(text.endsWith("\n"));
+    }
+}