Copilot commented on code in PR #15406: URL: https://github.com/apache/dubbo/pull/15406#discussion_r2249640592
########## dubbo-plugin/dubbo-mcp/src/main/java/org/apache/dubbo/mcp/transport/DubboMcpSseTransportProvider.java: ########## @@ -0,0 +1,202 @@ +/* + * 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.mcp.transport; + +import org.apache.dubbo.cache.support.expiring.ExpiringMap; +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.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 java.io.IOException; +import java.nio.charset.StandardCharsets; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.modelcontextprotocol.spec.McpError; +import io.modelcontextprotocol.spec.McpSchema; +import io.modelcontextprotocol.spec.McpServerSession; +import io.modelcontextprotocol.spec.McpServerTransport; +import io.modelcontextprotocol.spec.McpServerTransportProvider; +import io.netty.util.internal.StringUtil; Review Comment: Avoid using internal Netty classes like `io.netty.util.internal.StringUtil`. Use standard Java utilities like `org.apache.dubbo.common.utils.StringUtils` or `java.util.Objects` for null/empty checks instead. ```suggestion ``` ########## dubbo-plugin/dubbo-mcp/src/main/java/org/apache/dubbo/mcp/transport/DubboMcpSseTransportProvider.java: ########## @@ -0,0 +1,202 @@ +/* + * 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.mcp.transport; + +import org.apache.dubbo.cache.support.expiring.ExpiringMap; +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.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 java.io.IOException; +import java.nio.charset.StandardCharsets; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.modelcontextprotocol.spec.McpError; +import io.modelcontextprotocol.spec.McpSchema; +import io.modelcontextprotocol.spec.McpServerSession; +import io.modelcontextprotocol.spec.McpServerTransport; +import io.modelcontextprotocol.spec.McpServerTransportProvider; +import io.netty.util.internal.StringUtil; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_UNEXPECTED_EXCEPTION; + +public class DubboMcpSseTransportProvider implements McpServerTransportProvider { + + private static final ErrorTypeAwareLogger logger = + LoggerFactory.getErrorTypeAwareLogger(DubboMcpSseTransportProvider.class); + + /** + * Event type for JSON-RPC messages sent through the SSE connection. + */ + public static final String MESSAGE_EVENT_TYPE = "message"; + + /** + * Event type for sending the message endpoint URI to clients. + */ + public static final String ENDPOINT_EVENT_TYPE = "endpoint"; + + private McpServerSession.Factory sessionFactory; + + private final ObjectMapper objectMapper; + + private final ExpiringMap<String, McpServerSession> sessions = new ExpiringMap<>(30 * 60, 30); + + public DubboMcpSseTransportProvider(ObjectMapper objectMapper) { + this.objectMapper = objectMapper; + sessions.getExpireThread().startExpiryIfNotStarted(); + } + + @Override + public void setSessionFactory(McpServerSession.Factory sessionFactory) { + this.sessionFactory = sessionFactory; + } + + @Override + public Mono<Void> notifyClients(String method, Object params) { + if (sessions.isEmpty()) { + return Mono.empty(); + } + return Flux.fromIterable(sessions.values()) + .flatMap(session -> session.sendNotification(method, params) + .doOnError(e -> logger.error( + COMMON_UNEXPECTED_EXCEPTION, + "", + "", + String.format( + "Failed to send message to session %s: %s", session.getId(), e.getMessage()), + e)) + .onErrorComplete()) + .then(); + } + + @Override + public Mono<Void> closeGracefully() { + return Flux.fromIterable(sessions.values()) + .flatMap(McpServerSession::closeGracefully) + .then(); + } + + public void handleRequest(StreamObserver<ServerSentEvent<String>> responseObserver) { + // Handle the request and return the response + HttpRequest request = RpcContext.getServiceContext().getRequest(HttpRequest.class); + if (HttpMethods.isGet(request.method())) { + handleSseConnection(responseObserver); + } else if (HttpMethods.isPost(request.method())) { + handleMessage(); + } + return; + } + + public void handleMessage() { + HttpRequest request = RpcContext.getServiceContext().getRequest(HttpRequest.class); + String sessionId = request.parameter("sessionId"); + HttpResponse response = RpcContext.getServiceContext().getResponse(HttpResponse.class); + if (StringUtil.isNullOrEmpty(sessionId)) { Review Comment: Replace `StringUtil.isNullOrEmpty(sessionId)` with `StringUtils.isEmpty(sessionId)` from Dubbo's common utilities to avoid using internal Netty classes. ```suggestion if (StringUtils.isEmpty(sessionId)) { ``` ########## dubbo-plugin/dubbo-mcp/src/main/java/org/apache/dubbo/mcp/transport/DubboMcpStreamableTransportProvider.java: ########## @@ -0,0 +1,135 @@ +/* + * 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.mcp.transport; + +import org.apache.dubbo.cache.support.expiring.ExpiringMap; +import org.apache.dubbo.common.stream.StreamObserver; +import org.apache.dubbo.common.utils.CollectionUtils; +import org.apache.dubbo.common.utils.JsonUtils; +import org.apache.dubbo.common.utils.StringUtils; +import org.apache.dubbo.remoting.http12.HttpMethods; +import org.apache.dubbo.remoting.http12.HttpRequest; +import org.apache.dubbo.remoting.http12.HttpResult; +import org.apache.dubbo.remoting.http12.HttpStatus; +import org.apache.dubbo.remoting.http12.HttpUtils; +import org.apache.dubbo.remoting.http12.message.MediaType; +import org.apache.dubbo.remoting.http12.message.ServerSentEvent; +import org.apache.dubbo.rpc.RpcContext; + +import java.util.List; + +import com.fasterxml.jackson.databind.ObjectMapper; +import io.modelcontextprotocol.spec.McpError; +import io.modelcontextprotocol.spec.McpStreamableServerSession; +import io.modelcontextprotocol.spec.McpStreamableServerSession.Factory; +import io.modelcontextprotocol.spec.McpStreamableServerTransportProvider; +import reactor.core.publisher.Mono; + +/** + * Implementation of {@link McpStreamableServerTransportProvider} for the Dubbo MCP transport. + * This class provides methods to manage streamable server sessions and notify clients. + */ +public class DubboMcpStreamableTransportProvider implements McpStreamableServerTransportProvider { + + private Factory sessionFactory; + + private final ObjectMapper objectMapper; + + public static final String SESSION_ID_HEADER = "mcp-session-id"; + + private final ExpiringMap<String, McpStreamableServerSession> sessions = new ExpiringMap<>(30 * 60, 30); + + public DubboMcpStreamableTransportProvider(ObjectMapper objectMapper) { + this.objectMapper = objectMapper; + } + + @Override + public void setSessionFactory(Factory sessionFactory) { + this.sessionFactory = sessionFactory; + } + + @Override + public Mono<Void> notifyClients(String method, Object params) { + return null; + } + + @Override + public void close() {} + + @Override + public Mono<Void> closeGracefully() { + return null; + } + + public void handleRequest(StreamObserver<ServerSentEvent<String>> responseObserver) { + // Handle the request and return the response + HttpRequest request = RpcContext.getServiceContext().getRequest(HttpRequest.class); + if (HttpMethods.isGet(request.method())) { + handleGet(responseObserver); + + } else if (HttpMethods.isPost(request.method())) { + handlePost(responseObserver); + } + return; + } + + private void handleGet(StreamObserver<ServerSentEvent<String>> responseObserver) { + HttpRequest request = RpcContext.getServiceContext().getRequest(HttpRequest.class); + // check header + List<String> accepts = HttpUtils.parseAccept(request.accept()); + if (CollectionUtils.isEmpty(accepts) + || !accepts.contains(MediaType.TEXT_EVENT_STREAM.getName()) + || !accepts.contains(MediaType.APPLICATION_JSON.getName())) { + // 如果没有包含必须类型，则返回异常 Review Comment: [nitpick] Comments should be in English for consistency with the rest of the codebase. Consider translating Chinese comments to English. ```suggestion // If the required types are not included, return an error ``` ########## dubbo-plugin/dubbo-mcp/src/main/java/org/apache/dubbo/mcp/core/McpApplicationDeployListener.java: ########## @@ -0,0 +1,205 @@ +/* + * 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.mcp.core; + +import org.apache.dubbo.common.config.Configuration; +import org.apache.dubbo.common.config.ConfigurationUtils; +import org.apache.dubbo.common.constants.CommonConstants; +import org.apache.dubbo.common.constants.LoggerCodeConstants; +import org.apache.dubbo.common.deploy.ApplicationDeployListener; +import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository; +import org.apache.dubbo.common.utils.CollectionUtils; +import org.apache.dubbo.common.utils.NetUtils; +import org.apache.dubbo.config.ProtocolConfig; +import org.apache.dubbo.config.ServiceConfig; +import org.apache.dubbo.config.bootstrap.builders.InternalServiceConfigBuilder; +import org.apache.dubbo.mcp.McpConstant; +import org.apache.dubbo.mcp.tool.DubboMcpGenericCaller; +import org.apache.dubbo.mcp.tool.DubboOpenApiToolConverter; +import org.apache.dubbo.mcp.tool.DubboServiceToolRegistry; +import org.apache.dubbo.mcp.transport.DubboMcpSseTransportProvider; +import org.apache.dubbo.mcp.transport.DubboMcpStreamableTransportProvider; +import org.apache.dubbo.rpc.model.ApplicationModel; +import org.apache.dubbo.rpc.model.FrameworkModel; +import org.apache.dubbo.rpc.model.ProviderModel; +import org.apache.dubbo.rpc.protocol.tri.rest.openapi.DefaultOpenAPIService; + +import java.util.Collection; +import java.util.concurrent.ExecutorService; + +import com.fasterxml.jackson.databind.ObjectMapper; +import io.modelcontextprotocol.server.McpAsyncServer; +import io.modelcontextprotocol.server.McpServer; +import io.modelcontextprotocol.spec.McpSchema; + +import static org.apache.dubbo.metadata.util.MetadataServiceVersionUtils.V1; + +public class McpApplicationDeployListener implements ApplicationDeployListener { + + private static final ErrorTypeAwareLogger logger = + LoggerFactory.getErrorTypeAwareLogger(McpApplicationDeployListener.class); + private DubboServiceToolRegistry toolRegistry; + private McpServiceFilter mcpServiceFilter; + private boolean mcpEnable = true; + + private volatile ServiceConfig<McpSseService> serviceConfig; + + private static DubboMcpSseTransportProvider dubboMcpSseTransportProvider; + + private static DubboMcpStreamableTransportProvider dubboMcpStreamableTransportProvider; + + private McpAsyncServer mcpAsyncServer; + + @Override + public void onInitialize(ApplicationModel scopeModel) {} + + @Override + public void onStarting(ApplicationModel applicationModel) {} + + public static DubboMcpSseTransportProvider getDubboMcpSseTransportProvider() { + return dubboMcpSseTransportProvider; + } + + public static DubboMcpStreamableTransportProvider getDubboMcpStreamableTransportProvider() { + return dubboMcpStreamableTransportProvider; + } + + @Override + public void onStarted(ApplicationModel applicationModel) { + Configuration globalConf = ConfigurationUtils.getGlobalConfiguration(applicationModel); + mcpEnable = globalConf.getBoolean(McpConstant.SETTINGS_MCP_ENABLE, true); + if (!mcpEnable) { + logger.info("MCP service is disabled, skipping initialization"); + return; + } + try { + logger.info("Initializing MCP server and dynamic service registration"); + + // Initialize service filter + mcpServiceFilter = new McpServiceFilter(applicationModel); + + dubboMcpSseTransportProvider = new DubboMcpSseTransportProvider(new ObjectMapper()); + McpSchema.ServerCapabilities.ToolCapabilities toolCapabilities = + new McpSchema.ServerCapabilities.ToolCapabilities(true); + McpSchema.ServerCapabilities serverCapabilities = + new McpSchema.ServerCapabilities(null, null, null, null, null, toolCapabilities); + + mcpAsyncServer = McpServer.async(dubboMcpSseTransportProvider) + .capabilities(serverCapabilities) + .build(); + + FrameworkModel frameworkModel = applicationModel.getFrameworkModel(); + DefaultOpenAPIService defaultOpenAPIService = new DefaultOpenAPIService(frameworkModel); + + DubboOpenApiToolConverter toolConverter = new DubboOpenApiToolConverter(defaultOpenAPIService); + + DubboMcpGenericCaller genericCaller = new DubboMcpGenericCaller(applicationModel); + + toolRegistry = new DubboServiceToolRegistry(mcpAsyncServer, toolConverter, genericCaller, mcpServiceFilter); + + applicationModel.getBeanFactory().registerBean(toolRegistry); + + Collection<ProviderModel> providerModels = + applicationModel.getApplicationServiceRepository().allProviderModels(); + + int registeredCount = 0; + for (ProviderModel pm : providerModels) { + int serviceRegisteredCount = toolRegistry.registerService(pm); + registeredCount += serviceRegisteredCount; + } + + exportMcpService(applicationModel); + logger.info( + "MCP server initialized successfully, {} existing tools registered, dynamic registration enabled", + registeredCount); + } catch (Exception e) { + logger.error( + LoggerCodeConstants.COMMON_UNEXPECTED_EXCEPTION, + "", + "", + "MCP service initialization failed: " + e.getMessage(), + e); + } + } + + @Override + public void onStopping(ApplicationModel applicationModel) { + if (toolRegistry != null) { + logger.info("MCP server stopping, clearing tool registry"); + toolRegistry.clearRegistry(); + } + } + + @Override + public void onStopped(ApplicationModel applicationModel) { + if (mcpEnable && mcpAsyncServer != null) { + mcpAsyncServer.close(); + } + } + + @Override + public void onFailure(ApplicationModel applicationModel, Throwable cause) {} + + private void exportMcpService(ApplicationModel applicationModel) { + McpSseServiceImpl mcpSseServiceImpl = + applicationModel.getBeanFactory().getOrRegisterBean(McpSseServiceImpl.class); + + ExecutorService internalServiceExecutor = applicationModel + .getFrameworkModel() + .getBeanFactory() + .getBean(FrameworkExecutorRepository.class) + .getInternalServiceExecutor(); + + this.serviceConfig = InternalServiceConfigBuilder.<McpSseService>newBuilder(applicationModel) + .interfaceClass(McpSseService.class) + .protocol(CommonConstants.TRIPLE, McpConstant.MCP_SERVICE_PROTOCOL) + .port(getRegisterPort(), String.valueOf(McpConstant.MCP_SERVICE_PORT)) + .registryId("internal-mcp-registry") + .executor(internalServiceExecutor) + .ref(mcpSseServiceImpl) + .version(V1) + .build(); + serviceConfig.export(); + logger.info("MCP service exported on: {}", serviceConfig.getExportedUrls()); + } + + /** + * Get the Mcp service register port. + * First, try to get config from user configuration, if not found, get from protocol config. + * Second, try to get config from protocol config, if not found, get a random available port. + */ + private int getRegisterPort() { + Configuration globalConf = ConfigurationUtils.getGlobalConfiguration(ApplicationModel.defaultModel()); + int mcpPort = globalConf.getInt(McpConstant.SETTINGS_MCP_PORT, -1); + if (mcpPort != -1) { + return mcpPort; + } Review Comment: [nitpick] The method `getRegisterPort()` has multiple responsibilities (getting port from config, from protocol, or generating random port). Consider extracting this logic into separate methods for better readability and maintainability. ```suggestion int port = getPortFromConfig(); if (port != -1) { return port; } port = getPortFromProtocol(); if (port != -1) { return port; } return getRandomAvailablePort(); } private int getPortFromConfig() { Configuration globalConf = ConfigurationUtils.getGlobalConfiguration(ApplicationModel.defaultModel()); return globalConf.getInt(McpConstant.SETTINGS_MCP_PORT, -1); } private int getPortFromProtocol() { ``` ########## dubbo-plugin/dubbo-mcp/src/main/java/org/apache/dubbo/mcp/transport/DubboMcpStreamableTransportProvider.java: ########## @@ -0,0 +1,135 @@ +/* + * 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.mcp.transport; + +import org.apache.dubbo.cache.support.expiring.ExpiringMap; +import org.apache.dubbo.common.stream.StreamObserver; +import org.apache.dubbo.common.utils.CollectionUtils; +import org.apache.dubbo.common.utils.JsonUtils; +import org.apache.dubbo.common.utils.StringUtils; +import org.apache.dubbo.remoting.http12.HttpMethods; +import org.apache.dubbo.remoting.http12.HttpRequest; +import org.apache.dubbo.remoting.http12.HttpResult; +import org.apache.dubbo.remoting.http12.HttpStatus; +import org.apache.dubbo.remoting.http12.HttpUtils; +import org.apache.dubbo.remoting.http12.message.MediaType; +import org.apache.dubbo.remoting.http12.message.ServerSentEvent; +import org.apache.dubbo.rpc.RpcContext; + +import java.util.List; + +import com.fasterxml.jackson.databind.ObjectMapper; +import io.modelcontextprotocol.spec.McpError; +import io.modelcontextprotocol.spec.McpStreamableServerSession; +import io.modelcontextprotocol.spec.McpStreamableServerSession.Factory; +import io.modelcontextprotocol.spec.McpStreamableServerTransportProvider; +import reactor.core.publisher.Mono; + +/** + * Implementation of {@link McpStreamableServerTransportProvider} for the Dubbo MCP transport. + * This class provides methods to manage streamable server sessions and notify clients. + */ +public class DubboMcpStreamableTransportProvider implements McpStreamableServerTransportProvider { + + private Factory sessionFactory; + + private final ObjectMapper objectMapper; + + public static final String SESSION_ID_HEADER = "mcp-session-id"; + + private final ExpiringMap<String, McpStreamableServerSession> sessions = new ExpiringMap<>(30 * 60, 30); + + public DubboMcpStreamableTransportProvider(ObjectMapper objectMapper) { + this.objectMapper = objectMapper; + } + + @Override + public void setSessionFactory(Factory sessionFactory) { + this.sessionFactory = sessionFactory; + } + + @Override + public Mono<Void> notifyClients(String method, Object params) { + return null; + } + + @Override + public void close() {} + + @Override + public Mono<Void> closeGracefully() { + return null; + } + + public void handleRequest(StreamObserver<ServerSentEvent<String>> responseObserver) { + // Handle the request and return the response + HttpRequest request = RpcContext.getServiceContext().getRequest(HttpRequest.class); + if (HttpMethods.isGet(request.method())) { + handleGet(responseObserver); + + } else if (HttpMethods.isPost(request.method())) { + handlePost(responseObserver); + } + return; + } + + private void handleGet(StreamObserver<ServerSentEvent<String>> responseObserver) { + HttpRequest request = RpcContext.getServiceContext().getRequest(HttpRequest.class); + // check header + List<String> accepts = HttpUtils.parseAccept(request.accept()); + if (CollectionUtils.isEmpty(accepts) + || !accepts.contains(MediaType.TEXT_EVENT_STREAM.getName()) + || !accepts.contains(MediaType.APPLICATION_JSON.getName())) { + // 如果没有包含必须类型，则返回异常 + responseObserver.onError(HttpResult.builder() + .header("Content-Type", MediaType.APPLICATION_JSON.getName()) + .status(HttpStatus.NOT_ACCEPTABLE.getCode()) + .body(JsonUtils.toJson(new McpError("Unsupported accept type").getJsonRpcError())) + .build() + .toPayload()); + responseObserver.onCompleted(); + return; + } + + String sessionId = + RpcContext.getServiceContext().getRequest(HttpRequest.class).header(SESSION_ID_HEADER); + if (StringUtils.isBlank(sessionId)) { + // 如果没有sessionId，则返回异常 Review Comment: [nitpick] Comments should be in English for consistency with the rest of the codebase. Consider translating Chinese comments to English. ```suggestion // If there is no sessionId, return an error ``` ########## dubbo-plugin/dubbo-mcp/src/main/java/org/apache/dubbo/mcp/core/McpServiceFilter.java: ########## @@ -0,0 +1,383 @@ +/* + * 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.mcp.core; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.config.Configuration; +import org.apache.dubbo.common.config.ConfigurationUtils; +import org.apache.dubbo.common.constants.LoggerCodeConstants; +import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.utils.StringUtils; +import org.apache.dubbo.config.annotation.DubboService; +import org.apache.dubbo.mcp.McpConstant; +import org.apache.dubbo.mcp.annotations.McpTool; +import org.apache.dubbo.rpc.model.ApplicationModel; +import org.apache.dubbo.rpc.model.ProviderModel; + +import java.lang.reflect.Method; +import java.lang.reflect.Modifier; +import java.util.Arrays; +import java.util.List; +import java.util.regex.Pattern; + +public class McpServiceFilter { + + private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(McpServiceFilter.class); + + private final Configuration configuration; + private final Pattern[] includePatterns; + private final Pattern[] excludePatterns; + private final boolean defaultEnabled; + + public McpServiceFilter(ApplicationModel applicationModel) { + this.configuration = ConfigurationUtils.getGlobalConfiguration(applicationModel); + this.defaultEnabled = configuration.getBoolean(McpConstant.SETTINGS_MCP_DEFAULT_ENABLED, true); + + String includeStr = configuration.getString(McpConstant.SETTINGS_MCP_INCLUDE_PATTERNS, ""); + String excludeStr = configuration.getString(McpConstant.SETTINGS_MCP_EXCLUDE_PATTERNS, ""); + + this.includePatterns = parsePatterns(includeStr); + this.excludePatterns = parsePatterns(excludeStr); + } + + /** + * Check if service should be exposed as MCP tool. + * Priority: URL Parameters > Annotations > Configuration File > Default + */ + public boolean shouldExposeAsMcpTool(ProviderModel providerModel) { + String interfaceName = providerModel.getServiceModel().getInterfaceName(); + + if (isMatchedByPatterns(interfaceName, excludePatterns)) { + return false; + } + + URL serviceUrl = getServiceUrl(providerModel); + if (serviceUrl != null) { + String urlValue = serviceUrl.getParameter(McpConstant.PARAM_MCP_ENABLED); + if (urlValue != null && StringUtils.isNotEmpty(urlValue)) { + return Boolean.parseBoolean(urlValue); + } + } + + Object serviceBean = providerModel.getServiceInstance(); + if (serviceBean != null) { + DubboService dubboService = serviceBean.getClass().getAnnotation(DubboService.class); + if (dubboService != null && dubboService.mcpEnabled()) { + return true; + } Review Comment: The method `mcpEnabled()` is being called on `DubboService` annotation, but this method doesn't exist in the standard Dubbo annotation. This will cause a compilation error. ```suggestion // The standard DubboService annotation does not have mcpEnabled(). Remove this check. // If annotation-based configuration is needed, implement a custom annotation. ``` ########## dubbo-demo/dubbo-demo-mcp-server/src/main/java/org/apache/dubbo/mcp/server/demo/demo/HelloServiceImpl.java: ########## @@ -0,0 +1,46 @@ +/* + * 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.mcp.server.demo.demo; + +import org.apache.dubbo.config.annotation.DubboService; + +@DubboService(mcpEnabled = true) Review Comment: The `mcpEnabled` attribute doesn't exist in the `@DubboService` annotation. This will cause a compilation error. Consider removing this attribute or implementing it as a custom extension. ```suggestion @DubboService ``` ########## dubbo-plugin/dubbo-mcp/src/main/java/org/apache/dubbo/mcp/transport/DubboMcpStreamableTransportProvider.java: ########## @@ -0,0 +1,135 @@ +/* + * 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.mcp.transport; + +import org.apache.dubbo.cache.support.expiring.ExpiringMap; +import org.apache.dubbo.common.stream.StreamObserver; +import org.apache.dubbo.common.utils.CollectionUtils; +import org.apache.dubbo.common.utils.JsonUtils; +import org.apache.dubbo.common.utils.StringUtils; +import org.apache.dubbo.remoting.http12.HttpMethods; +import org.apache.dubbo.remoting.http12.HttpRequest; +import org.apache.dubbo.remoting.http12.HttpResult; +import org.apache.dubbo.remoting.http12.HttpStatus; +import org.apache.dubbo.remoting.http12.HttpUtils; +import org.apache.dubbo.remoting.http12.message.MediaType; +import org.apache.dubbo.remoting.http12.message.ServerSentEvent; +import org.apache.dubbo.rpc.RpcContext; + +import java.util.List; + +import com.fasterxml.jackson.databind.ObjectMapper; +import io.modelcontextprotocol.spec.McpError; +import io.modelcontextprotocol.spec.McpStreamableServerSession; +import io.modelcontextprotocol.spec.McpStreamableServerSession.Factory; +import io.modelcontextprotocol.spec.McpStreamableServerTransportProvider; +import reactor.core.publisher.Mono; + +/** + * Implementation of {@link McpStreamableServerTransportProvider} for the Dubbo MCP transport. + * This class provides methods to manage streamable server sessions and notify clients. + */ +public class DubboMcpStreamableTransportProvider implements McpStreamableServerTransportProvider { + + private Factory sessionFactory; + + private final ObjectMapper objectMapper; + + public static final String SESSION_ID_HEADER = "mcp-session-id"; + + private final ExpiringMap<String, McpStreamableServerSession> sessions = new ExpiringMap<>(30 * 60, 30); + + public DubboMcpStreamableTransportProvider(ObjectMapper objectMapper) { + this.objectMapper = objectMapper; + } + + @Override + public void setSessionFactory(Factory sessionFactory) { + this.sessionFactory = sessionFactory; + } + + @Override + public Mono<Void> notifyClients(String method, Object params) { + return null; + } + + @Override + public void close() {} + + @Override + public Mono<Void> closeGracefully() { + return null; + } + + public void handleRequest(StreamObserver<ServerSentEvent<String>> responseObserver) { + // Handle the request and return the response + HttpRequest request = RpcContext.getServiceContext().getRequest(HttpRequest.class); + if (HttpMethods.isGet(request.method())) { + handleGet(responseObserver); + + } else if (HttpMethods.isPost(request.method())) { + handlePost(responseObserver); + } + return; + } + + private void handleGet(StreamObserver<ServerSentEvent<String>> responseObserver) { + HttpRequest request = RpcContext.getServiceContext().getRequest(HttpRequest.class); + // check header + List<String> accepts = HttpUtils.parseAccept(request.accept()); + if (CollectionUtils.isEmpty(accepts) + || !accepts.contains(MediaType.TEXT_EVENT_STREAM.getName()) + || !accepts.contains(MediaType.APPLICATION_JSON.getName())) { Review Comment: The condition checks if accepts contains both TEXT_EVENT_STREAM AND APPLICATION_JSON, but uses OR logic. This will reject requests that have one but not the other. Consider using AND logic or separate the checks based on the actual requirements. ```suggestion || (!accepts.contains(MediaType.TEXT_EVENT_STREAM.getName()) && !accepts.contains(MediaType.APPLICATION_JSON.getName()))) { ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
