Fix SSE event parsing and error signalling in HttpClient transports

The SSE parser now follows the SSE spec: an event's type resets after
each event and defaults to "message", and an empty id clears the last
event id. Before, the type carried over from the previous event, so a
message following an "endpoint" event was taken for another endpoint.

Behaviour changes compared to main:

- Errors caused by the server (non-2xx responses, unknown content
  types, malformed messages) are McpTransportException instead of
  plain RuntimeException. Error messages no longer include the
  response body.
- A 400 on the streamable GET stream no longer invalidates the session.
- A non-2xx SSE connect with an empty body now fails instead of
  hanging, and sendMessage() completes when the server closes an SSE
  response without sending any events.
- Errors that happen after connect() or sendMessage() has completed
  now reach the exception handler and stop showing up as Reactor
  onErrorDropped logs.

Signed-off-by: Dariusz Jędrzejczyk <dariusz.jedrzejczyk@broadcom.com>

Author: chemicL

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientSseClientTransport.java

@@ -11,6 +11,7 @@
 import java.net.http.HttpResponse;
 import java.time.Duration;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -388,6 +389,14 @@ public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> h
 			var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
 			return Mono.from(this.httpRequestCustomizer.customize(builder, "GET", uri, null, transportContext));
 		}).flatMap(requestBuilder -> Mono.create(sink -> {
+			// Once connect() has completed, a later failure can no longer be reported
+			// through its sink: signalling it there would only have Reactor drop it.
+			AtomicBoolean connected = new AtomicBoolean();
+			Runnable markConnected = () -> {
+				if (connected.compareAndSet(false, true)) {
+					sink.success();
+				}
+			};
 			Disposable connection = Mono
 				.fromFuture(() -> this.httpClient.sendAsync(requestBuilder.build(),
 						HttpResponse.BodyHandlers.ofPublisher()))
@@ -407,7 +416,7 @@ public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> h
 					}
 					else {
 						return ResponseBodyHandlers.drainThenError(response.body(), this.maxResponseSize,
-								new RuntimeException("Failed to connect to SSE stream: " + statusCode));
+								new McpTransportException("Failed to connect to SSE stream: " + statusCode));
 					}
 				})
 				.flatMap(sseEvent -> {
@@ -418,44 +427,44 @@ public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> h
 								messageEndpointValidator.validate(uri, messageEndpointUri);
 							}
 							catch (InvalidSseMessageEndpointException e) {
-								sink.error(e);
 								this.messageEndpointSink.tryEmitError(e);
 								return Flux.error(e);
 							}
 							if (this.messageEndpointSink.tryEmitValue(messageEndpointUri).isSuccess()) {
-								sink.success();
+								markConnected.run();
 								return Flux.empty(); // No further processing needed
 							}
-							else {
-								sink.error(new RuntimeException("Failed to handle SSE endpoint event"));
-							}
+							return Flux.error(new McpTransportException("Failed to handle SSE endpoint event"));
 						}
 						else if (MESSAGE_EVENT_TYPE.equals(sseEvent.event())) {
 							String data = sseEvent.data();
 							if (data == null || data.isBlank()) {
 								logger.debug("Skipping SSE event with empty data (stream primer)");
-								sink.success();
+								markConnected.run();
 								return Flux.<McpSchema.JSONRPCMessage>empty();
 							}
 							JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(jsonMapper, data);
-							sink.success();
+							markConnected.run();
 							return Flux.just(message);
 						}
 						else {
 							logger.debug("Received unrecognized SSE event type: {}", sseEvent);
-							sink.success();
+							markConnected.run();
+							return Flux.<McpSchema.JSONRPCMessage>empty();
 						}
 					}
 					catch (IOException e) {
-						sink.error(new McpTransportException("Error processing SSE event", e));
+						return Flux.<McpSchema.JSONRPCMessage>error(
+								new McpTransportException("Error processing SSE event", e));
 					}
