增加SSE客户端支持

Author: relat-ivity

framework/fel/java/plugins/tool-mcp-client/src/main/java/modelengine/fel/tool/mcp/client/support/DefaultMcpClientFactory.java

@@ -6,6 +6,11 @@
 
 package modelengine.fel.tool.mcp.client.support;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import io.modelcontextprotocol.client.transport.HttpClientSseClientTransport;
+import io.modelcontextprotocol.client.transport.HttpClientStreamableHttpTransport;
+import io.modelcontextprotocol.json.jackson.JacksonMcpJsonMapper;
 import modelengine.fel.tool.mcp.client.McpClient;
 import modelengine.fel.tool.mcp.client.McpClientFactory;
 import modelengine.fitframework.annotation.Component;
@@ -32,7 +37,26 @@ public DefaultMcpClientFactory(@Value("${mcp.client.request.timeout-seconds}") i
     }
 
     @Override
+    public McpClient createStreamable(String baseUri, String sseEndpoint) {
+        HttpClientStreamableHttpTransport transport = HttpClientStreamableHttpTransport.builder(baseUri)
+                .jsonMapper(new JacksonMcpJsonMapper(new ObjectMapper()))
+                .endpoint(sseEndpoint)
+                .build();
+        return new DefaultMcpStreamableClient(baseUri, sseEndpoint, this.requestTimeoutSeconds, transport);
+    }
+
+    @Override
+    public McpClient createSse(String baseUri, String sseEndpoint) {
+        HttpClientSseClientTransport transport = HttpClientSseClientTransport.builder(baseUri)
+                .jsonMapper(new JacksonMcpJsonMapper(new ObjectMapper()))
+                .sseEndpoint(sseEndpoint)
+                .build();
+        return new DefaultMcpStreamableClient(baseUri, sseEndpoint, this.requestTimeoutSeconds, transport);
+    }
+
+    @Override
+    @Deprecated
     public McpClient create(String baseUri, String sseEndpoint) {
-        return new DefaultMcpStreamableClient(baseUri, sseEndpoint, requestTimeoutSeconds);
+        return this.createStreamable(baseUri, sseEndpoint);
     }
 }

framework/fel/java/plugins/tool-mcp-client/src/main/java/modelengine/fel/tool/mcp/client/support/DefaultMcpStreamableClient.java

@@ -11,9 +11,8 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 import io.modelcontextprotocol.client.McpSyncClient;
-import io.modelcontextprotocol.client.transport.HttpClientStreamableHttpTransport;
-import io.modelcontextprotocol.json.jackson.JacksonMcpJsonMapper;
 import io.modelcontextprotocol.json.schema.jackson.DefaultJsonSchemaValidator;
+import io.modelcontextprotocol.spec.McpClientTransport;
 import io.modelcontextprotocol.spec.McpSchema;
 import modelengine.fel.tool.mcp.client.McpClient;
 import modelengine.fel.tool.mcp.entity.Tool;
@@ -54,23 +53,18 @@ public class DefaultMcpStreamableClient implements McpClient {
      * @param sseEndpoint The endpoint for the Server-Sent Events (SSE) connection.
      * @param requestTimeoutSeconds The timeout duration of requests. Units: seconds.
      */
-    public DefaultMcpStreamableClient(String baseUri, String sseEndpoint, int requestTimeoutSeconds) {
+    public DefaultMcpStreamableClient(String baseUri, String sseEndpoint, int requestTimeoutSeconds,
+            McpClientTransport transport) {
         this.clientId = UuidUtils.randomUuidString();
         notBlank(baseUri, "The MCP server base URI cannot be blank.");
         notBlank(sseEndpoint, "The MCP server SSE endpoint cannot be blank.");
         log.info("Creating MCP client. [clientId={}, baseUri={}]", this.clientId, baseUri);
-        ObjectMapper mapper = new ObjectMapper();
-        HttpClientStreamableHttpTransport transport = HttpClientStreamableHttpTransport.builder(baseUri)
-                .jsonMapper(new JacksonMcpJsonMapper(mapper))
-                .endpoint(sseEndpoint)
-                .build();
-
         this.logHandler = new DefaultMcpClientLogHandler(this.clientId);
         this.mcpSyncClient = io.modelcontextprotocol.client.McpClient.sync(transport)
                 .requestTimeout(Duration.ofSeconds(requestTimeoutSeconds))
                 .capabilities(McpSchema.ClientCapabilities.builder().build())
                 .loggingConsumer(this.logHandler::handleLoggingMessage)
-                .jsonSchemaValidator(new DefaultJsonSchemaValidator(mapper))
+                .jsonSchemaValidator(new DefaultJsonSchemaValidator(new ObjectMapper()))
                 .build();
     }
 

framework/fel/java/plugins/tool-mcp-test/src/main/java/modelengine/fel/tool/mcp/test/TestController.java

@@ -51,9 +51,17 @@ public TestController(McpClientFactory mcpClientFactory) {
      * @return A string indicating that the initialization was successful.
      */
     @PostMapping(path = "/initialize")
-    public String initialize(@RequestQuery(name = "baseUri") String baseUri,
+    public String initializeStreamable(@RequestQuery(name = "baseUri") String baseUri,
             @RequestQuery(name = "sseEndpoint") String sseEndpoint) {
-        this.client = this.mcpClientFactory.create(baseUri, sseEndpoint);
+        this.client = this.mcpClientFactory.createStreamable(baseUri, sseEndpoint);
+        this.client.initialize();
+        return "Initialized";
+    }
+
+    @PostMapping(path = "/initialize-sse")
+    public String initializeSse(@RequestQuery(name = "baseUri") String baseUri,
+            @RequestQuery(name = "sseEndpoint") String sseEndpoint) {
+        this.client = this.mcpClientFactory.createSse(baseUri, sseEndpoint);
         this.client.initialize();
         return "Initialized";
     }

framework/fel/java/services/tool-mcp-client-service/src/main/java/modelengine/fel/tool/mcp/client/McpClientFactory.java

@@ -16,11 +16,30 @@
  */
 public interface McpClientFactory {
     /**
-     * Creates a {@link McpClient} instance.
+     * Creates a {@link McpClient} instance with streamable HTTP transport.
      *
      * @param baseUri The base URI of the MCP server.
      * @param sseEndpoint The SSE endpoint of the MCP server.
      * @return The connected {@link McpClient} instance.
      */
-    McpClient create(String baseUri, String sseEndpoint);
+    public McpClient createStreamable(String baseUri, String sseEndpoint);
+
+    /**
+     * Creates a {@link McpClient} instance with SSE transport.
+     *
+     * @param baseUri The base URI of the MCP server.
+     * @param sseEndpoint The SSE endpoint of the MCP server.
+     * @return The connected {@link McpClient} instance.
+     */
+    public McpClient createSse(String baseUri, String sseEndpoint);
+
+    /**
+     * Creates a {@link McpClient} instance with streamable HTTP transport.
+     *
+     * @param baseUri The base URI of the MCP server.
+     * @param sseEndpoint The SSE endpoint of the MCP server.
+     * @return The connected {@link McpClient} instance.
+     */
+    @Deprecated
+    public McpClient create(String baseUri, String sseEndpoint);
 }