diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpClientBuilder.java b/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpClientBuilder.java index 11dc47b1f9..9bf463def4 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpClientBuilder.java +++ b/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpClientBuilder.java @@ -103,6 +103,8 @@ public class McpClientBuilder { private Function> asyncElicitationHandler; private Function syncElicitationHandler; private List protocolVersions; + private Boolean resumableStreams; + private Boolean openConnectionOnStartup; private McpClientBuilder(String name) { this.name = name; @@ -189,7 +191,14 @@ public McpClientBuilder customizeSseClient(Consumer 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(ConsumerSet 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. + * + *

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 httpClientCustomizer = null; + private Boolean resumableStreams = null; + private Boolean openConnectionOnStartup = null; public StreamableHttpTransportConfig(String url) { super(url); @@ -764,6 +812,14 @@ public void customizeHttpClient(Consumer 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()) { diff --git a/agentscope-core/src/test/java/io/agentscope/core/tool/mcp/McpClientBuilderTest.java b/agentscope-core/src/test/java/io/agentscope/core/tool/mcp/McpClientBuilderTest.java index 589785bd95..1172e78584 100644 --- a/agentscope-core/src/test/java/io/agentscope/core/tool/mcp/McpClientBuilderTest.java +++ b/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")); + } } diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/tools/McpServerConfig.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/tools/McpServerConfig.java index 4293c13656..2b0e0168ce 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/tools/McpServerConfig.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/tools/McpServerConfig.java @@ -69,6 +69,14 @@ public class McpServerConfig { @JsonProperty("queryParams") private Map 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 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 getEnableTools() { return enableTools; } diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/tools/McpServerRegistrar.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/tools/McpServerRegistrar.java index c3dd22928f..290ba22d6c 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/tools/McpServerRegistrar.java +++ b/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()); } diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/tools/ToolsConfigLoaderTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/tools/ToolsConfigLoaderTest.java index a8d659c0f0..b3b0d93899 100644 --- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/tools/ToolsConfigLoaderTest.java +++ b/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());