This is an automated email from the ASF dual-hosted git repository. lizhimins pushed a commit to branch rocketmq-studio in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
commit c7dcc22240f3d5768c3a85610641739998e83c29 Author: PiliLily <[email protected]> AuthorDate: Fri Jul 24 13:43:43 2026 +0800 feat: add catalog-driven read-only tools (#514) Introduce versioned YAML tool catalog with JSON Schema validation, SHA-256 digest, deep immutability, and two read-only L1 tool handlers (rmq.cluster.list / rmq.capabilities). --- server/pom.xml | 9 + .../common/exception/GlobalExceptionHandler.java | 6 +- .../com/rocketmq/studio/ops/ai/AiController.java | 27 +- .../java/com/rocketmq/studio/ops/ai/AiService.java | 23 ++ .../java/com/rocketmq/studio/ops/ai/AiToolVO.java | 9 + .../com/rocketmq/studio/ops/ai/McpServerImpl.java | 61 ++-- .../rocketmq/studio/ops/ai/McpServerRegistry.java | 11 + .../ops/ai/tool/CapabilitiesToolHandler.java | 55 ++++ .../studio/ops/ai/tool/CapabilityResolver.java | 64 ++++ .../studio/ops/ai/tool/ClusterListToolHandler.java | 64 ++++ .../rocketmq/studio/ops/ai/tool/ToolCatalog.java | 183 ++++++++++++ .../studio/ops/ai/tool/ToolDefinition.java | 75 +++++ .../studio/ops/ai/tool/ToolGatewayService.java | 224 ++++++++++++++ .../ToolHandler.java} | 9 +- .../resources/tool-catalog/rmq-tools.schema.json | 139 +++++++++ .../src/main/resources/tool-catalog/rmq-tools.yaml | 76 +++++ .../rocketmq/studio/StudioApplicationTest.java} | 24 +- .../exception/GlobalExceptionHandlerTest.java | 67 +++++ .../rocketmq/studio/ops/ai/AiControllerTest.java | 128 ++++++++ .../com/rocketmq/studio/ops/ai/AiServiceTest.java | 22 ++ .../studio/ops/ai/tool/ToolCatalogTest.java | 158 ++++++++++ .../studio/ops/ai/tool/ToolGatewayServiceTest.java | 326 +++++++++++++++++++++ 22 files changed, 1716 insertions(+), 44 deletions(-) diff --git a/server/pom.xml b/server/pom.xml index 55e3e94d..6450dcc0 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -30,6 +30,15 @@ <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-validation</artifactId> </dependency> + <dependency> + <groupId>com.fasterxml.jackson.dataformat</groupId> + <artifactId>jackson-dataformat-yaml</artifactId> + </dependency> + <dependency> + <groupId>com.networknt</groupId> + <artifactId>json-schema-validator</artifactId> + <version>2.0.4</version> + </dependency> <dependency> <groupId>org.springdoc</groupId> <artifactId>springdoc-openapi-starter-webmvc-ui</artifactId> diff --git a/server/src/main/java/com/rocketmq/studio/common/exception/GlobalExceptionHandler.java b/server/src/main/java/com/rocketmq/studio/common/exception/GlobalExceptionHandler.java index 26cffb94..79eefd1b 100644 --- a/server/src/main/java/com/rocketmq/studio/common/exception/GlobalExceptionHandler.java +++ b/server/src/main/java/com/rocketmq/studio/common/exception/GlobalExceptionHandler.java @@ -34,10 +34,10 @@ public class GlobalExceptionHandler { private static final Logger log = LoggerFactory.getLogger(GlobalExceptionHandler.class); @ExceptionHandler(BusinessException.class) - @ResponseStatus(HttpStatus.BAD_REQUEST) - public Result<?> handleBusinessException(BusinessException ex) { + public ResponseEntity<Result<?>> handleBusinessException(BusinessException ex) { log.warn("Business exception: {}", ex.getMessage()); - return Result.error(ex.getCode(), ex.getMessage()); + return ResponseEntity.status(ex.getCode()) + .body(Result.error(ex.getCode(), ex.getMessage())); } @ExceptionHandler(PrometheusException.class) diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/AiController.java b/server/src/main/java/com/rocketmq/studio/ops/ai/AiController.java index 27ce1461..3ddd7c8b 100644 --- a/server/src/main/java/com/rocketmq/studio/ops/ai/AiController.java +++ b/server/src/main/java/com/rocketmq/studio/ops/ai/AiController.java @@ -19,14 +19,19 @@ package com.rocketmq.studio.ops.ai; import com.rocketmq.studio.common.domain.Result; import lombok.RequiredArgsConstructor; import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; +import java.util.Collections; import java.util.List; +import java.util.Map; @RestController @RequestMapping("/api/ai") @@ -46,7 +51,25 @@ public class AiController { } @GetMapping("/tools") - public Result<List<AiToolVO>> listTools() { - return Result.ok(aiService.listTools()); + public ResponseEntity<Result<List<AiToolVO>>> listTools( + @RequestParam(required = false) String cluster) { + List<AiToolVO> tools = cluster == null + ? aiService.listTools() + : aiService.listTools(cluster); + return ResponseEntity.ok() + .header("X-RMQ-Catalog-Version", aiService.catalogVersion()) + .header("X-RMQ-Catalog-Digest", aiService.catalogDigest()) + .header("X-RMQ-Minimum-Client-Version", aiService.minimumClientVersion()) + .body(Result.ok(tools)); + } + + @PostMapping("/tools/{name}/execute") + public Result<Object> executeTool( + @PathVariable String name, + @RequestBody(required = false) Map<String, Object> input) { + Map<String, Object> normalizedInput = input == null + ? Collections.emptyMap() + : input; + return Result.ok(aiService.executeTool(name, normalizedInput)); } } diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/AiService.java b/server/src/main/java/com/rocketmq/studio/ops/ai/AiService.java index b14479a7..3196f417 100644 --- a/server/src/main/java/com/rocketmq/studio/ops/ai/AiService.java +++ b/server/src/main/java/com/rocketmq/studio/ops/ai/AiService.java @@ -22,6 +22,7 @@ import org.springframework.stereotype.Service; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import java.util.List; +import java.util.Map; @Slf4j @Service @@ -60,4 +61,26 @@ public class AiService { log.debug("Listing available AI tools"); return mcpServerRegistry.listTools(); } + + public List<AiToolVO> listTools(String clusterId) { + log.debug("Listing available AI tools for cluster: {}", clusterId); + return mcpServerRegistry.listTools(clusterId); + } + + public Object executeTool(String name, Map<String, Object> input) { + log.info("Executing registered AI tool: {}", name); + return mcpServerRegistry.execute(name, input); + } + + public String catalogVersion() { + return mcpServerRegistry.catalogVersion(); + } + + public String catalogDigest() { + return mcpServerRegistry.catalogDigest(); + } + + public String minimumClientVersion() { + return mcpServerRegistry.minimumClientVersion(); + } } diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/AiToolVO.java b/server/src/main/java/com/rocketmq/studio/ops/ai/AiToolVO.java index fc2dbe5c..3683ff44 100644 --- a/server/src/main/java/com/rocketmq/studio/ops/ai/AiToolVO.java +++ b/server/src/main/java/com/rocketmq/studio/ops/ai/AiToolVO.java @@ -21,6 +21,8 @@ import lombok.Builder; import lombok.Data; import lombok.NoArgsConstructor; +import java.util.List; + @Data @Builder @NoArgsConstructor @@ -29,4 +31,11 @@ public class AiToolVO { private String name; private String description; private Object parameters; + private String riskLevel; + private String permission; + private List<String> requiredCapabilities; + private Object outputSchema; + private String viewHint; + private boolean deprecated; + private String replacement; } diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/McpServerImpl.java b/server/src/main/java/com/rocketmq/studio/ops/ai/McpServerImpl.java index 50aaa45f..d079e5e7 100644 --- a/server/src/main/java/com/rocketmq/studio/ops/ai/McpServerImpl.java +++ b/server/src/main/java/com/rocketmq/studio/ops/ai/McpServerImpl.java @@ -16,49 +16,48 @@ */ package com.rocketmq.studio.ops.ai; -import lombok.extern.slf4j.Slf4j; +import com.rocketmq.studio.ops.ai.tool.ToolCatalog; +import com.rocketmq.studio.ops.ai.tool.ToolGatewayService; +import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Component; -import java.util.Arrays; -import java.util.Collections; -import java.util.HashMap; import java.util.List; import java.util.Map; -@Slf4j @Component +@RequiredArgsConstructor public class McpServerImpl implements McpServerRegistry { + private final ToolGatewayService toolGatewayService; + private final ToolCatalog toolCatalog; + @Override public List<AiToolVO> listTools() { - log.debug("Listing available MCP tools (stub)"); + return toolGatewayService.discover(null); + } - Map<String, Object> queryParams = new HashMap<>(); - queryParams.put("type", "object"); - queryParams.put("properties", Collections.singletonMap("query", - Collections.singletonMap("type", "string"))); + @Override + public List<AiToolVO> listTools(String clusterId) { + return toolGatewayService.discover(clusterId); + } - Map<String, Object> brokerParams = new HashMap<>(); - brokerParams.put("type", "object"); - brokerParams.put("properties", Collections.singletonMap("brokerName", - Collections.singletonMap("type", "string"))); + @Override + public Object execute(String name, Map<String, Object> input) { + return toolGatewayService.execute(name, input); + } - return Arrays.asList( - AiToolVO.builder() - .name("query_metrics") - .description("Query RocketMQ metrics from Prometheus") - .parameters(queryParams) - .build(), - AiToolVO.builder() - .name("list_brokers") - .description("List all RocketMQ brokers in the cluster") - .parameters(Collections.emptyMap()) - .build(), - AiToolVO.builder() - .name("diagnose_broker") - .description("Diagnose issues with a specific broker") - .parameters(brokerParams) - .build() - ); + @Override + public String catalogVersion() { + return toolCatalog.getVersion(); + } + + @Override + public String catalogDigest() { + return toolCatalog.getDigest(); + } + + @Override + public String minimumClientVersion() { + return toolCatalog.getMinimumClientVersion(); } } diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/McpServerRegistry.java b/server/src/main/java/com/rocketmq/studio/ops/ai/McpServerRegistry.java index 24b5fff4..8dfcb81e 100644 --- a/server/src/main/java/com/rocketmq/studio/ops/ai/McpServerRegistry.java +++ b/server/src/main/java/com/rocketmq/studio/ops/ai/McpServerRegistry.java @@ -18,8 +18,19 @@ package com.rocketmq.studio.ops.ai; import java.util.List; +import java.util.Map; public interface McpServerRegistry { List<AiToolVO> listTools(); + + List<AiToolVO> listTools(String clusterId); + + Object execute(String name, Map<String, Object> input); + + String catalogVersion(); + + String catalogDigest(); + + String minimumClientVersion(); } diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/tool/CapabilitiesToolHandler.java b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/CapabilitiesToolHandler.java new file mode 100644 index 00000000..706087cd --- /dev/null +++ b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/CapabilitiesToolHandler.java @@ -0,0 +1,55 @@ +/* + * 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 com.rocketmq.studio.ops.ai.tool; + +import com.rocketmq.studio.cluster.broker.ClusterService; +import com.rocketmq.studio.cluster.broker.ClusterVO; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Component; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +@Component +@RequiredArgsConstructor +public class CapabilitiesToolHandler implements ToolHandler { + + private static final String NAME = "rmq.capabilities"; + + private final ClusterService clusterService; + private final CapabilityResolver capabilityResolver; + + @Override + public String name() { + return NAME; + } + + @Override + public Object execute(Map<String, Object> input) { + String clusterId = (String) input.get("cluster"); + ClusterVO cluster = clusterService.getCluster(clusterId); + List<String> capabilities = capabilityResolver.resolve(cluster); + + Map<String, Object> result = new LinkedHashMap<>(); + result.put("cluster", cluster.getId()); + result.put("type", cluster.getType().name()); + result.put("version", cluster.getVersion()); + result.put("capabilities", capabilities); + return result; + } +} diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/tool/CapabilityResolver.java b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/CapabilityResolver.java new file mode 100644 index 00000000..58254984 --- /dev/null +++ b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/CapabilityResolver.java @@ -0,0 +1,64 @@ +/* + * 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 com.rocketmq.studio.ops.ai.tool; + +import com.rocketmq.studio.cluster.broker.ClusterService; +import com.rocketmq.studio.cluster.broker.ClusterVO; +import com.rocketmq.studio.common.exception.BusinessException; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Component; + +import java.util.List; + +@Component +@RequiredArgsConstructor +public class CapabilityResolver { + + private final ClusterService clusterService; + + public List<String> resolve(String clusterId) { + return resolve(clusterService.getCluster(clusterId)); + } + + List<String> resolve(ClusterVO cluster) { + if (cluster.getType() == null) { + throw new BusinessException( + 400, "Cluster type is unavailable: " + cluster.getId()); + } + return switch (cluster.getType()) { + case V4_DIRECT -> List.of( + "REMOTING", + "ROCKETMQ_4"); + case V5_PROXY_LOCAL -> List.of( + "ACL_V2", + "GRPC", + "LITE_TOPIC", + "LOCAL_PROXY", + "POP", + "REMOTING", + "ROCKETMQ_5"); + case V5_PROXY_CLUSTER -> List.of( + "ACL_V2", + "CLUSTER_PROXY", + "GRPC", + "LITE_TOPIC", + "POP", + "REMOTING", + "ROCKETMQ_5"); + }; + } +} diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ClusterListToolHandler.java b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ClusterListToolHandler.java new file mode 100644 index 00000000..47c313f5 --- /dev/null +++ b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ClusterListToolHandler.java @@ -0,0 +1,64 @@ +/* + * 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 com.rocketmq.studio.ops.ai.tool; + +import com.rocketmq.studio.cluster.broker.ClusterService; +import com.rocketmq.studio.cluster.broker.ClusterVO; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Component; + +import java.util.LinkedHashMap; +import java.util.Map; + +@Component +@RequiredArgsConstructor +public class ClusterListToolHandler implements ToolHandler { + + private static final String NAME = "rmq.cluster.list"; + + private final ClusterService clusterService; + + @Override + public String name() { + return NAME; + } + + @Override + public Object execute(Map<String, Object> input) { + return clusterService.listClusters().stream() + .map(ClusterListToolHandler::safeProjection) + .toList(); + } + + private static Map<String, Object> safeProjection(ClusterVO cluster) { + Map<String, Object> result = new LinkedHashMap<>(); + result.put("id", cluster.getId()); + result.put("name", cluster.getName()); + result.put("type", requiredEnumName(cluster.getType(), "type", cluster.getId())); + result.put("status", requiredEnumName(cluster.getStatus(), "status", cluster.getId())); + result.put("version", cluster.getVersion()); + return result; + } + + private static String requiredEnumName(Enum<?> value, String field, String clusterId) { + if (value == null) { + throw new IllegalStateException( + "Cluster " + field + " is unavailable: " + clusterId); + } + return value.name(); + } +} diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ToolCatalog.java b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ToolCatalog.java new file mode 100644 index 00000000..3e81e983 --- /dev/null +++ b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ToolCatalog.java @@ -0,0 +1,183 @@ +/* + * 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 com.rocketmq.studio.ops.ai.tool; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.dataformat.yaml.YAMLFactory; +import com.networknt.schema.Error; +import com.networknt.schema.InputFormat; +import com.networknt.schema.Schema; +import com.networknt.schema.SchemaRegistry; +import com.networknt.schema.SpecificationVersion; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.core.io.Resource; +import org.springframework.core.io.ResourceLoader; +import org.springframework.stereotype.Component; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HexFormat; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +@Component +public class ToolCatalog { + + static final String CATALOG_RESOURCE = "classpath:tool-catalog/rmq-tools.yaml"; + static final String SCHEMA_RESOURCE = "classpath:tool-catalog/rmq-tools.schema.json"; + + private static final String CLUSTER_LIST_TOOL = "rmq.cluster.list"; + private static final ObjectMapper YAML_MAPPER = new ObjectMapper(new YAMLFactory()); + + private final String version; + private final String minimumClientVersion; + private final String digest; + private final List<ToolDefinition> definitions; + private final Map<String, ToolDefinition> definitionsByName; + + @Autowired + public ToolCatalog(ResourceLoader resourceLoader) { + ToolCatalog loaded = load( + resourceLoader.getResource(CATALOG_RESOURCE), + resourceLoader.getResource(SCHEMA_RESOURCE)); + this.version = loaded.version; + this.minimumClientVersion = loaded.minimumClientVersion; + this.digest = loaded.digest; + this.definitions = loaded.definitions; + this.definitionsByName = loaded.definitionsByName; + } + + private ToolCatalog( + String version, + String minimumClientVersion, + String digest, + List<ToolDefinition> definitions, + Map<String, ToolDefinition> definitionsByName) { + this.version = version; + this.minimumClientVersion = minimumClientVersion; + this.digest = digest; + this.definitions = definitions; + this.definitionsByName = definitionsByName; + } + + static ToolCatalog load(Resource catalogResource, Resource schemaResource) { + try { + byte[] catalogBytes = catalogResource.getContentAsByteArray(); + String catalogYaml = new String(catalogBytes, StandardCharsets.UTF_8); + String schemaJson = schemaResource.getContentAsString(StandardCharsets.UTF_8); + + SchemaRegistry registry = SchemaRegistry.withDefaultDialect( + SpecificationVersion.DRAFT_2020_12); + Schema schema = registry.getSchema(schemaJson, InputFormat.JSON); + List<Error> validationErrors = new ArrayList<>( + schema.validate(catalogYaml, InputFormat.YAML)); + if (!validationErrors.isEmpty()) { + validationErrors.sort(Comparator.comparing( + error -> error.getInstanceLocation().toString())); + throw new IllegalStateException( + "Tool catalog schema validation failed: " + validationErrors); + } + + CatalogDocument document = YAML_MAPPER.readValue(catalogBytes, CatalogDocument.class); + return validatedCatalog(document, sha256(catalogBytes)); + } catch (IOException e) { + throw new IllegalStateException("Unable to load RocketMQ tool catalog", e); + } + } + + private static ToolCatalog validatedCatalog(CatalogDocument document, String digest) { + int catalogMajor = majorVersion(document.version()); + int minimumClientMajor = majorVersion(document.minimumClientVersion()); + if (catalogMajor != minimumClientMajor) { + throw new IllegalStateException( + "Catalog and minimum client major versions must match"); + } + + Map<String, ToolDefinition> byName = new LinkedHashMap<>(); + for (ToolDefinition definition : document.tools()) { + if (byName.putIfAbsent(definition.name(), definition) != null) { + throw new IllegalStateException( + "Tool catalog contains duplicate tool name: " + definition.name()); + } + validateClusterConvention(definition); + } + + List<ToolDefinition> immutableDefinitions = List.copyOf(byName.values()); + return new ToolCatalog( + document.version(), + document.minimumClientVersion(), + digest, + immutableDefinitions, + Map.copyOf(byName)); + } + + private static void validateClusterConvention(ToolDefinition definition) { + if (CLUSTER_LIST_TOOL.equals(definition.name())) { + return; + } + Object required = definition.inputSchema().get("required"); + if (!(required instanceof List<?> requiredFields) || !requiredFields.contains("cluster")) { + throw new IllegalStateException( + "Remote tool must require cluster: " + definition.name()); + } + } + + private static int majorVersion(String version) { + return Integer.parseInt(version.substring(0, version.indexOf('.'))); + } + + private static String sha256(byte[] bytes) { + try { + MessageDigest digest = MessageDigest.getInstance("SHA-256"); + return HexFormat.of().formatHex(digest.digest(bytes)); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException("SHA-256 is unavailable", e); + } + } + + public String getVersion() { + return version; + } + + public String getMinimumClientVersion() { + return minimumClientVersion; + } + + public String getDigest() { + return digest; + } + + public List<ToolDefinition> list() { + return definitions; + } + + public Optional<ToolDefinition> find(String name) { + return Optional.ofNullable(definitionsByName.get(name)); + } + + private record CatalogDocument( + String version, + String minimumClientVersion, + List<ToolDefinition> tools) { + } +} diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ToolDefinition.java b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ToolDefinition.java new file mode 100644 index 00000000..493e02ed --- /dev/null +++ b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ToolDefinition.java @@ -0,0 +1,75 @@ +/* + * 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 com.rocketmq.studio.ops.ai.tool; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +public record ToolDefinition( + String name, + Cli cli, + String description, + String riskLevel, + String permission, + List<String> requiredCapabilities, + Map<String, Object> inputSchema, + Map<String, Object> outputSchema, + String viewHint, + boolean deprecated, + String replacement) { + + public ToolDefinition { + requiredCapabilities = List.copyOf(requiredCapabilities); + inputSchema = immutableMap(inputSchema); + outputSchema = immutableMap(outputSchema); + } + + public String getName() { + return name; + } + + private static Map<String, Object> immutableMap(Map<String, Object> source) { + Map<String, Object> copy = new LinkedHashMap<>(); + source.forEach((key, value) -> copy.put(key, immutableValue(value))); + return Collections.unmodifiableMap(copy); + } + + private static Object immutableValue(Object value) { + if (value instanceof Map<?, ?> nestedMap) { + Map<String, Object> copy = new LinkedHashMap<>(); + nestedMap.forEach((key, nestedValue) -> { + if (!(key instanceof String stringKey)) { + throw new IllegalArgumentException("JSON Schema keys must be strings"); + } + copy.put(stringKey, immutableValue(nestedValue)); + }); + return Collections.unmodifiableMap(copy); + } + if (value instanceof List<?> nestedList) { + List<Object> copy = new ArrayList<>(nestedList.size()); + nestedList.forEach(item -> copy.add(immutableValue(item))); + return Collections.unmodifiableList(copy); + } + return value; + } + + public record Cli(String resource, String verb) { + } +} diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ToolGatewayService.java b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ToolGatewayService.java new file mode 100644 index 00000000..c8de44b9 --- /dev/null +++ b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ToolGatewayService.java @@ -0,0 +1,224 @@ +/* + * 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 com.rocketmq.studio.ops.ai.tool; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.networknt.schema.Error; +import com.networknt.schema.Schema; +import com.networknt.schema.SchemaLocation; +import com.networknt.schema.SchemaRegistry; +import com.networknt.schema.SpecificationVersion; +import com.networknt.schema.dialect.Dialects; +import com.rocketmq.studio.common.exception.BusinessException; +import com.rocketmq.studio.ops.ai.AiToolVO; +import org.springframework.stereotype.Service; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +@Service +public class ToolGatewayService { + + private static final String L1 = "L1"; + + private final ToolCatalog catalog; + private final CapabilityResolver capabilityResolver; + private final ObjectMapper objectMapper; + private final Map<String, ToolHandler> handlers; + private final Map<String, Schema> inputSchemas; + private final Map<String, Schema> outputSchemas; + + public ToolGatewayService( + ToolCatalog catalog, + CapabilityResolver capabilityResolver, + ObjectMapper objectMapper, + List<ToolHandler> handlers) { + this.catalog = catalog; + this.capabilityResolver = capabilityResolver; + this.objectMapper = objectMapper; + this.handlers = registerHandlers(catalog, handlers); + + SchemaRegistry registry = SchemaRegistry.withDefaultDialect( + SpecificationVersion.DRAFT_2020_12); + SchemaRegistry metaSchemaRegistry = SchemaRegistry.withDialect( + Dialects.getDraft202012()); + Schema metaSchema = metaSchemaRegistry.getSchema( + SchemaLocation.of(Dialects.getDraft202012().getId())); + this.inputSchemas = compileSchemas(catalog, registry, metaSchema, true); + this.outputSchemas = compileSchemas(catalog, registry, metaSchema, false); + } + + public List<AiToolVO> discover(String clusterId) { + boolean clusterSelected = clusterId != null && !clusterId.isBlank(); + Set<String> capabilities = clusterSelected + ? Set.copyOf(capabilityResolver.resolve(clusterId)) + : Collections.emptySet(); + + return catalog.list().stream() + .filter(definition -> clusterSelected || !requiresCluster(definition)) + .filter(definition -> capabilities.containsAll( + definition.requiredCapabilities())) + .map(ToolGatewayService::toView) + .toList(); + } + + public Object execute(String name, Map<String, Object> input) { + ToolDefinition definition = catalog.find(name) + .orElseThrow(() -> new BusinessException(404, "Tool not found: " + name)); + ToolHandler handler = handlers.get(name); + Map<String, Object> normalizedInput = input == null + ? Collections.emptyMap() + : input; + + validateInput(definition, normalizedInput); + if (!L1.equals(definition.riskLevel())) { + throw new BusinessException( + 400, "Execution rejected; only L1 tools are enabled: " + name); + } + enforceCapabilities(definition, normalizedInput); + + Object output = handler.execute(normalizedInput); + validateOutput(definition, output); + return output; + } + + private void validateInput(ToolDefinition definition, Map<String, Object> input) { + List<Error> errors = sortedErrors( + inputSchemas.get(definition.name()).validate(objectMapper.valueToTree(input))); + if (!errors.isEmpty()) { + throw new BusinessException( + 400, + "Tool input validation failed for " + definition.name() + ": " + errors); + } + } + + private void enforceCapabilities( + ToolDefinition definition, + Map<String, Object> input) { + if (definition.requiredCapabilities().isEmpty()) { + return; + } + Object cluster = input.get("cluster"); + if (!(cluster instanceof String clusterId) || clusterId.isBlank()) { + throw new BusinessException( + 400, "Tool requires a cluster for capability checks: " + definition.name()); + } + Set<String> capabilities = Set.copyOf(capabilityResolver.resolve(clusterId)); + if (!capabilities.containsAll(definition.requiredCapabilities())) { + throw new BusinessException( + 400, "Cluster does not support tool: " + definition.name()); + } + } + + private void validateOutput(ToolDefinition definition, Object output) { + JsonNode outputNode = objectMapper.valueToTree(output); + List<Error> errors = sortedErrors( + outputSchemas.get(definition.name()).validate(outputNode)); + if (!errors.isEmpty()) { + throw new IllegalStateException( + "Tool output validation failed for " + definition.name() + ": " + errors); + } + } + + private static Map<String, ToolHandler> registerHandlers( + ToolCatalog catalog, + List<ToolHandler> handlers) { + Map<String, ToolHandler> registered = new LinkedHashMap<>(); + for (ToolHandler handler : handlers) { + if (registered.putIfAbsent(handler.name(), handler) != null) { + throw new IllegalStateException( + "Tool gateway contains duplicate handler: " + handler.name()); + } + if (catalog.find(handler.name()).isEmpty()) { + throw new IllegalStateException( + "Tool handler is absent from catalog: " + handler.name()); + } + } + + List<String> missing = catalog.list().stream() + .map(ToolDefinition::name) + .filter(name -> !registered.containsKey(name)) + .toList(); + if (!missing.isEmpty()) { + throw new IllegalStateException( + "Tool catalog contains missing handler: " + missing); + } + return Collections.unmodifiableMap(registered); + } + + private Map<String, Schema> compileSchemas( + ToolCatalog catalog, + SchemaRegistry registry, + Schema metaSchema, + boolean input) { + String schemaKind = input ? "input" : "output"; + Map<String, Schema> compiled = new LinkedHashMap<>(); + for (ToolDefinition definition : catalog.list()) { + JsonNode schemaNode = objectMapper.valueToTree( + input ? definition.inputSchema() : definition.outputSchema()); + List<Error> metaSchemaErrors = sortedErrors(metaSchema.validate(schemaNode)); + if (!metaSchemaErrors.isEmpty()) { + throw new IllegalStateException( + "Tool " + schemaKind + " schema is invalid for " + + definition.name() + ": " + metaSchemaErrors); + } + + try { + Schema schema = registry.getSchema(schemaNode); + schema.initializeValidators(); + compiled.put(definition.name(), schema); + } catch (RuntimeException ex) { + throw new IllegalStateException( + "Tool " + schemaKind + " schema is invalid for " + + definition.name(), ex); + } + } + return Collections.unmodifiableMap(compiled); + } + + private static boolean requiresCluster(ToolDefinition definition) { + Object required = definition.inputSchema().get("required"); + return required instanceof List<?> fields && fields.contains("cluster"); + } + + private static AiToolVO toView(ToolDefinition definition) { + return AiToolVO.builder() + .name(definition.name()) + .description(definition.description()) + .parameters(definition.inputSchema()) + .riskLevel(definition.riskLevel()) + .permission(definition.permission()) + .requiredCapabilities(definition.requiredCapabilities()) + .outputSchema(definition.outputSchema()) + .viewHint(definition.viewHint()) + .deprecated(definition.deprecated()) + .replacement(definition.replacement()) + .build(); + } + + private static List<Error> sortedErrors(List<Error> errors) { + List<Error> sorted = new ArrayList<>(errors); + sorted.sort(Comparator.comparing(error -> error.getInstanceLocation().toString())); + return sorted; + } +} diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/McpServerRegistry.java b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ToolHandler.java similarity index 83% copy from server/src/main/java/com/rocketmq/studio/ops/ai/McpServerRegistry.java copy to server/src/main/java/com/rocketmq/studio/ops/ai/tool/ToolHandler.java index 24b5fff4..3a6f8af2 100644 --- a/server/src/main/java/com/rocketmq/studio/ops/ai/McpServerRegistry.java +++ b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ToolHandler.java @@ -14,12 +14,13 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package com.rocketmq.studio.ops.ai; +package com.rocketmq.studio.ops.ai.tool; +import java.util.Map; -import java.util.List; +public interface ToolHandler { -public interface McpServerRegistry { + String name(); - List<AiToolVO> listTools(); + Object execute(Map<String, Object> input); } diff --git a/server/src/main/resources/tool-catalog/rmq-tools.schema.json b/server/src/main/resources/tool-catalog/rmq-tools.schema.json new file mode 100644 index 00000000..f192ffc0 --- /dev/null +++ b/server/src/main/resources/tool-catalog/rmq-tools.schema.json @@ -0,0 +1,139 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema";, + "$id": "https://rocketmq.apache.org/schemas/rmq-tools.schema.json";, + "title": "RocketMQ Studio tool catalog", + "type": "object", + "additionalProperties": false, + "required": [ + "version", + "minimumClientVersion", + "tools" + ], + "properties": { + "version": { + "$ref": "#/$defs/semanticVersion" + }, + "minimumClientVersion": { + "$ref": "#/$defs/semanticVersion" + }, + "tools": { + "type": "array", + "minItems": 1, + "items": { + "$ref": "#/$defs/tool" + } + } + }, + "$defs": { + "semanticVersion": { + "type": "string", + "pattern": "^[0-9]+\\.[0-9]+\\.[0-9]+(?:-[0-9A-Za-z.-]+)?(?:\\+[0-9A-Za-z.-]+)?$" + }, + "toolName": { + "type": "string", + "pattern": "^rmq\\.[a-z][a-z0-9]*(\\.[a-z][a-z0-9]*)*$" + }, + "cli": { + "type": "object", + "additionalProperties": false, + "required": [ + "resource", + "verb" + ], + "properties": { + "resource": { + "type": "string", + "pattern": "^[a-z][a-z0-9-]*$" + }, + "verb": { + "type": "string", + "pattern": "^[a-z][a-z0-9-]*$" + } + } + }, + "tool": { + "type": "object", + "additionalProperties": false, + "required": [ + "name", + "cli", + "description", + "riskLevel", + "permission", + "requiredCapabilities", + "inputSchema", + "outputSchema", + "viewHint", + "deprecated" + ], + "properties": { + "name": { + "$ref": "#/$defs/toolName" + }, + "cli": { + "$ref": "#/$defs/cli" + }, + "description": { + "type": "string", + "minLength": 1 + }, + "riskLevel": { + "enum": [ + "L1", + "L2", + "L3" + ] + }, + "permission": { + "type": "string", + "pattern": "^[a-z][a-z0-9-]*:[a-z][a-z0-9-]*$" + }, + "requiredCapabilities": { + "type": "array", + "uniqueItems": true, + "items": { + "type": "string", + "pattern": "^[A-Z][A-Z0-9_]*$" + } + }, + "inputSchema": { + "type": "object" + }, + "outputSchema": { + "type": "object" + }, + "viewHint": { + "enum": [ + "object", + "table", + "text", + "timeline", + "topology" + ] + }, + "deprecated": { + "type": "boolean" + }, + "replacement": { + "$ref": "#/$defs/toolName" + } + }, + "allOf": [ + { + "if": { + "required": [ + "replacement" + ] + }, + "then": { + "properties": { + "deprecated": { + "const": true + } + } + } + } + ] + } + } +} diff --git a/server/src/main/resources/tool-catalog/rmq-tools.yaml b/server/src/main/resources/tool-catalog/rmq-tools.yaml new file mode 100644 index 00000000..4327667c --- /dev/null +++ b/server/src/main/resources/tool-catalog/rmq-tools.yaml @@ -0,0 +1,76 @@ +version: 1.0.0 +minimumClientVersion: 1.0.0 +tools: + - name: rmq.cluster.list + cli: + resource: cluster + verb: list + description: List RocketMQ clusters available in Studio. + riskLevel: L1 + permission: cluster:read + requiredCapabilities: [] + inputSchema: + type: object + additionalProperties: false + outputSchema: + type: array + items: + type: object + required: + - id + - name + - type + - status + - version + additionalProperties: false + properties: + id: + type: string + name: + type: string + type: + type: string + status: + type: string + version: + type: string + viewHint: table + deprecated: false + - name: rmq.capabilities + cli: + resource: capabilities + verb: get + description: Describe the capabilities of one RocketMQ cluster. + riskLevel: L1 + permission: cluster:read + requiredCapabilities: [] + inputSchema: + type: object + required: + - cluster + additionalProperties: false + properties: + cluster: + type: string + minLength: 1 + outputSchema: + type: object + required: + - cluster + - type + - version + - capabilities + additionalProperties: false + properties: + cluster: + type: string + type: + type: string + version: + type: string + capabilities: + type: array + items: + type: string + viewHint: object + deprecated: false diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/McpServerRegistry.java b/server/src/test/java/com/rocketmq/studio/StudioApplicationTest.java similarity index 52% copy from server/src/main/java/com/rocketmq/studio/ops/ai/McpServerRegistry.java copy to server/src/test/java/com/rocketmq/studio/StudioApplicationTest.java index 24b5fff4..3f67d482 100644 --- a/server/src/main/java/com/rocketmq/studio/ops/ai/McpServerRegistry.java +++ b/server/src/test/java/com/rocketmq/studio/StudioApplicationTest.java @@ -14,12 +14,28 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package com.rocketmq.studio.ops.ai; +package com.rocketmq.studio; +import com.rocketmq.studio.ops.ai.tool.ToolCatalog; +import com.rocketmq.studio.ops.ai.tool.ToolGatewayService; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; -import java.util.List; +import static org.assertj.core.api.Assertions.assertThat; -public interface McpServerRegistry { +@SpringBootTest +class StudioApplicationTest { - List<AiToolVO> listTools(); + @Autowired + private ToolCatalog toolCatalog; + + @Autowired + private ToolGatewayService toolGatewayService; + + @Test + void applicationContextLoadsWithToolGateway() { + assertThat(toolCatalog.getVersion()).isEqualTo("1.0.0"); + assertThat(toolGatewayService.discover(null)).isNotEmpty(); + } } diff --git a/server/src/test/java/com/rocketmq/studio/common/exception/GlobalExceptionHandlerTest.java b/server/src/test/java/com/rocketmq/studio/common/exception/GlobalExceptionHandlerTest.java new file mode 100644 index 00000000..3b5f525c --- /dev/null +++ b/server/src/test/java/com/rocketmq/studio/common/exception/GlobalExceptionHandlerTest.java @@ -0,0 +1,67 @@ +/* + * 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 com.rocketmq.studio.common.exception; + +import com.rocketmq.studio.common.domain.Result; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RestController; + +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +class GlobalExceptionHandlerTest { + + private MockMvc mockMvc; + + @BeforeEach + void setUp() { + mockMvc = MockMvcBuilders.standaloneSetup(new FailingController()) + .setControllerAdvice(new GlobalExceptionHandler()) + .build(); + } + + @Test + void preservesNotFoundBusinessStatusAndEnvelope() throws Exception { + mockMvc.perform(get("/test/business/404")) + .andExpect(status().isNotFound()) + .andExpect(jsonPath("$.code").value(404)) + .andExpect(jsonPath("$.message").value("failure-404")); + } + + @Test + void preservesBadRequestBusinessStatusAndEnvelope() throws Exception { + mockMvc.perform(get("/test/business/400")) + .andExpect(status().isBadRequest()) + .andExpect(jsonPath("$.code").value(400)) + .andExpect(jsonPath("$.message").value("failure-400")); + } + + @RestController + static class FailingController { + + @GetMapping("/test/business/{code}") + Result<Void> fail(@PathVariable int code) { + throw new BusinessException(code, "failure-" + code); + } + } +} diff --git a/server/src/test/java/com/rocketmq/studio/ops/ai/AiControllerTest.java b/server/src/test/java/com/rocketmq/studio/ops/ai/AiControllerTest.java new file mode 100644 index 00000000..7b9e0277 --- /dev/null +++ b/server/src/test/java/com/rocketmq/studio/ops/ai/AiControllerTest.java @@ -0,0 +1,128 @@ +/* + * 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 com.rocketmq.studio.ops.ai; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.http.MediaType; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; + +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +class AiControllerTest { + + private static final String DIGEST = "a".repeat(64); + + private AiService aiService; + private MockMvc mockMvc; + + @BeforeEach + void setUp() { + aiService = mock(AiService.class); + mockMvc = MockMvcBuilders.standaloneSetup(new AiController(aiService)).build(); + + when(aiService.catalogVersion()).thenReturn("1.0.0"); + when(aiService.catalogDigest()).thenReturn(DIGEST); + when(aiService.minimumClientVersion()).thenReturn("1.0.0"); + } + + @Test + void listToolsKeepsTheExistingBodyAndAddsCatalogHeaders() throws Exception { + AiToolVO tool = AiToolVO.builder() + .name("rmq.cluster.list") + .description("List clusters") + .parameters(Map.of("type", "object")) + .riskLevel("L1") + .permission("cluster:read") + .requiredCapabilities(Collections.emptyList()) + .outputSchema(Map.of("type", "array")) + .viewHint("table") + .deprecated(false) + .build(); + when(aiService.listTools()).thenReturn(List.of(tool)); + + mockMvc.perform(get("/api/ai/tools")) + .andExpect(status().isOk()) + .andExpect(header().string("X-RMQ-Catalog-Version", "1.0.0")) + .andExpect(header().string("X-RMQ-Catalog-Digest", DIGEST)) + .andExpect(header().string("X-RMQ-Minimum-Client-Version", "1.0.0")) + .andExpect(jsonPath("$.code").value(200)) + .andExpect(jsonPath("$.message").value("success")) + .andExpect(jsonPath("$.data[0].name").value("rmq.cluster.list")) + .andExpect(jsonPath("$.data[0].parameters.type").value("object")) + .andExpect(jsonPath("$.data[0].riskLevel").value("L1")) + .andExpect(jsonPath("$.data[0].permission").value("cluster:read")) + .andExpect(jsonPath("$.data[0].viewHint").value("table")); + } + + @Test + void listToolsDelegatesTheSelectedCluster() throws Exception { + when(aiService.listTools("cluster-001")).thenReturn(Collections.emptyList()); + + mockMvc.perform(get("/api/ai/tools").queryParam("cluster", "cluster-001")) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.data").isArray()); + + verify(aiService).listTools("cluster-001"); + } + + @Test + void executeToolPreservesStructuredInputAndDottedName() throws Exception { + Map<String, Object> input = Map.of("cluster", "cluster-001"); + Map<String, Object> output = Map.of( + "cluster", "cluster-001", + "capabilities", List.of("GRPC")); + when(aiService.executeTool("rmq.capabilities", input)).thenReturn(output); + + mockMvc.perform(post("/api/ai/tools/rmq.capabilities/execute") + .contentType(MediaType.APPLICATION_JSON) + .content(""" + {"cluster":"cluster-001"} + """)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.code").value(200)) + .andExpect(jsonPath("$.data.cluster").value("cluster-001")) + .andExpect(jsonPath("$.data.capabilities[0]").value("GRPC")); + + verify(aiService).executeTool("rmq.capabilities", input); + } + + @Test + void executeClusterListNormalizesAnAbsentBody() throws Exception { + when(aiService.executeTool("rmq.cluster.list", Collections.emptyMap())) + .thenReturn(Collections.emptyList()); + + mockMvc.perform(post("/api/ai/tools/rmq.cluster.list/execute") + .contentType(MediaType.APPLICATION_JSON)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.data").isArray()); + + verify(aiService).executeTool("rmq.cluster.list", Collections.emptyMap()); + } +} diff --git a/server/src/test/java/com/rocketmq/studio/ops/ai/AiServiceTest.java b/server/src/test/java/com/rocketmq/studio/ops/ai/AiServiceTest.java index ac46c7d1..a7f46a48 100644 --- a/server/src/test/java/com/rocketmq/studio/ops/ai/AiServiceTest.java +++ b/server/src/test/java/com/rocketmq/studio/ops/ai/AiServiceTest.java @@ -181,4 +181,26 @@ class AiServiceTest { assertThat(result.get(0).getParameters()).isNotNull(); assertThat(result.get(0).getParameters()).isInstanceOf(Map.class); } + + @Test + void listToolsForClusterDelegatesClusterSelection() { + when(mcpServerRegistry.listTools("cluster-001")).thenReturn(Collections.emptyList()); + + List<AiToolVO> result = aiService.listTools("cluster-001"); + + assertThat(result).isEmpty(); + verify(mcpServerRegistry).listTools("cluster-001"); + } + + @Test + void executeToolDelegatesStructuredInput() { + Map<String, Object> input = Map.of("cluster", "cluster-001"); + Map<String, Object> output = Map.of("cluster", "cluster-001"); + when(mcpServerRegistry.execute("rmq.capabilities", input)).thenReturn(output); + + Object result = aiService.executeTool("rmq.capabilities", input); + + assertThat(result).isSameAs(output); + verify(mcpServerRegistry).execute("rmq.capabilities", input); + } } diff --git a/server/src/test/java/com/rocketmq/studio/ops/ai/tool/ToolCatalogTest.java b/server/src/test/java/com/rocketmq/studio/ops/ai/tool/ToolCatalogTest.java new file mode 100644 index 00000000..285965b8 --- /dev/null +++ b/server/src/test/java/com/rocketmq/studio/ops/ai/tool/ToolCatalogTest.java @@ -0,0 +1,158 @@ +/* + * 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 com.rocketmq.studio.ops.ai.tool; + +import org.junit.jupiter.api.Test; +import org.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.ClassPathResource; +import org.springframework.core.io.Resource; + +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class ToolCatalogTest { + + @Test + void loadsAndIndexesTheCanonicalCatalog() { + ToolCatalog catalog = ToolCatalog.load(canonicalCatalog(), canonicalSchema()); + + assertThat(catalog.getVersion()).isEqualTo("1.0.0"); + assertThat(catalog.getMinimumClientVersion()).isEqualTo("1.0.0"); + assertThat(catalog.getDigest()).matches("[0-9a-f]{64}"); + assertThat(catalog.list()).extracting(ToolDefinition::getName) + .containsExactly("rmq.cluster.list", "rmq.capabilities"); + assertThat(catalog.find("rmq.cluster.list")).isPresent(); + assertThat(catalog.find("rmq.unknown")).isEmpty(); + } + + @Test + void rejectsCatalogThatDoesNotMatchItsJsonSchema() { + Resource invalid = utf8Resource(""" + version: 1.0.0 + minimumClientVersion: 1.0.0 + tools: not-a-list + """); + + assertThatThrownBy(() -> ToolCatalog.load(invalid, canonicalSchema())) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("catalog schema validation failed"); + } + + @Test + void rejectsDuplicateToolNames() { + Resource duplicate = catalog( + tool("rmq.cluster.list", false), + tool("rmq.cluster.list", false)); + + assertThatThrownBy(() -> ToolCatalog.load(duplicate, canonicalSchema())) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("duplicate tool name"); + } + + @Test + void requiresClusterForEveryRemoteToolExceptClusterList() { + Resource missingCluster = catalog(tool("rmq.capabilities", false)); + + assertThatThrownBy(() -> ToolCatalog.load(missingCluster, canonicalSchema())) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("must require cluster"); + } + + @Test + void rejectsIncompatibleMinimumClientMajorVersion() { + Resource incompatible = utf8Resource(""" + version: 2.0.0 + minimumClientVersion: 1.0.0 + tools: + %s + """.formatted(tool("rmq.cluster.list", false))); + + assertThatThrownBy(() -> ToolCatalog.load(incompatible, canonicalSchema())) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("major versions must match"); + } + + @Test + @SuppressWarnings("unchecked") + void definitionsAreDeeplyImmutableAfterStartupValidation() { + ToolDefinition capabilities = ToolCatalog.load(canonicalCatalog(), canonicalSchema()) + .find("rmq.capabilities") + .orElseThrow(); + Map<String, Object> properties = + (Map<String, Object>) capabilities.inputSchema().get("properties"); + List<String> required = + (List<String>) capabilities.inputSchema().get("required"); + + assertThatThrownBy(() -> properties.put("endpoint", Map.of("type", "string"))) + .isInstanceOf(UnsupportedOperationException.class); + assertThatThrownBy(() -> required.add("endpoint")) + .isInstanceOf(UnsupportedOperationException.class); + } + + private static Resource canonicalCatalog() { + return new ClassPathResource("tool-catalog/rmq-tools.yaml"); + } + + private static Resource canonicalSchema() { + return new ClassPathResource("tool-catalog/rmq-tools.schema.json"); + } + + private static Resource catalog(String... tools) { + return utf8Resource(""" + version: 1.0.0 + minimumClientVersion: 1.0.0 + tools: + %s + """.formatted(String.join("\n", tools))); + } + + private static String tool(String name, boolean clusterRequired) { + String required = clusterRequired ? """ + required: + - cluster + properties: + cluster: + type: string + minLength: 1 + """ : ""; + return """ + - name: %s + cli: + resource: cluster + verb: list + description: Test tool. + riskLevel: L1 + permission: cluster:read + requiredCapabilities: [] + inputSchema: + type: object + %s additionalProperties: false + outputSchema: + type: object + viewHint: object + deprecated: false + """.formatted(name, required); + } + + private static Resource utf8Resource(String value) { + return new ByteArrayResource(value.getBytes(StandardCharsets.UTF_8)); + } +} diff --git a/server/src/test/java/com/rocketmq/studio/ops/ai/tool/ToolGatewayServiceTest.java b/server/src/test/java/com/rocketmq/studio/ops/ai/tool/ToolGatewayServiceTest.java new file mode 100644 index 00000000..7d7cb37c --- /dev/null +++ b/server/src/test/java/com/rocketmq/studio/ops/ai/tool/ToolGatewayServiceTest.java @@ -0,0 +1,326 @@ +/* + * 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 com.rocketmq.studio.ops.ai.tool; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.rocketmq.studio.cluster.broker.ClusterService; +import com.rocketmq.studio.cluster.broker.ClusterVO; +import com.rocketmq.studio.common.domain.enums.ClusterStatus; +import com.rocketmq.studio.common.domain.enums.ClusterType; +import com.rocketmq.studio.common.exception.BusinessException; +import com.rocketmq.studio.ops.ai.AiToolVO; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.ClassPathResource; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +class ToolGatewayServiceTest { + + private ToolCatalog catalog; + private ClusterService clusterService; + private CapabilityResolver capabilityResolver; + private ClusterListToolHandler clusterListHandler; + private CapabilitiesToolHandler capabilitiesHandler; + private ToolGatewayService gateway; + + @BeforeEach + void setUp() { + catalog = canonicalCatalog(); + clusterService = mock(ClusterService.class); + capabilityResolver = new CapabilityResolver(clusterService); + clusterListHandler = new ClusterListToolHandler(clusterService); + capabilitiesHandler = new CapabilitiesToolHandler(clusterService, capabilityResolver); + gateway = gateway(catalog, clusterListHandler, capabilitiesHandler); + } + + @Test + void discoveryWithoutClusterOnlyExposesClusterList() { + assertThat(gateway.discover(null)) + .extracting(AiToolVO::getName) + .containsExactly("rmq.cluster.list"); + verifyNoInteractions(clusterService); + } + + @Test + void discoveryWithClusterExposesRegisteredSupportedTools() { + when(clusterService.getCluster("cluster-v5")).thenReturn(cluster(ClusterType.V5_PROXY_CLUSTER)); + + assertThat(gateway.discover("cluster-v5")) + .extracting(AiToolVO::getName) + .containsExactly("rmq.cluster.list", "rmq.capabilities"); + } + + @Test + void executesClusterListThroughADataMinimizingProjection() { + ClusterVO cluster = cluster(ClusterType.V5_PROXY_CLUSTER); + cluster.setEndpoint("do-not-expose.example:9876"); + when(clusterService.listClusters()).thenReturn(List.of(cluster)); + + Object output = gateway.execute("rmq.cluster.list", Map.of()); + + assertThat(output).isEqualTo(List.of(Map.of( + "id", "cluster-v5", + "name", "test", + "type", "V5_PROXY_CLUSTER", + "status", "healthy", + "version", "5.2.0"))); + assertThat(output.toString()).doesNotContain("do-not-expose"); + } + + @Test + void executesCapabilitiesWithAStableSortedCapabilityList() { + when(clusterService.getCluster("cluster-v5")).thenReturn(cluster(ClusterType.V5_PROXY_CLUSTER)); + + Object output = gateway.execute( + "rmq.capabilities", Map.of("cluster", "cluster-v5")); + + assertThat(output).isEqualTo(Map.of( + "cluster", "cluster-v5", + "type", "V5_PROXY_CLUSTER", + "version", "5.2.0", + "capabilities", List.of( + "ACL_V2", + "CLUSTER_PROXY", + "GRPC", + "LITE_TOPIC", + "POP", + "REMOTING", + "ROCKETMQ_5"))); + } + + @Test + void resolvesCapabilitiesForEveryExistingClusterType() { + when(clusterService.getCluster("v4")).thenReturn(cluster("v4", ClusterType.V4_DIRECT)); + when(clusterService.getCluster("v5-local")) + .thenReturn(cluster("v5-local", ClusterType.V5_PROXY_LOCAL)); + when(clusterService.getCluster("v5-cluster")) + .thenReturn(cluster("v5-cluster", ClusterType.V5_PROXY_CLUSTER)); + + assertThat(capabilityResolver.resolve("v4")) + .containsExactly("REMOTING", "ROCKETMQ_4"); + assertThat(capabilityResolver.resolve("v5-local")) + .containsExactly( + "ACL_V2", + "GRPC", + "LITE_TOPIC", + "LOCAL_PROXY", + "POP", + "REMOTING", + "ROCKETMQ_5"); + assertThat(capabilityResolver.resolve("v5-cluster")) + .containsExactly( + "ACL_V2", + "CLUSTER_PROXY", + "GRPC", + "LITE_TOPIC", + "POP", + "REMOTING", + "ROCKETMQ_5"); + } + + @Test + void rejectsClusterWithMissingType() { + when(clusterService.getCluster("unknown")).thenReturn(cluster("unknown", null)); + + assertThatThrownBy(() -> capabilityResolver.resolve("unknown")) + .isInstanceOf(BusinessException.class) + .hasMessageContaining("Cluster type is unavailable"); + } + + @Test + void rejectsCapabilitiesExecutionWhenClusterTypeIsMissing() { + when(clusterService.getCluster("unknown")).thenReturn(cluster("unknown", null)); + + assertThatThrownBy(() -> gateway.execute( + "rmq.capabilities", Map.of("cluster", "unknown"))) + .isInstanceOf(BusinessException.class) + .hasMessageContaining("Cluster type is unavailable"); + } + + @Test + void rejectsClusterListEntriesWithMissingTypeUsingAStableError() { + when(clusterService.listClusters()).thenReturn(List.of(cluster("unknown", null))); + + assertThatThrownBy(() -> gateway.execute("rmq.cluster.list", Map.of())) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Cluster type is unavailable"); + } + + @Test + void rejectsCapabilitiesWithoutRequiredClusterBeforeHandlerRuns() { + assertThatThrownBy(() -> gateway.execute("rmq.capabilities", Map.of())) + .isInstanceOf(BusinessException.class) + .hasMessageContaining("input validation failed"); + verifyNoInteractions(clusterService); + } + + @Test + void rejectsUnexpectedInputPropertiesBeforeHandlerRuns() { + assertThatThrownBy(() -> gateway.execute( + "rmq.cluster.list", Map.of("endpoint", "attacker.example"))) + .isInstanceOf(BusinessException.class) + .hasMessageContaining("input validation failed"); + verifyNoInteractions(clusterService); + } + + @Test + void rejectsUnknownTools() { + assertThatThrownBy(() -> gateway.execute("rmq.unknown", Map.of())) + .isInstanceOf(BusinessException.class) + .hasMessageContaining("Tool not found"); + } + + @Test + void refusesNonL1CatalogEntriesEvenWhenAHandlerIsRegistered() throws IOException { + String yaml = canonicalCatalogText().replaceFirst("riskLevel: L1", "riskLevel: L2"); + ToolCatalog l2Catalog = ToolCatalog.load( + new ByteArrayResource(yaml.getBytes(StandardCharsets.UTF_8)), + new ClassPathResource("tool-catalog/rmq-tools.schema.json")); + ToolGatewayService l2Gateway = gateway(l2Catalog, clusterListHandler, capabilitiesHandler); + + assertThatThrownBy(() -> l2Gateway.execute("rmq.cluster.list", Map.of())) + .isInstanceOf(BusinessException.class) + .hasMessageContaining("only L1 tools are enabled"); + verifyNoInteractions(clusterService); + } + + @Test + void failsStartupForDuplicateHandlerNames() { + assertThatThrownBy(() -> gateway( + catalog, clusterListHandler, clusterListHandler, capabilitiesHandler)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("duplicate handler"); + } + + @Test + void failsStartupWhenCatalogAndHandlersDoNotMatch() { + assertThatThrownBy(() -> gateway(catalog, clusterListHandler)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("missing handler"); + } + + @Test + void failsStartupWhenToolSchemaContainsAnUnresolvedReference() throws IOException { + String yaml = canonicalCatalogText().replace( + """ + inputSchema: + type: object + additionalProperties: false + """, + """ + inputSchema: + $ref: '#/missing' + """); + ToolCatalog invalidCatalog = ToolCatalog.load( + new ByteArrayResource(yaml.getBytes(StandardCharsets.UTF_8)), + new ClassPathResource("tool-catalog/rmq-tools.schema.json")); + + assertThatThrownBy(() -> gateway( + invalidCatalog, clusterListHandler, capabilitiesHandler)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("input schema") + .hasMessageContaining("rmq.cluster.list"); + } + + @Test + void failsStartupWhenToolSchemaViolatesTheJsonSchemaMetaSchema() throws IOException { + String yaml = canonicalCatalogText().replace( + """ + inputSchema: + type: object + """, + """ + inputSchema: + type: unsupported + """); + ToolCatalog invalidCatalog = ToolCatalog.load( + new ByteArrayResource(yaml.getBytes(StandardCharsets.UTF_8)), + new ClassPathResource("tool-catalog/rmq-tools.schema.json")); + + assertThatThrownBy(() -> gateway( + invalidCatalog, clusterListHandler, capabilitiesHandler)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("input schema") + .hasMessageContaining("rmq.cluster.list"); + } + + @Test + void validatesHandlerOutputAgainstTheCatalog() { + ToolHandler invalidClusterListHandler = new ToolHandler() { + @Override + public String name() { + return "rmq.cluster.list"; + } + + @Override + public Object execute(Map<String, Object> input) { + return Map.of("bad", "shape"); + } + }; + ToolGatewayService invalidGateway = gateway( + catalog, invalidClusterListHandler, capabilitiesHandler); + + assertThatThrownBy(() -> invalidGateway.execute("rmq.cluster.list", Map.of())) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("output validation failed"); + } + + private ToolGatewayService gateway(ToolCatalog toolCatalog, ToolHandler... handlers) { + return new ToolGatewayService( + toolCatalog, + capabilityResolver, + new ObjectMapper(), + List.of(handlers)); + } + + private static ToolCatalog canonicalCatalog() { + return ToolCatalog.load( + new ClassPathResource("tool-catalog/rmq-tools.yaml"), + new ClassPathResource("tool-catalog/rmq-tools.schema.json")); + } + + private static String canonicalCatalogText() throws IOException { + return new ClassPathResource("tool-catalog/rmq-tools.yaml") + .getContentAsString(StandardCharsets.UTF_8); + } + + private static ClusterVO cluster(ClusterType type) { + return cluster("cluster-v5", type); + } + + private static ClusterVO cluster(String id, ClusterType type) { + ClusterVO cluster = ClusterVO.builder() + .name("test") + .type(type) + .status(ClusterStatus.healthy) + .version("5.2.0") + .build(); + cluster.setId(id); + return cluster; + } +}
