Merge branch 'opensearch-project:main' into main

Author: owaiskazi19

.gitignore

@@ -10,3 +10,4 @@ ml-algorithms/build/
 plugin/build/
 .DS_Store
 */bin/
+**/*.factorypath

common/build.gradle

@@ -23,7 +23,7 @@ dependencies {
     testImplementation "org.opensearch.test:framework:${opensearch_version}"
 
     compileOnly group: 'org.apache.commons', name: 'commons-text', version: '1.10.0'
-    compileOnly group: 'com.google.code.gson', name: 'gson', version: '2.13.2'
+    compileOnly group: 'com.google.code.gson', name: 'gson', version: "${versions.gson}"
     compileOnly group: 'org.json', name: 'json', version: '20231013'
     testImplementation group: 'org.json', name: 'json', version: '20231013'
     implementation('com.google.guava:guava:32.1.3-jre') {
@@ -43,6 +43,10 @@ dependencies {
     compileOnly group: 'com.networknt' , name: 'json-schema-validator', version: '1.4.0'
     // Multi-tenant SDK Client
     compileOnly "org.opensearch:opensearch-remote-metadata-sdk:${opensearch_build}"
+    compileOnly (group: 'software.amazon.awssdk', name: 'netty-nio-client', version: "${versions.aws}") {
+        exclude(group: 'org.reactivestreams', module: 'reactive-streams')
+        exclude(group: 'org.slf4j', module: 'slf4j-api')
+    }
 }
 
 lombok {

common/src/main/java/org/opensearch/ml/common/CommonValue.java

@@ -86,6 +86,8 @@ public class CommonValue {
     public static final Version VERSION_2_19_0 = Version.fromString("2.19.0");
     public static final Version VERSION_3_0_0 = Version.fromString("3.0.0");
     public static final Version VERSION_3_1_0 = Version.fromString("3.1.0");
+    public static final Version VERSION_3_2_0 = Version.fromString("3.2.0");
+    public static final Version VERSION_3_3_0 = Version.fromString("3.3.0");
 
     // Connector Constants
     public static final String NAME_FIELD = "name";
@@ -112,7 +114,9 @@ public class CommonValue {
     public static final String MCP_CONNECTORS_FIELD = "mcp_connectors";
     public static final String MCP_CONNECTOR_ID_FIELD = "mcp_connector_id";
     public static final String MCP_DEFAULT_SSE_ENDPOINT = "/sse";
-    public static final String SSE_ENDPOINT_FILED = "sse_endpoint";
+    public static final String SSE_ENDPOINT_FIELD = "sse_endpoint";
+    public static final String MCP_DEFAULT_STREAMABLE_HTTP_ENDPOINT = "/mcp/";
+    public static final String ENDPOINT_FIELD = "endpoint";
 
     // TOOL Constants
     public static final String TOOL_INPUT_SCHEMA_FIELD = "input_schema";
@@ -121,4 +125,16 @@ public class CommonValue {
     public static final long INDEX_INSIGHT_GENERATING_TIMEOUT = 3 * 60 * 1000;
     public static final long INDEX_INSIGHT_UPDATE_INTERVAL = 24 * 60 * 60 * 1000;
 
+    // JSON-RPC Error Codes
+    public static final int JSON_RPC_PARSE_ERROR = -32700;
+    public static final int JSON_RPC_INTERNAL_ERROR = -32603;
+    public static final int JSON_RPC_SERVER_NOT_READY_ERROR = -32000;
+
+    // MCP Server response fields
+    public static final String ACKNOWLEDGE_FIELD = "acknowledged";
+    public static final String MCP_RESPONSE_FIELD = "mcp_response";
+    public static final String ERROR_FIELD = "error";
+    public static final String MESSAGE_FIELD = "message";
+    public static final String ID_FIELD = "id";
+    public static final String ERROR_CODE_FIELD = "error_code";
 }

common/src/main/java/org/opensearch/ml/common/connector/ConnectorProtocols.java

@@ -13,8 +13,9 @@ public class ConnectorProtocols {
     public static final String HTTP = "http";
     public static final String AWS_SIGV4 = "aws_sigv4";
     public static final String MCP_SSE = "mcp_sse";
+    public static final String MCP_STREAMABLE_HTTP = "mcp_streamable_http";
 
-    public static final List<String> VALID_PROTOCOLS = Arrays.asList(AWS_SIGV4, HTTP, MCP_SSE);
+    public static final List<String> VALID_PROTOCOLS = Arrays.asList(AWS_SIGV4, HTTP, MCP_SSE, MCP_STREAMABLE_HTTP);
 
     public static void validateProtocol(String protocol) {
         if (protocol == null) {

common/src/main/java/org/opensearch/ml/common/connector/HttpConnector.java

@@ -19,12 +19,14 @@
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.BiFunction;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import org.apache.commons.text.StringEscapeUtils;
 import org.apache.commons.text.StringSubstitutor;
 import org.opensearch.Version;
 import org.opensearch.common.io.stream.BytesStreamOutput;
@@ -36,6 +38,9 @@
 import org.opensearch.ml.common.AccessMode;
 import org.opensearch.ml.common.transport.connector.MLCreateConnectorInput;
 
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+
 import lombok.Builder;
 import lombok.EqualsAndHashCode;
 import lombok.NoArgsConstructor;
@@ -351,12 +356,41 @@ public <T> T createPayload(String action, Map<String, String> parameters) {
 
             if (!isJson(payload)) {
                 throw new IllegalArgumentException("Invalid payload: " + payload);
+            } else if (neededStreamParameterInPayload(parameters)) {
+                JsonObject jsonObject = JsonParser.parseString(payload).getAsJsonObject();
+                jsonObject.addProperty("stream", true);
+                payload = jsonObject.toString();
             }
             return (T) payload;
         }
         return (T) parameters.get("http_body");
     }
 
+    private boolean neededStreamParameterInPayload(Map<String, String> parameters) {
+        if (parameters == null) {
+            return false;
+        }
+
+        boolean isStream = parameters.containsKey("stream");
+        if (!isStream) {
+            return false;
+        }
+
+        String llmInterface = parameters.get("_llm_interface");
+        if (llmInterface.isBlank()) {
+            return false;
+        }
+
+        llmInterface = llmInterface.trim().toLowerCase(Locale.ROOT);
+        llmInterface = StringEscapeUtils.unescapeJava(llmInterface);
+        switch (llmInterface) {
+            case "openai/v1/chat/completions":
+                return true;
+            default:
+                return false;
+        }
+    }
+
     protected String fillNullParameters(Map<String, String> parameters, String payload) {
         List<String> bodyParams = findStringParametersWithNullDefaultValue(payload);
         String newPayload = payload;

common/src/main/java/org/opensearch/ml/common/connector/McpConnector.java

@@ -358,6 +358,9 @@ public void update(MLCreateConnectorInput updateContent, BiFunction<String, Stri
         if (updateContent.getConnectorClientConfig() != null) {
             this.connectorClientConfig = updateContent.getConnectorClientConfig();
         }
+        if (updateContent.getUrl() != null && updateContent.getUrl().isBlank()) {
+            throw new IllegalArgumentException("MCP Connector url is blank");
+        }
         if (updateContent.getUrl() != null) {
             this.url = updateContent.getUrl();
         }