From 95ae0be2cb21ddf20c697ea3fb6e7051a73a0094 Mon Sep 17 00:00:00 2001 From: wkflash Date: Fri, 3 Jul 2026 23:03:17 +0800 Subject: [PATCH 1/2] feat: expose resumableStreams and openConnectionOnStartup on McpClientBuilder MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds two new methods to McpClientBuilder for StreamableHTTP transport: - resumableStreams(boolean) — disable SSE streams for POST-only servers - openConnectionOnStartup(boolean) — skip persistent connection probe Both default to null (SDK built-in default), fully backward compatible. Fixes: tool calls hanging on Streamable HTTP MCP servers that disable SSE. --- .../core/tool/mcp/McpClientBuilder.java | 56 ++++++++++++++++++- 1 file changed, 55 insertions(+), 1 deletion(-) 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..6b3aaf431b 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,41 @@ 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; + return this; + } + + /** + * Controls whether the transport opens a persistent connection on startup + * for StreamableHTTP transport. + * Only applicable after calling {@link #streamableHttpTransport(String)}. + * + *

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; + return this; + } + /** * Adds an HTTP header (only applicable for HTTP transports). * @@ -750,6 +794,8 @@ public McpClientTransport createTransport() { private static class StreamableHttpTransportConfig extends HttpTransportConfig { private HttpClientStreamableHttpTransport.Builder clientTransportBuilder = null; private Consumer httpClientCustomizer = null; + Boolean resumableStreams; + Boolean openConnectionOnStartup; public StreamableHttpTransportConfig(String url) { super(url); @@ -775,6 +821,14 @@ public McpClientTransport createTransport() { clientTransportBuilder.customizeClient(httpClientCustomizer); } + // Apply SSE/stream options if explicitly set + if (resumableStreams != null) { + clientTransportBuilder.resumableStreams(resumableStreams); + } + if (openConnectionOnStartup != null) { + clientTransportBuilder.openConnectionOnStartup(openConnectionOnStartup); + } + clientTransportBuilder.endpoint(extractEndpoint()); if (!headers.isEmpty()) { From 2937d081288fc97d4471bffa49e5f5aeeb45f86b Mon Sep 17 00:00:00 2001 From: wkflash Date: Sun, 5 Jul 2026 08:21:00 +0800 Subject: [PATCH 2/2] fix(mcp): expose resumableStreams and openConnectionOnStartup for streamable HTTP transport - Add resumableStreams(boolean) and openConnectionOnStartup(boolean) to McpClientBuilder with bidirectional safety: builder field + instanceof check covers both call orders - Wire options through to HttpClientStreamableHttpTransport.Builder - Add resumableStreams and openConnectionOnStartup to tools.json McpServerConfig - Wire JSON config through McpServerRegistrar - Add tests covering both call orders, SSE ignore, no-transport safety, and JSON parsing --- .../core/tool/mcp/McpClientBuilder.java | 36 ++++--- .../core/tool/mcp/McpClientBuilderTest.java | 102 ++++++++++++++++++ .../harness/agent/tools/McpServerConfig.java | 24 +++++ .../agent/tools/McpServerRegistrar.java | 6 ++ .../agent/tools/ToolsConfigLoaderTest.java | 4 + 5 files changed, 159 insertions(+), 13 deletions(-) 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 6b3aaf431b..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 @@ -193,10 +193,10 @@ public McpClientBuilder customizeSseClient(Consumer customiz public McpClientBuilder streamableHttpTransport(String url) { StreamableHttpTransportConfig config = new StreamableHttpTransportConfig(url); if (resumableStreams != null) { - config.resumableStreams = resumableStreams; + config.resumableStreams(resumableStreams); } if (openConnectionOnStartup != null) { - config.openConnectionOnStartup = openConnectionOnStartup; + config.openConnectionOnStartup(openConnectionOnStartup); } this.transportConfig = config; return this; @@ -232,7 +232,6 @@ 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 @@ -243,18 +242,18 @@ public McpClientBuilder customizeStreamableHttpClient(ConsumerSet 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. + *

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 @@ -262,6 +261,9 @@ public McpClientBuilder resumableStreams(boolean value) { */ public McpClientBuilder openConnectionOnStartup(boolean value) { this.openConnectionOnStartup = value; + if (transportConfig instanceof StreamableHttpTransportConfig config) { + config.openConnectionOnStartup(value); + } return this; } @@ -794,8 +796,8 @@ public McpClientTransport createTransport() { private static class StreamableHttpTransportConfig extends HttpTransportConfig { private HttpClientStreamableHttpTransport.Builder clientTransportBuilder = null; private Consumer httpClientCustomizer = null; - Boolean resumableStreams; - Boolean openConnectionOnStartup; + private Boolean resumableStreams = null; + private Boolean openConnectionOnStartup = null; public StreamableHttpTransportConfig(String url) { super(url); @@ -810,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) { @@ -821,10 +831,10 @@ public McpClientTransport createTransport() { clientTransportBuilder.customizeClient(httpClientCustomizer); } - // Apply SSE/stream options if explicitly set if (resumableStreams != null) { clientTransportBuilder.resumableStreams(resumableStreams); } + if (openConnectionOnStartup != null) { clientTransportBuilder.openConnectionOnStartup(openConnectionOnStartup); } 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());