fix: stop truncating SSE field values containing U+2028/U+2029/U+0085

Author: lzcGeek

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

@@ -13,7 +13,6 @@
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Flow;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.regex.Pattern;
 
 import org.reactivestreams.FlowAdapters;
 import org.reactivestreams.Subscription;
@@ -147,21 +146,6 @@ static BodyHandler<String> boundedStringBodyHandler(int maxSize) {
 
 	static class SseLineSubscriber extends BaseSubscriber<String> {
 
-		/**
-		 * Pattern to extract data content from SSE "data:" lines.
-		 */
-		private static final Pattern EVENT_DATA_PATTERN = Pattern.compile("^data:(.+)$", Pattern.MULTILINE);
-
-		/**
-		 * Pattern to extract event ID from SSE "id:" lines.
-		 */
-		private static final Pattern EVENT_ID_PATTERN = Pattern.compile("^id:(.+)$", Pattern.MULTILINE);
-
-		/**
-		 * Pattern to extract event type from SSE "event:" lines.
-		 */
-		private static final Pattern EVENT_TYPE_PATTERN = Pattern.compile("^event:(.+)$", Pattern.MULTILINE);
-
 		/**
 		 * The sink for emitting parsed response events.
 		 */
@@ -227,49 +211,75 @@ protected void hookOnSubscribe(Subscription subscription) {
 			});
 		}
 
