feat: expose resumableStreams and openConnectionOnStartup on McpClientBuilder

agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpClientBuilder.java

@@ -103,6 +103,8 @@ public class McpClientBuilder {
     private Function<ElicitRequest, Mono<ElicitResult>> asyncElicitationHandler;
     private Function<ElicitRequest, ElicitResult> syncElicitationHandler;
     private List<String> protocolVersions;
+    private Boolean resumableStreams;
+    private Boolean openConnectionOnStartup;
 
     private McpClientBuilder(String name) {
         this.name = name;
@@ -189,7 +191,14 @@ public McpClientBuilder customizeSseClient(Consumer<HttpClient.Builder> customiz
      * @return this builder
      */
     public McpClientBuilder streamableHttpTransport(String url) {
-        this.transportConfig = new StreamableHttpTransportConfig(url);
+        StreamableHttpTransportConfig config = new StreamableHttpTransportConfig(url);
+        if (resumableStreams != null) {
+            config.resumableStreams(resumableStreams);
+        }
+        if (openConnectionOnStartup != null) {
+            config.openConnectionOnStartup(openConnectionOnStartup);
+        }
+        this.transportConfig = config;
         return this;
     }
 
@@ -221,6 +230,43 @@ public McpClientBuilder customizeStreamableHttpClient(Consumer<HttpClient.Builde
         return this;
     }
 
+    /**
+     * Enables or disables resumable (SSE) streams for StreamableHTTP transport.
+     *
+     * <p>Set to {@code false} when connecting to Streamable HTTP MCP servers that
+     * disable SSE and only accept POST requests. Defaults to the SDK's built-in
+     * default ({@code true}) when not called.
+     *
+     * @param value {@code true} to enable SSE streams (default), {@code false} to disable
+     * @return this builder
+     */
+    public McpClientBuilder resumableStreams(boolean value) {
+        this.resumableStreams = value;
+        if (transportConfig instanceof StreamableHttpTransportConfig config) {
+            config.resumableStreams(value);
+        }
+        return this;
+    }
+
+    /**
+     * Controls whether StreamableHTTP transport opens a persistent connection on startup.
+     *
+     * <p>Set to {@code false} together with {@link #resumableStreams(boolean)
+     * resumableStreams(false)} when connecting to SSE-disabled servers.
+     * Defaults to the SDK's built-in default ({@code true}) when not called.
+     *
+     * @param value {@code true} to open connection on startup (default),
+     *              {@code false} to skip
+     * @return this builder
+     */
+    public McpClientBuilder openConnectionOnStartup(boolean value) {
+        this.openConnectionOnStartup = value;
+        if (transportConfig instanceof StreamableHttpTransportConfig config) {
+            config.openConnectionOnStartup(value);
+        }
+        return this;
+    }
+
     /**
      * Adds an HTTP header (only applicable for HTTP transports).
      *
@@ -750,6 +796,8 @@ public McpClientTransport createTransport() {
     private static class StreamableHttpTransportConfig extends HttpTransportConfig {
         private HttpClientStreamableHttpTransport.Builder clientTransportBuilder = null;
         private Consumer<HttpClient.Builder> httpClientCustomizer = null;
+        private Boolean resumableStreams = null;
+        private Boolean openConnectionOnStartup = null;
 
         public StreamableHttpTransportConfig(String url) {
             super(url);
@@ -764,6 +812,14 @@ public void customizeHttpClient(Consumer<HttpClient.Builder> customizer) {
             this.httpClientCustomizer = customizer;
         }
 
+        public void resumableStreams(boolean resumableStreams) {
+            this.resumableStreams = resumableStreams;
+        }
+
+        public void openConnectionOnStartup(boolean openConnectionOnStartup) {
+            this.openConnectionOnStartup = openConnectionOnStartup;
+        }
+
         @Override
         public McpClientTransport createTransport() {
             if (clientTransportBuilder == null) {
@@ -775,6 +831,14 @@ public McpClientTransport createTransport() {
                 clientTransportBuilder.customizeClient(httpClientCustomizer);
             }
 
+            if (resumableStreams != null) {
+                clientTransportBuilder.resumableStreams(resumableStreams);
+            }
+
+            if (openConnectionOnStartup != null) {
+                clientTransportBuilder.openConnectionOnStartup(openConnectionOnStartup);
+            }
+
             clientTransportBuilder.endpoint(extractEndpoint());
 
             if (!headers.isEmpty()) {

agentscope-core/src/test/java/io/agentscope/core/tool/mcp/McpClientBuilderTest.java

@@ -1446,4 +1446,106 @@ void testProtocolVersions_AllNullElementsThrows() {
                                 .stdioTransport("echo", "test")
                                 .protocolVersions(null, null));
     }
+
+    // ==================== Resumable Streams Tests ====================
+
+    private static Object getTransportConfig(McpClientBuilder builder) throws Exception {
+        Field field = McpClientBuilder.class.getDeclaredField("transportConfig");
+        field.setAccessible(true);
+        return field.get(builder);
+    }
+
+    private static Object getTransportConfigFieldValue(McpClientBuilder builder, String fieldName)
+            throws Exception {
+        Object transportConfig = getTransportConfig(builder);
+        Field field = transportConfig.getClass().getDeclaredField(fieldName);
+        field.setAccessible(true);
+        return field.get(transportConfig);
+    }
+
+    @Test
+    void testResumableStreams_AfterStreamableHttpTransport() throws Exception {
+        McpClientBuilder builder =
+                McpClientBuilder.create("resumable-after")
+                        .streamableHttpTransport("https://mcp.example.com/http")
+                        .resumableStreams(false);
+
+        assertEquals(
+                Boolean.FALSE,
+                getTransportConfigFieldValue(builder, "resumableStreams"),
+                "resumableStreams should be stored on transport config");
+    }
+
+    @Test
+    void testOpenConnectionOnStartup_AfterStreamableHttpTransport() throws Exception {
+        McpClientBuilder builder =
+                McpClientBuilder.create("open-after")
+                        .streamableHttpTransport("https://mcp.example.com/http")
+                        .openConnectionOnStartup(false);
+
+        assertEquals(
+                Boolean.FALSE,
+                getTransportConfigFieldValue(builder, "openConnectionOnStartup"),
+                "openConnectionOnStartup should be stored on transport config");
+    }
+
+    @Test
+    void testResumableStreams_BeforeStreamableHttpTransport() throws Exception {
+        McpClientBuilder builder =
+                McpClientBuilder.create("resumable-before")
+                        .resumableStreams(false)
+                        .streamableHttpTransport("https://mcp.example.com/http");
+
+        assertEquals(
+                Boolean.FALSE,
+                getTransportConfigFieldValue(builder, "resumableStreams"),
+                "resumableStreams should be stored even when called before"
+                        + " streamableHttpTransport");
+    }
+
+    @Test
+    void testOpenConnectionOnStartup_BeforeStreamableHttpTransport() throws Exception {
+        McpClientBuilder builder =
+                McpClientBuilder.create("open-before")
+                        .openConnectionOnStartup(false)
+                        .streamableHttpTransport("https://mcp.example.com/http");
+
+        assertEquals(
+                Boolean.FALSE,
+                getTransportConfigFieldValue(builder, "openConnectionOnStartup"),
+                "openConnectionOnStartup should be stored even when called before"
+                        + " streamableHttpTransport");
+    }
+
+    @Test
+    void testResumableStreams_OnSseTransportIsIgnored() throws Exception {
+        McpClientBuilder builder =
+                McpClientBuilder.create("resumable-sse")
+                        .sseTransport("https://mcp.example.com/sse")
+                        .resumableStreams(false);
+
+        Object transportConfig = getTransportConfig(builder);
+        assertEquals("SseTransportConfig", transportConfig.getClass().getSimpleName());
+    }
+
+    @Test
+    void testResumableStreams_BeforeAnyTransportDoesNotThrow() {
+        McpClientBuilder builder =
+                McpClientBuilder.create("resumable-no-transport").resumableStreams(false);
+
+        assertNotNull(builder);
+    }
+
+    @Test
+    void testBothOptions_Combined() throws Exception {
+        McpClientBuilder builder =
+                McpClientBuilder.create("both-options")
+                        .streamableHttpTransport("https://mcp.example.com/http")
+                        .resumableStreams(false)
+                        .openConnectionOnStartup(false);
+
+        assertEquals(Boolean.FALSE, getTransportConfigFieldValue(builder, "resumableStreams"));
+        assertEquals(
+                Boolean.FALSE, getTransportConfigFieldValue(builder, "openConnectionOnStartup"));
+    }
 }

agentscope-harness/src/main/java/io/agentscope/harness/agent/tools/McpServerConfig.java

@@ -69,6 +69,14 @@ public class McpServerConfig {
     @JsonProperty("queryParams")
     private Map<String, String> queryParams;
 
+    /** http: whether StreamableHTTP should use resumable streams. */
+    @JsonProperty("resumableStreams")
+    private Boolean resumableStreams;
+
+    /** http: whether StreamableHTTP should open a connection during startup. */
+    @JsonProperty("openConnectionOnStartup")
+    private Boolean openConnectionOnStartup;
+
     /**
      * Optional allowlist of tools to import from this server. When {@code null} or empty, all
      * tools the server advertises are registered.
@@ -140,6 +148,22 @@ public void setQueryParams(Map<String, String> queryParams) {
         this.queryParams = queryParams;
     }
 
+    public Boolean getResumableStreams() {
+        return resumableStreams;
+    }
+
+    public void setResumableStreams(Boolean resumableStreams) {
+        this.resumableStreams = resumableStreams;
+    }
+
+    public Boolean getOpenConnectionOnStartup() {
+        return openConnectionOnStartup;
+    }
+
+    public void setOpenConnectionOnStartup(Boolean openConnectionOnStartup) {
+        this.openConnectionOnStartup = openConnectionOnStartup;
+    }
+
     public List<String> getEnableTools() {
         return enableTools;
     }

agentscope-harness/src/main/java/io/agentscope/harness/agent/tools/McpServerRegistrar.java

@@ -142,6 +142,12 @@ private static void configureStreamableHttp(
             throw new IllegalArgumentException("http MCP server '" + name + "' requires a 'url'.");
         }
         builder.streamableHttpTransport(cfg.getUrl());
+        if (cfg.getResumableStreams() != null) {
+            builder.resumableStreams(cfg.getResumableStreams());
+        }
+        if (cfg.getOpenConnectionOnStartup() != null) {
+            builder.openConnectionOnStartup(cfg.getOpenConnectionOnStartup());
+        }
         if (cfg.getHeaders() != null && !cfg.getHeaders().isEmpty()) {
             builder.headers(cfg.getHeaders());
         }

agentscope-harness/src/test/java/io/agentscope/harness/agent/tools/ToolsConfigLoaderTest.java

@@ -107,6 +107,8 @@ void load_parsesMcpServers_allTransports() throws Exception {
                     "http": {
                       "transport": "http",
                       "url": "https://mcp.example.com/http",
+                      "resumableStreams": false,
+                      "openConnectionOnStartup": false,
                       "headers": { "X-Tenant": "acme" },
                       "queryParams": { "region": "us" }
                     },
@@ -135,6 +137,8 @@ void load_parsesMcpServers_allTransports() throws Exception {
         assertEquals("https://mcp.example.com/http", http.getUrl());
         assertEquals("acme", http.getHeaders().get("X-Tenant"));
         assertEquals("us", http.getQueryParams().get("region"));
+        assertEquals(Boolean.FALSE, http.getResumableStreams());
+        assertEquals(Boolean.FALSE, http.getOpenConnectionOnStartup());
 
         McpServerConfig sse = cfg.getMcpServers().get("sse");
         assertEquals("sse", sse.getTransport());