feat: implement HttpClient Streamable HTTP transport #337
+1,601
−394
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…to reactive streams This change imporves the transport layer with reactive patterns and adds support for the latest MCP specification while maintaining backward compatibility with existing SSE transport. - Add HttpClientStreamableHttpTransport implementing 2025-03-26 MCP Streamable HTTP spec - Add ResponseSubscribers utility for handling SSE and JSON HTTP responses - Refactor HttpClientSseClientTransport to use reactive streams instead of CompletableFuture - Replace FlowSseClient with direct reactive stream handling - Use Disposable-based connection management instead of CountDownLatch - Replace message endpoint discovery with Sinks.One approach - Add resiliency tests using Toxiproxy for network failure scenarios - Minor type safety improvements in StdioClientTransport and DefaultMcpTransportStream Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
- Add proper exception handling with CompletableFuture.exceptionallyCompose for async HTTP operations - Add test for specific exception type handling in resiliency tests This change makes the HTTP client transports more robust by ensuring exceptions are properly propagated. Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
9 tasks
Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
- Replace exceptionallyCompose with whenComplete for better async error handling - Make ResponseSubscribers class package-private Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
9 tasks
tzolov
changed the title
|
@tzolov is there a companion PR that handles Streamable HTTP on the server side? |
Signed-off-by: Dariusz Jędrzejczyk <dariusz.jedrzejczyk@broadcom.com>
…zation Signed-off-by: Dariusz Jędrzejczyk <dariusz.jedrzejczyk@broadcom.com>
chemicL
reviewed
Jul 3, 2025
Comment on lines
+49
to
+95
| static ToxiproxyContainer toxiproxy = new ToxiproxyContainer("ghcr.io/shopify/toxiproxy:2.5.0").withNetwork(network) | ||
| .withExposedPorts(8474, 3000); | ||
|
|
||
| static Proxy proxy; | ||
|
|
||
| static { | ||
| container.start(); | ||
|
|
||
| toxiproxy.start(); | ||
|
|
||
| final ToxiproxyClient toxiproxyClient = new ToxiproxyClient(toxiproxy.getHost(), toxiproxy.getControlPort()); | ||
| try { | ||
| proxy = toxiproxyClient.createProxy("everything-server", "0.0.0.0:3000", "everything-server:3001"); | ||
| } | ||
| catch (IOException e) { | ||
| throw new RuntimeException("Can't create proxy!", e); | ||
| } | ||
|
|
||
| final String ipAddressViaToxiproxy = toxiproxy.getHost(); | ||
| final int portViaToxiproxy = toxiproxy.getMappedPort(3000); | ||
|
|
||
| host = "http://" + ipAddressViaToxiproxy + ":" + portViaToxiproxy; | ||
| } | ||
|
|
||
| static void disconnect() { | ||
| long start = System.nanoTime(); | ||
| try { | ||
| proxy.toxics().resetPeer("RESET_DOWNSTREAM", ToxicDirection.DOWNSTREAM, 0); | ||
| proxy.toxics().resetPeer("RESET_UPSTREAM", ToxicDirection.UPSTREAM, 0); | ||
| logger.info("Disconnect took {} ms", Duration.ofNanos(System.nanoTime() - start).toMillis()); | ||
| } | ||
| catch (IOException e) { | ||
| throw new RuntimeException("Failed to disconnect", e); | ||
| } | ||
| } | ||
|
|
||
| static void reconnect() { | ||
| long start = System.nanoTime(); | ||
| try { | ||
| proxy.toxics().get("RESET_UPSTREAM").remove(); | ||
| proxy.toxics().get("RESET_DOWNSTREAM").remove(); | ||
| logger.info("Reconnect took {} ms", Duration.ofNanos(System.nanoTime() - start).toMillis()); | ||
| } | ||
| catch (IOException e) { | ||
| throw new RuntimeException("Failed to reconnect", e); | ||
| } | ||
| } |
There was a problem hiding this comment.
It looks like this is copied from the resiliency tests, can we unify it in some shared test util?
| disconnect(); | ||
|
|
||
| // Veryfiy that the exception type is IOException and not TimeoutException | ||
| StepVerifier.create(mcpAsyncClient.ping()).expectError(IOException.class).verify(); |
|
|
||
| disconnect(); | ||
|
|
||
| StepVerifier.create(mcpAsyncClient.ping()).expectError(CompletionException.class).verify(); |
There was a problem hiding this comment.
Should we look at the cause to find an IOException maybe to make sure it's not a CompletableFuture's timeout?
| * tools, resources, prompts, etc. | ||
| * | ||
| * @author Dariusz Jędrzejczyk | ||
| */ |
There was a problem hiding this comment.
We need to add a note that it's a copy from mcp-test module as we do in other copied test classes to make sure we keep them in sync.
|
|
||
| } | ||
|
|
||
| public static class BodylessResponseLineSubscriber extends BaseSubscriber<String> { |
There was a problem hiding this comment.
Suggested change
| public static class BodylessResponseLineSubscriber extends BaseSubscriber<String> { | |
| public static class BodilessResponseLineSubscriber extends BaseSubscriber<String> { |
There was a problem hiding this comment.
| * Represents a Server-Sent Event with its standard fields. | ||
| * | ||
| * @param id the event ID, may be {@code null} | ||
| * @param event the event type, may be {@code null} (defaults to "message") |
There was a problem hiding this comment.
The record does not impose any default.
| public static record SseEvent(String id, String event, String data) { | ||
| } | ||
|
|
||
| public record ResponseEvent(ResponseInfo responseInfo, SseEvent sseEvent, JSONRPCMessage jsonRpcMessage) { |
There was a problem hiding this comment.
Seems responseInfo should not be null, we should add a constructor with a check.
Comment on lines
+159
to
+174
| var matcher = EVENT_DATA_PATTERN.matcher(line); | ||
| if (matcher.find()) { | ||
| this.eventBuilder.append(matcher.group(1).trim()).append("\n"); | ||
| } | ||
| } | ||
| else if (line.startsWith("id:")) { | ||
| var matcher = EVENT_ID_PATTERN.matcher(line); | ||
| if (matcher.find()) { | ||
| this.currentEventId.set(matcher.group(1).trim()); | ||
| } | ||
| } | ||
| else if (line.startsWith("event:")) { | ||
| var matcher = EVENT_TYPE_PATTERN.matcher(line); | ||
| if (matcher.find()) { | ||
| this.currentEventType.set(matcher.group(1).trim()); | ||
| } |
There was a problem hiding this comment.
Actually, matchers are quite slow. Perhaps we can replace the data/id/event with line.substring(5).trim() where 5 is the length of "data:"?
Comment on lines
118
to
121
| /** | ||
| * Initializes the subscription and sets up disposal callback. | ||
| * @param subscription the {@link Subscription} to the upstream line source | ||
| */ |
There was a problem hiding this comment.
We don't need these javadocs on protected methods.
Comment on lines
139
to
144
| /** | ||
| * Processes each line from the SSE stream according to the SSE protocol. Empty | ||
| * lines trigger event emission, other lines are parsed for data, id, or event | ||
| * type. | ||
| * @param line the line to process from the SSE stream | ||
| */ |
There was a problem hiding this comment.
This javadoc should only be needed at the top of the class.
Signed-off-by: Dariusz Jędrzejczyk <dariusz.jedrzejczyk@broadcom.com>
Signed-off-by: Dariusz Jędrzejczyk <dariusz.jedrzejczyk@broadcom.com>
Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
chemicL
reviewed
Jul 3, 2025
| @@ -373,7 +375,7 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sendMessage) { | |||
| else { | |||
| logger.debug("SSE connection established successfully"); | |||
| } | |||
| })).onErrorComplete().subscribe(); | |||
| })).onErrorMap(CompletionException.class, t -> t.getCause()).onErrorComplete().subscribe(); | |||
There was a problem hiding this comment.
this is not necessary, since it's swallowed in the next operator. And that's ok since we emit the throwable to the sink. We can clean this up afterwards.
chemicL
approved these changes
Jul 3, 2025
tzolov
added a commit
that referenced
this pull request
Jul 3, 2025
…r handling (#337) - Add HttpClientStreamableHttpTransport implementing the 2025-03-26 MCP specification - Add ResponseSubscribers utility for handling different HTTP response body types - Refactor HttpClientSseClientTransport to use reactive streams instead of FlowSseClient - Remove FlowSseClient in favor of reactive stream patterns - Use Disposable-based connection management instead of CountDownLatch - Replace message endpoint discovery with Sinks.One approach - Improve error handling by replacing doOnError/onErrorResume with onErrorComplete - Add test coverage for new Streamable HTTP transport - Add resiliency tests for connection handling and session invalidation BREAKING CHANGE: FlowSseClient has been removed in favor of reactive streams Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com> Co-authored-by: Dariusz Jędrzejczyk <dariusz.jedrzejczyk@broadcom.com>
|
Rebased, squashed and merged at c711f83 |
7 tasks
This was referenced Jul 3, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This change improves the transport layer with reactive patterns and adds support for the latest MCP specification while maintaining backward compatibility with existing SSE transport.