+		/**
+		 * Extracts the value of an SSE field from a line, per the <a href=
+		 * "https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation">
+		 * SSE specification</a>: the characters after the colon with a single leading
+		 * space removed.
+		 *
+		 * <p>
+		 * A value may legally contain U+2028 (LINE SEPARATOR), U+2029 (PARAGRAPH
+		 * SEPARATOR) and U+0085 (NEXT LINE). Those are not SSE line terminators, so
+		 * extracting the value with a {@code MULTILINE} regex instead of this method
+		 * silently truncates it there.
+		 * @param line the SSE line, already stripped of its terminator by the line
+		 * subscriber
+		 * @param field the field prefix, e.g. {@code "data:"}
+		 * @return the field value with a single leading space removed, never truncated
+		 * @see #hookOnNext(String)
+		 */
+		private static String fieldValue(String line, String field) {
+			String value = line.substring(field.length());
+			if (value.startsWith(" ")) {
+				value = value.substring(1);
+			}
+			return value;
+		}
+
+		/**
+		 * Returns the buffered data lines joined by the separators the data: handler
+		 * appended, with only the final separator removed. Trimming the whole buffer
+		 * instead would also strip significant leading/trailing whitespace from the first
+		 * and last data lines, which the SSE field rules explicitly preserve.
+		 */
+		private String concatenatedDataLines() {
+			String buffered = this.eventBuilder.toString();
+			return buffered.substring(0, buffered.length() - 1);
+		}
+
 		@Override
 		protected void hookOnNext(String line) {
 			if (line.isEmpty()) {
 				// Empty line means end of event
 				if (this.eventBuilder.length() > 0) {
-					String eventData = this.eventBuilder.toString();
-					SseEvent sseEvent = new SseEvent(currentEventId.get(), currentEventType.get(), eventData.trim());
+					String eventData = concatenatedDataLines();
+					SseEvent sseEvent = new SseEvent(currentEventId.get(), currentEventType.get(), eventData);
 
 					this.sink.next(new SseResponseEvent(responseInfo, sseEvent));
 					this.eventBuilder.setLength(0);
 				}
 			}
 			else {
 				if (line.startsWith("data:")) {
-					var matcher = EVENT_DATA_PATTERN.matcher(line);
-					if (matcher.find()) {
-						String data = matcher.group(1).trim();
-						// Measured before appending, so that an event carrying exactly
-						// maxSize of data is accepted: the trailing separator below is
-						// stripped again before the event is emitted.
-						if (this.eventBuilder.length() + data.length() > this.maxSize) {
-							upstream().cancel();
-							this.sink.error(
-									new McpTransportException("Inbound SSE event exceeds the maximum allowed size of "
-											+ this.maxSize + " bytes"));
-							return;
-						}
-						this.eventBuilder.append(data).append("\n");
+					String data = fieldValue(line, "data:");
+					// Measured before appending, so that an event carrying exactly
+					// maxSize of data is accepted: the trailing separator below is
+					// stripped again before the event is emitted.
+					if (this.eventBuilder.length() + data.length() > this.maxSize) {
+						upstream().cancel();
+						this.sink.error(new McpTransportException(
+								"Inbound SSE event exceeds the maximum allowed size of " + this.maxSize + " bytes"));
+						return;
 					}
+					this.eventBuilder.append(data).append("\n");
 					upstream().request(1);
 				}
 				else if (line.startsWith("id:")) {
-					var matcher = EVENT_ID_PATTERN.matcher(line);
-					if (matcher.find()) {
-						this.currentEventId.set(matcher.group(1).trim());
-					}
+					this.currentEventId.set(fieldValue(line, "id:"));
 					upstream().request(1);
 				}
 				else if (line.startsWith("event:")) {
-					var matcher = EVENT_TYPE_PATTERN.matcher(line);
-					if (matcher.find()) {
-						this.currentEventType.set(matcher.group(1).trim());
-					}
+					this.currentEventType.set(fieldValue(line, "event:"));
 					upstream().request(1);
 				}
 				else if (line.startsWith(":")) {
@@ -290,8 +300,8 @@ else if (line.startsWith(":")) {
 		@Override
 		protected void hookOnComplete() {
 			if (this.eventBuilder.length() > 0) {
-				String eventData = this.eventBuilder.toString();
-				SseEvent sseEvent = new SseEvent(currentEventId.get(), currentEventType.get(), eventData.trim());
+				String eventData = concatenatedDataLines();
+				SseEvent sseEvent = new SseEvent(currentEventId.get(), currentEventType.get(), eventData);
 				this.sink.next(new SseResponseEvent(responseInfo, sseEvent));
 			}
 			this.sink.complete();

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

@@ -0,0 +1,140 @@
+/*
+ * Copyright 2024-2026 the original author or authors.
+ */
+
+package io.modelcontextprotocol.client.transport;
+
+import java.net.http.HttpClient;
+import java.net.http.HttpHeaders;
+import java.net.http.HttpResponse;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Unit tests for {@link ResponseSubscribers.SseLineSubscriber}.
+ *
+ * <p>
+ * Verifies that SSE field values are extracted per the <a href=
+ * "https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation">
+ * WHATWG HTML Living Standard §9.2.6</a>: the field value is everything after the colon
+ * minus a single leading space. In particular, U+2028 (LINE SEPARATOR), U+2029 (PARAGRAPH
+ * SEPARATOR) and U+0085 (NEXT LINE) are legal inside a field value and must not truncate
+ * it — they are not SSE line terminators.
+ *
+ * @see <a href="https://github.com/modelcontextprotocol/java-sdk/issues/1136">#1136</a>
+ */
+class ResponseSubscribersTest {
+
+	private static final HttpResponse.ResponseInfo RESPONSE_INFO = new HttpResponse.ResponseInfo() {
+
+		@Override
+		public int statusCode() {
+			return 200;
+		}
+
+		@Override
+		public HttpHeaders headers() {
+			return HttpHeaders.of(Map.of(), (name, value) -> true);
+		}
+
+		@Override
+		public HttpClient.Version version() {
+			return HttpClient.Version.HTTP_1_1;
+		}
+
+	};
+
+	private static List<ResponseSubscribers.SseEvent> parse(List<String> lines) {
+		return Flux.<ResponseSubscribers.ResponseEvent>create(sink -> Flux.fromIterable(lines)
+			.subscribe(new ResponseSubscribers.SseLineSubscriber(RESPONSE_INFO, sink, Integer.MAX_VALUE)))
+			.map(event -> ((ResponseSubscribers.SseResponseEvent) event).sseEvent())
+			.collectList()
+			.block();
+	}
+
+	/**
+	 * A {@code data:} payload containing U+2028, U+2029 or U+0085 must survive parsing
+	 * intact. A MULTILINE regex used to truncate the value at those characters, because
+	 * the Java regex engine treats them as line terminators.
+	 */
+	@Test
+	void shouldNotTruncateDataAtUnicodeLineSeparators() {
+		List<String> separators = List.of("\u2028", "\u2029", "\u0085");
+
+		for (String separator : separators) {
+			String payload = "{\"text\":\"a" + separator + "b\"}";
+			List<ResponseSubscribers.SseEvent> events = parse(List.of("data: " + payload, ""));
+
+			assertThat(events).as("payload with U+%04X", (int) separator.charAt(0)).hasSize(1);
+			assertThat(events.get(0).data()).isEqualTo(payload);
+		}
+	}
+
+	@Test
+	void shouldPreserveVerticalTabInData() {
+		String payload = "{\"text\":\"a\u000Bb\"}";
+
+		List<ResponseSubscribers.SseEvent> events = parse(List.of("data: " + payload, ""));
+
+		assertThat(events).hasSize(1);
+		assertThat(events.get(0).data()).isEqualTo(payload);
+	}
+
+	@Test
+	void shouldStripOnlySingleLeadingSpacePerDataLine() {
+		// The leading space after the colon is stripped per line; any further
+		// whitespace is part of the value.
+		List<ResponseSubscribers.SseEvent> events = parse(List.of("data: first", "data:  second", ""));
+
+		assertThat(events).hasSize(1);
+		assertThat(events.get(0).data()).isEqualTo("first\n second");
+	}
+
+	/**
+	 * Only the separator the {@code data:} handler appended after the last line may be
+	 * removed when the event is dispatched; trimming the whole buffer would also strip
+	 * significant whitespace from the first and last data lines.
+	 */
+	@Test
+	void shouldNotTrimSignificantWhitespaceOfSingleDataLine() {
+		List<ResponseSubscribers.SseEvent> events = parse(List.of("data:  padded ", ""));
+
+		assertThat(events).hasSize(1);
+		assertThat(events.get(0).data()).isEqualTo(" padded ");
+	}
+
+	@Test
+	void shouldPreserveLeadingAndTrailingWhitespaceOfFirstAndLastDataLines() {
+		List<ResponseSubscribers.SseEvent> events = parse(List.of("data:  first", "data: last ", ""));
+
+		assertThat(events).hasSize(1);
+		assertThat(events.get(0).data()).isEqualTo(" first\nlast ");
+	}
+
+	@Test
+	void shouldJoinMultipleDataLinesWithNewline() {
+		List<ResponseSubscribers.SseEvent> events = parse(List.of("data: first", "data: second", ""));
+
+		assertThat(events).hasSize(1);
+		assertThat(events.get(0).data()).isEqualTo("first\nsecond");
+	}
+
+	@Test
+	void shouldCaptureEventIdAndTypeWithUnicodeValue() {
+		List<ResponseSubscribers.SseEvent> events = parse(
+				List.of("event: message\u2028tail", "id: 42\u2028tail", "data: body", ""));
+
+		assertThat(events).hasSize(1);
+		assertThat(events.get(0).event()).isEqualTo("message\u2028tail");
+		assertThat(events.get(0).id()).isEqualTo("42\u2028tail");
+		assertThat(events.get(0).data()).isEqualTo("body");
+	}
+
+}