-					return Flux.<McpSchema.JSONRPCMessage>empty();
 				})
 				.flatMap(jsonRpcMessage -> handler.apply(Mono.just(jsonRpcMessage)))
 				.onErrorComplete(t -> {
 					if (!isClosing) {
 						logger.warn("SSE stream observed an error", t);
-						sink.error(t);
+						if (connected.compareAndSet(false, true)) {
+							sink.error(t);
+						}
 					}
 					return true;
 				})
@@ -540,7 +549,7 @@ private Mono<Void> sendHttpPost(final String endpoint, final String body) {
 						return ResponseBodyHandlers.drain(response.body(), this.maxResponseSize).then();
 					}
 					return ResponseBodyHandlers.decodeAggregateResponse(response.body(), this.maxResponseSize)
-					                           .flatMap(text -> Mono.error(new RuntimeException(
+						.flatMap(text -> Mono.error(new McpTransportException(
 								"Sending message failed with a non-OK HTTP code: " + statusCode + " - " + text)));
 				});
 		});

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java

@@ -403,15 +403,11 @@ else if (statusCode == NOT_FOUND) {
 							}
 						}
 						else if (statusCode == BAD_REQUEST) {
-							// if the session id was set, but the session no longer
-							// exists, some servers can return 400 instead of 404
-							if (maybeSessionId.isPresent()) {
-								String sessionIdRepresentation = sessionIdOrPlaceholder(maybeSessionId);
-								exception = new McpTransportSessionNotFoundException(sessionIdRepresentation);
-							}
-							else {
-								exception = new McpTransportException("Bad Request. Status code:" + statusCode);
-							}
+							// Unlike a POST, a GET is not treated as a session-not-found
+							// signal on 400: servers also reject the listening stream
+							// itself with 400, which must not cost the client its
+							// session.
+							exception = new McpTransportException("Bad Request. Status code:" + statusCode);
 						}
 						else if (statusCode >= 200 && statusCode < 300) {
 							String contentType = httpResponse.headers()
@@ -521,6 +517,15 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sentMessage) {
 
 			final AtomicReference<Disposable> disposableRef = new AtomicReference<>();
 
+			// Once sendMessage() has completed, a later failure can no longer be reported
+			// through its sink: signalling it there would only have Reactor drop it.
+			final AtomicBoolean delivered = new AtomicBoolean();
+			final Runnable markDelivered = () -> {
+				if (delivered.compareAndSet(false, true)) {
+					deliveredSink.success();
+				}
+			};
+
 			Disposable connection = Mono.deferContextual(ctx -> {
 				HttpRequest.Builder requestBuilder = this.requestBuilder.copy();
 
@@ -579,22 +584,17 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sentMessage) {
 
 							if (contentType.isBlank() || "0".equals(contentLength) || statusCode == 202) {
 								logger.debug("No body returned for POST in session {}", sessionRepresentation);
-								deliveredSink.success();
+								markDelivered.run();
 								return ResponseBodyHandlers.drain(httpResponse.body(), this.maxResponseSize);
 							}
 							else if (contentType.contains(TEXT_EVENT_STREAM)) {
-								AtomicBoolean delivered = new AtomicBoolean();
-								return consumeSseStream(httpResponse.body(), null, () -> {
-									if (delivered.compareAndSet(false, true)) {
-										deliveredSink.success();
-									}
-								});
+								return consumeSseStream(httpResponse.body(), null, markDelivered);
 							}
 							else if (contentType.contains(APPLICATION_JSON)) {
 								return ResponseBodyHandlers
 									.decodeAggregateResponse(httpResponse.body(), this.maxResponseSize)
 									.flatMapMany(data -> {
-										deliveredSink.success();
+										markDelivered.run();
 										if (sentMessage instanceof McpSchema.JSONRPCNotification) {
 											logger.warn("Notification: {} received non-compliant response: {}",
 													sentMessage, Utils.hasText(data) ? data : "[empty]");
@@ -613,7 +613,7 @@ else if (contentType.contains(APPLICATION_JSON)) {
 							logger.warn("Unknown media type {} returned for POST in session {}", contentType,
 									sessionRepresentation);
 							return ResponseBodyHandlers.drainThenError(httpResponse.body(), this.maxResponseSize,
-									new RuntimeException("Unknown media type returned: " + contentType));
+									new McpTransportException("Unknown media type returned: " + contentType));
 						}
 						else if (statusCode == NOT_FOUND) {
 							if (maybeSessionId.isPresent()) {
@@ -640,7 +640,7 @@ else if (statusCode >= 400 && statusCode < 500) {
 						}
 
 						return ResponseBodyHandlers.drainThenError(httpResponse.body(), this.maxResponseSize,
-								new RuntimeException("Failed to send message, status code: " + statusCode));
+								new McpTransportException("Failed to send message, status code: " + statusCode));
 					})
 					.onErrorMap(CompletionException.class, Throwable::getCause))
 				.retryWhen(authorizationErrorRetrySpec())
@@ -660,10 +660,16 @@ else if (statusCode >= 400 && statusCode < 500) {
 					catch (Exception e) {
 						logger.error("Error handling exception {}", t.getMessage(), e);
 					}
-					// inform the caller of sendMessage
-					deliveredSink.error(t);
+					// inform the caller of sendMessage, unless it has already completed
+					if (delivered.compareAndSet(false, true)) {
+						deliveredSink.error(t);
+					}
 					return true;
 				})
+				// An exchange can end without anything having signalled delivery, e.g.
+				// an SSE response closed before its first event. The server accepted
+				// the message all the same, so sendMessage() must not be left pending.
+				.doOnComplete(markDelivered)
 				.contextWrite(deliveredSink.contextView())
 				.subscribe();
 

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/ResponseBodyHandlers.java

@@ -47,6 +47,11 @@ class ResponseBodyHandlers {
 	 */
 	private static final int SSE_FRAMING_OVERHEAD = "event: ".length();
 
+	/**
+	 * The type of an SSE event that does not name one with an {@code event:} field.
+	 */
+	private static final String DEFAULT_EVENT_TYPE = "message";
+
 	record SseEvent(String id, String event, String data) {
 	}
 
@@ -368,11 +373,17 @@ private int indexOfLineTerminator(int from) {
 
 	/**
 	 * Stateful SSE line parser. Accumulates {@code data:}, {@code id:} and {@code event:}
-	 * fields until a blank line dispatches the event. Per the SSE spec, {@code id} and
-	 * {@code event} persist across events until re-set; {@code data} is reset after each
-	 * dispatch, and a blank line dispatches only when a {@code data:} field was seen,
-	 * whether or not it carried a value. Comments and fields the parser does not handle,
-	 * such as {@code retry:}, are ignored as the spec requires.
+	 * fields until a blank line dispatches the event. Per the SSE spec, {@code id} is the
+	 * last event ID and persists across events until re-set, with an empty value clearing
+	 * it; {@code event} and {@code data} are reset by every blank line, so an event that
+	 * does not name its type is a {@code message} event whatever preceded it. A blank
+	 * line dispatches only when a {@code data:} field was seen, whether or not it carried
+	 * a value. Comments and fields the parser does not handle, such as {@code retry:},
+	 * are ignored as the spec requires.
+	 *
+	 * @see <a href=
+	 * "https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation">Interpreting
+	 * an event stream</a>
 	 */
 	static final class SseEventParser {
 
@@ -399,12 +410,7 @@ static final class SseEventParser {
 
 		Optional<SseEvent> feed(String line) {
 			if (line.isEmpty()) {
-				if (data.length() == 0) {
-					return Optional.empty();
-				}
-				SseEvent result = new SseEvent(id, event, data.toString().trim());
-				data.setLength(0);
-				return Optional.of(result);
+				return flush();
 			}
 			if (line.startsWith("data:")) {
 				// Every data field appends its value followed by a separator, so a
@@ -422,16 +428,16 @@ Optional<SseEvent> feed(String line) {
 				data.append(value).append('\n');
 			}
 			else if (line.startsWith("id:")) {
-				String rest = line.substring(3);
-				if (!rest.isEmpty()) {
-					id = rest.trim();
+				String value = line.substring(3).trim();
+				// The spec ignores an id carrying a NULL, and an empty id resets the last
+				// event ID, which leaves nothing to resume from.
+				if (value.indexOf('\0') == -1) {
+					id = value.isEmpty() ? null : value;
 				}
 			}
 			else if (line.startsWith("event:")) {
-				String rest = line.substring(6);
-				if (!rest.isEmpty()) {
-					event = rest.trim();
-				}
+				String value = line.substring(6).trim();
+				event = value.isEmpty() ? null : value;
 			}
 			else if (line.startsWith(":")) {
 				logger.debug("Ignoring comment line: {}", line);
@@ -444,11 +450,19 @@ else if (line.startsWith(":")) {
 			return Optional.empty();
 		}
 
+		/**
+		 * Emits the pending event, if a {@code data:} field was seen, and resets the
+		 * per-event state. The event type is reset even when nothing is dispatched, as
+		 * the spec requires, while the id is the last event ID and so survives. An event
+		 * that did not name its type is emitted as a {@code message} event.
+		 */
 		Optional<SseEvent> flush() {
-			if (data.length() == 0) {
+			String type = this.event;
+			this.event = null;
+			if (data.isEmpty()) {
 				return Optional.empty();
 			}
-			SseEvent result = new SseEvent(id, event, data.toString().trim());
+			SseEvent result = new SseEvent(id, type != null ? type : DEFAULT_EVENT_TYPE, data.toString().trim());
 			data.setLength(0);
 			return Optional.of(result);
 		}

mcp-core/src/test/java/io/modelcontextprotocol/client/transport/SseEventParserTests.java

@@ -23,7 +23,7 @@ void simpleDataEvent() {
 		assertThat(event).isPresent();
 		assertThat(event.get().data()).isEqualTo("hello");
 		assertThat(event.get().id()).isNull();
-		assertThat(event.get().event()).isNull();
+		assertThat(event.get().event()).isEqualTo("message");
 	}
 
 	@Test
@@ -50,22 +50,55 @@ void idAndEventFieldsCaptured() {
 	}
 
 	@Test
-	void idAndEventPersistAcrossEvents() {
+	void idPersistsAcrossEventsButEventTypeDoesNot() {
 		SseEventParser p = new SseEventParser(Integer.MAX_VALUE);
 		p.feed("id: 1");
-		p.feed("event: message");
+		p.feed("event: endpoint");
 		p.feed("data: one");
 		SseEvent first = p.feed("").orElseThrow();
 		assertThat(first.id()).isEqualTo("1");
-		assertThat(first.event()).isEqualTo("message");
+		assertThat(first.event()).isEqualTo("endpoint");
 
+		// An event that does not name its type is a message event, not another
+		// endpoint event
 		p.feed("data: two");
 		SseEvent second = p.feed("").orElseThrow();
 		assertThat(second.id()).isEqualTo("1");
 		assertThat(second.event()).isEqualTo("message");
 		assertThat(second.data()).isEqualTo("two");
 	}
 
+	@Test
+	void blankLineWithNoDataStillResetsEventType() {
+		SseEventParser p = new SseEventParser(Integer.MAX_VALUE);
+		p.feed("event: endpoint");
+		assertThat(p.feed("")).isEmpty();
+		p.feed("data: payload");
+		SseEvent event = p.feed("").orElseThrow();
+		assertThat(event.event()).isEqualTo("message");
+	}
+
+	@Test
+	void emptyIdClearsLastEventId() {
+		SseEventParser p = new SseEventParser(Integer.MAX_VALUE);
+		p.feed("id: 1");
+		p.feed("data: one");
+		assertThat(p.feed("").orElseThrow().id()).isEqualTo("1");
+
+		p.feed("id:");
+		p.feed("data: two");
+		assertThat(p.feed("").orElseThrow().id()).isNull();
+	}
+
+	@Test
+	void idContainingNullIsIgnored() {
+		SseEventParser p = new SseEventParser(Integer.MAX_VALUE);
+		p.feed("id: 1");
+		p.feed("id: 2\u0000" + "3");
+		p.feed("data: payload");
+		assertThat(p.feed("").orElseThrow().id()).isEqualTo("1");
+	}
+
 	@Test
 	void commentLineIgnored() {
 		SseEventParser p = new SseEventParser(Integer.MAX_VALUE);