fix: improve SSE event handling to gracefully ignore unrecognized events (#423)
- Replace McpError exceptions with debug/warning logs for unrecognized SSE event types - Continue processing instead of failing when encountering unknown SSE events - Update transport implementations: - WebClientStreamableHttpTransport: return empty tuple instead of throwing - WebFluxSseClientTransport: complete stream instead of erroring - HttpClientSseClientTransport: call sink.success() instead of sink.error() - HttpClientStreamableHttpTransport: return empty Flux for unknown events This improves client resilience when servers send non-standard or future SSE event types. Resolves #272 , #223 , #93, #421 Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
This commit is contained in:
parent
4937fc1ef5
commit
c3a0b1855a
3 changed files with 37 additions and 6 deletions
|
|
@ -432,7 +432,8 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
|
|||
}
|
||||
}
|
||||
else {
|
||||
throw new McpError("Received unrecognized SSE event type: " + event.event());
|
||||
logger.debug("Received SSE event with type: {}", event);
|
||||
return Tuples.of(Optional.empty(), List.of());
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -216,7 +216,8 @@ public class WebFluxSseClientTransport implements McpClientTransport {
|
|||
}
|
||||
}
|
||||
else {
|
||||
s.error(new McpError("Received unrecognized SSE event type: " + event.event()));
|
||||
logger.debug("Received unrecognized SSE event type: {}", event);
|
||||
s.complete();
|
||||
}
|
||||
}).transform(handler)).subscribe();
|
||||
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ package io.modelcontextprotocol.client.transport;
|
|||
|
||||
import java.time.Duration;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.function.Function;
|
||||
|
||||
|
|
@ -77,6 +78,11 @@ class WebFluxSseClientTransportTests {
|
|||
return inboundMessageCount.get();
|
||||
}
|
||||
|
||||
public void simulateSseComment(String comment) {
|
||||
events.tryEmitNext(ServerSentEvent.<String>builder().comment(comment).build());
|
||||
inboundMessageCount.incrementAndGet();
|
||||
}
|
||||
|
||||
public void simulateEndpointEvent(String jsonMessage) {
|
||||
events.tryEmitNext(ServerSentEvent.<String>builder().event("endpoint").data(jsonMessage).build());
|
||||
inboundMessageCount.incrementAndGet();
|
||||
|
|
@ -158,6 +164,27 @@ class WebFluxSseClientTransportTests {
|
|||
assertThatCode(() -> transport4.closeGracefully().block()).doesNotThrowAnyException();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testCommentSseMessage() {
|
||||
// If the line starts with a character (:) are comment lins and should be ingored
|
||||
// https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation
|
||||
|
||||
CopyOnWriteArrayList<Throwable> droppedErrors = new CopyOnWriteArrayList<>();
|
||||
reactor.core.publisher.Hooks.onErrorDropped(droppedErrors::add);
|
||||
|
||||
try {
|
||||
// Simulate receiving the SSE comment line
|
||||
transport.simulateSseComment("sse comment");
|
||||
|
||||
StepVerifier.create(transport.closeGracefully()).verifyComplete();
|
||||
|
||||
assertThat(droppedErrors).hasSize(0);
|
||||
}
|
||||
finally {
|
||||
reactor.core.publisher.Hooks.resetOnErrorDropped();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void testMessageProcessing() {
|
||||
// Create a test message
|
||||
|
|
|
|||
Loading…
Reference in a new issue