Skip to content

Commit bbb3330

Browse files
committed
Minor fix on stream primer
Signed-off-by: Daniel Garnier-Moiroux <git@garnier.wf>
1 parent eb4745f commit bbb3330

4 files changed

Lines changed: 113 additions & 19 deletions

File tree

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

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -431,7 +431,13 @@ public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> h
431431
}
432432
}
433433
else if (MESSAGE_EVENT_TYPE.equals(sseEvent.event())) {
434-
JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(jsonMapper, sseEvent.data());
434+
String data = sseEvent.data();
435+
if (data == null || data.isBlank()) {
436+
logger.debug("Skipping SSE event with empty data (stream primer)");
437+
sink.success();
438+
return Flux.<McpSchema.JSONRPCMessage>empty();
439+
}
440+
JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(jsonMapper, data);
435441
sink.success();
436442
return Flux.just(message);
437443
}

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

Lines changed: 14 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -355,7 +355,8 @@ private int indexOfLineTerminator(int from) {
355355
* Stateful SSE line parser. Accumulates {@code data:}, {@code id:} and {@code event:}
356356
* fields until a blank line dispatches the event. Per the SSE spec, {@code id} and
357357
* {@code event} persist across events until re-set; {@code data} is reset after each
358-
* dispatch.
358+
* dispatch, and a blank line dispatches only when a {@code data:} field was seen,
359+
* whether or not it carried a value.
359360
*/
360361
static final class SseEventParser {
361362

@@ -390,18 +391,19 @@ Optional<SseEvent> feed(String line) {
390391
return Optional.of(result);
391392
}
392393
if (line.startsWith("data:")) {
393-
String rest = line.substring(5);
394-
if (!rest.isEmpty()) {
395-
String value = rest.trim();
396-
// Measured before appending, so that an event carrying exactly
397-
// maxSize of data is accepted: the trailing separator below is
398-
// stripped again before the event is emitted.
399-
if (data.length() + value.length() > this.maxSize) {
400-
throw new McpTransportException(
401-
"Inbound SSE event exceeds the maximum allowed size of " + this.maxSize + " bytes");
402-
}
403-
data.append(value).append('\n');
394+
// Every data field appends its value followed by a separator, so a
395+
// valueless `data:` line still marks the event as carrying data and gets
396+
// dispatched with empty data. Servers send such an event to prime a
397+
// stream, and dropping it leaves the request it answers hanging.
398+
String value = line.substring(5).trim();
399+
// Measured before appending, so that an event carrying exactly
400+
// maxSize of data is accepted: the trailing separator below is
401+
// stripped again before the event is emitted.
402+
if (data.length() + value.length() > this.maxSize) {
403+
throw new McpTransportException(
404+
"Inbound SSE event exceeds the maximum allowed size of " + this.maxSize + " bytes");
404405
}
406+
data.append(value).append('\n');
405407
}
406408
else if (line.startsWith("id:")) {
407409
String rest = line.substring(3);

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

Lines changed: 33 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -108,13 +108,42 @@ void unknownFieldThrowsMcpTransportException() {
108108
}
109109

110110
@Test
111-
void dataFieldWithEmptyValueIsIgnored() {
112-
// preserves the pre-refactor regex behavior where `^data:(.+)$` required at
113-
// least one char after the colon — a lone `data:` line contributes nothing.
111+
void dataFieldWithEmptyValueStillDispatchesAnEvent() {
112+
// Per the SSE spec a data field appends its value plus a separator, so a lone
113+
// `data:` line leaves the buffer non-empty and the event is dispatched carrying
114+
// empty data. Servers send exactly this to prime a stream, and dropping it leaves
115+
// the request the stream answers hanging.
114116
SseEventParser p = new SseEventParser(Integer.MAX_VALUE);
115117
assertThat(p.feed("data:")).isEmpty();
116-
// no pending data → blank line dispatches nothing
118+
Optional<SseEvent> event = p.feed("");
119+
assertThat(event).isPresent();
120+
assertThat(event.get().data()).isEmpty();
121+
}
122+
123+
@Test
124+
void dataFieldWithOnlyASpaceIsEquivalentToNoValue() {
125+
SseEventParser p = new SseEventParser(Integer.MAX_VALUE);
126+
assertThat(p.feed("data: ")).isEmpty();
127+
Optional<SseEvent> event = p.feed("");
128+
assertThat(event).isPresent();
129+
assertThat(event.get().data()).isEmpty();
130+
}
131+
132+
@Test
133+
void blankLineWithNoDataFieldDispatchesNothing() {
134+
// `event:` alone leaves the data buffer empty, which per the spec is not an event
135+
SseEventParser p = new SseEventParser(Integer.MAX_VALUE);
136+
assertThat(p.feed("event: message")).isEmpty();
117137
assertThat(p.feed("")).isEmpty();
118138
}
119139

140+
@Test
141+
void valuelessDataFieldIsDispatchedOnFlush() {
142+
SseEventParser p = new SseEventParser(Integer.MAX_VALUE);
143+
assertThat(p.feed("data:")).isEmpty();
144+
Optional<SseEvent> flushed = p.flush();
145+
assertThat(flushed).isPresent();
146+
assertThat(flushed.get().data()).isEmpty();
147+
}
148+
120149
}

mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransportEmptyJsonResponseTest.java renamed to mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransportEmptyResponseTests.java

Lines changed: 59 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,9 +11,12 @@
1111
import static org.mockito.Mockito.verify;
1212

1313
import java.io.IOException;
14+
import java.io.OutputStream;
1415
import java.net.InetSocketAddress;
1516
import java.net.URI;
1617
import java.net.URISyntaxException;
18+
import java.nio.charset.StandardCharsets;
19+
import java.util.Map;
1720

1821
import org.junit.jupiter.api.AfterAll;
1922
import org.junit.jupiter.api.BeforeAll;
@@ -29,18 +32,33 @@
2932
import reactor.test.StepVerifier;
3033

3134
/**
32-
* Handles emplty application/json response with 200 OK status code.
35+
* Handles 200 OK responses that carry no usable body, either as an empty application/json
36+
* document or as a text/event-stream containing nothing but a stream primer.
3337
*
3438
* @author codezkk
3539
*/
36-
public class HttpClientStreamableHttpTransportEmptyJsonResponseTest {
40+
public class HttpClientStreamableHttpTransportEmptyResponseTests {
3741

3842
static int PORT = TomcatTestUtil.findAvailablePort();
3943

4044
static String host = "http://localhost:" + PORT;
4145

4246
static HttpServer server;
4347

48+
/**
49+
* An SSE event with an {@code event:} field but no data, which some servers send to
50+
* open the response stream before any JSON-RPC payload is available. Note the
51+
* valueless {@code data:} field: per the SSE spec this is identical to {@code data: }
52+
* with a trailing space.
53+
* @see <a href=
54+
* "https://github.com/modelcontextprotocol/modelcontextprotocol/pull/1699">SEP-1699</a>
55+
*/
56+
private static final byte[] SSE_PRIMER = """
57+
event: message
58+
data:
59+
60+
""".getBytes(StandardCharsets.UTF_8);
61+
4462
@BeforeAll
4563
static void startContainer() throws IOException {
4664

@@ -53,6 +71,24 @@ static void startContainer() throws IOException {
5371
exchange.close();
5472
});
5573

74+
// 200 OK text/event-stream carrying only a primer, for POSTs. The
75+
// server-initiated GET stream is refused so that the transport falls back to
76+
// request-response mode and the POST is the only thing under test.
77+
server.createContext("/mcp-sse-primer", exchange -> {
78+
try (exchange) {
79+
if (!"POST".equals(exchange.getRequestMethod())) {
80+
exchange.sendResponseHeaders(405, -1);
81+
return;
82+
}
83+
exchange.getRequestBody().readAllBytes();
84+
exchange.getResponseHeaders().set("Content-Type", "text/event-stream");
85+
exchange.sendResponseHeaders(200, SSE_PRIMER.length);
86+
try (OutputStream out = exchange.getResponseBody()) {
87+
out.write(SSE_PRIMER);
88+
}
89+
}
90+
});
91+
5692
server.setExecutor(null);
5793
server.start();
5894
}
@@ -91,4 +127,25 @@ void testNotificationInitialized() throws URISyntaxException {
91127

92128
}
93129

130+
/**
131+
* A POST answered with {@code 200 text/event-stream} whose body holds only a stream
132+
* primer must still complete, because the primer tells the client the stream is live
133+
* and the message has been accepted. The primer's {@code data:} field carries no
134+
* value, so this only holds as long as such a field still produces an event: a parser
135+
* that drops it leaves no event to fire the transport's first-message callback, and
136+
* {@code sendMessage} then never completes at all.
137+
*/
138+
@Test
139+
@Timeout(5)
140+
void testNotificationAnsweredWithSsePrimerOnly() {
141+
142+
var transport = HttpClientStreamableHttpTransport.builder(host).endpoint("/mcp-sse-primer").build();
143+
144+
var testMessage = new McpSchema.JSONRPCNotification(McpSchema.JSONRPC_VERSION, "notifications/initialized",
145+
Map.of());
146+
147+
StepVerifier.create(transport.sendMessage(testMessage)).verifyComplete();
148+
149+
}
150+
94151
}

0 commit comments

Comments
 (0)