Merge branch 'main' into streamable-server-webmvc

Author: tzolov

mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/client/transport/WebClientStreamableHttpTransport.java

@@ -30,6 +30,7 @@
 import io.modelcontextprotocol.spec.McpTransportSessionNotFoundException;
 import io.modelcontextprotocol.spec.McpTransportStream;
 import io.modelcontextprotocol.util.Assert;
+import io.modelcontextprotocol.util.Utils;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -244,7 +245,7 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
 
 			Disposable connection = webClient.post()
 				.uri(this.endpoint)
-				.accept(MediaType.TEXT_EVENT_STREAM, MediaType.APPLICATION_JSON)
+				.accept(MediaType.APPLICATION_JSON, MediaType.TEXT_EVENT_STREAM)
 				.headers(httpHeaders -> {
 					transportSession.sessionId().ifPresent(id -> httpHeaders.add("mcp-session-id", id));
 				})
@@ -287,7 +288,7 @@ else if (mediaType.isCompatibleWith(MediaType.APPLICATION_JSON)) {
 								logger.trace("Received response to POST for session {}", sessionRepresentation);
 								// communicate to caller the message was delivered
 								sink.success();
-								return responseFlux(response);
+								return directResponseFlux(message, response);
 							}
 							else {
 								logger.warn("Unknown media type {} returned for POST in session {}", contentType,
@@ -385,14 +386,22 @@ private static String sessionIdOrPlaceholder(McpTransportSession<?> transportSes
 		return transportSession.sessionId().orElse("[missing_session_id]");
 	}
 
-	private Flux<McpSchema.JSONRPCMessage> responseFlux(ClientResponse response) {
+	private Flux<McpSchema.JSONRPCMessage> directResponseFlux(McpSchema.JSONRPCMessage sentMessage,
+			ClientResponse response) {
 		return response.bodyToMono(String.class).<Iterable<McpSchema.JSONRPCMessage>>handle((responseMessage, s) -> {
 			try {
-				McpSchema.JSONRPCMessage jsonRpcResponse = McpSchema.deserializeJsonRpcMessage(objectMapper,
-						responseMessage);
-				s.next(List.of(jsonRpcResponse));
+				if (sentMessage instanceof McpSchema.JSONRPCNotification && Utils.hasText(responseMessage)) {
+					logger.warn("Notification: {} received non-compliant response: {}", sentMessage, responseMessage);
+					s.complete();
+				}
+				else {
+					McpSchema.JSONRPCMessage jsonRpcResponse = McpSchema.deserializeJsonRpcMessage(objectMapper,
+							responseMessage);
+					s.next(List.of(jsonRpcResponse));
+				}
 			}
 			catch (IOException e) {
+				// TODO: this should be a McpTransportError
 				s.error(e);
 			}
 		}).flatMapIterable(Function.identity());
@@ -424,7 +433,8 @@ private Tuple2<Optional<String>, Iterable<McpSchema.JSONRPCMessage>> parse(Serve
 			}
 		}
 		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());
 		}
 	}
 

mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/client/transport/WebFluxSseClientTransport.java

@@ -216,7 +216,8 @@ else if (MESSAGE_EVENT_TYPE.equals(event.event())) {
 				}
 			}
 			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();
 

mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/client/transport/WebFluxSseClientTransportTests.java

@@ -6,6 +6,7 @@
 
 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 @@ public int getInboundMessageCount() {
 			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 @@ void testBuilderPattern() {
 		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

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

@@ -366,10 +366,8 @@ else if (MESSAGE_EVENT_TYPE.equals(responseEvent.sseEvent().event())) {
 								return Flux.just(message);
 							}
 							else {
-								logger.error("Received unrecognized SSE event type: {}",
-										responseEvent.sseEvent().event());
-								sink.error(new McpError(
-										"Received unrecognized SSE event type: " + responseEvent.sseEvent().event()));
+								logger.debug("Received unrecognized SSE event type: {}", responseEvent.sseEvent());
+								sink.success();
 							}
 						}
 						catch (IOException e) {

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

@@ -264,6 +264,10 @@ private Mono<Disposable> reconnect(McpTransportStream<Disposable> stream) {
 										"Error parsing JSON-RPC message: " + responseEvent.sseEvent().data()));
 							}
 						}
+						else {
+							logger.debug("Received SSE event with type: {}", responseEvent.sseEvent());
+							return Flux.empty();
+						}
 					}
 					else if (statusCode == METHOD_NOT_ALLOWED) { // NotAllowed
 						logger.debug("The server does not support SSE streams, using request-response mode.");
@@ -342,9 +346,9 @@ public String toString(McpSchema.JSONRPCMessage message) {
 		}
 	}
 
-	public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sendMessage) {
+	public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sentMessage) {
 		return Mono.create(messageSink -> {
-			logger.debug("Sending message {}", sendMessage);
+			logger.debug("Sending message {}", sentMessage);
 
 			final AtomicReference<Disposable> disposableRef = new AtomicReference<>();
 			final McpTransportSession<Disposable> transportSession = this.activeSession.get();
@@ -355,10 +359,10 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sendMessage) {
 				requestBuilder = requestBuilder.header("mcp-session-id", transportSession.sessionId().get());
 			}
 
-			String jsonBody = this.toString(sendMessage);
+			String jsonBody = this.toString(sentMessage);
 
 			HttpRequest request = requestBuilder.uri(Utils.resolveUri(this.baseUri, this.endpoint))
-				.header("Accept", TEXT_EVENT_STREAM + ", " + APPLICATION_JSON)
+				.header("Accept", APPLICATION_JSON + ", " + TEXT_EVENT_STREAM)
 				.header("Content-Type", APPLICATION_JSON)
 				.header("Cache-Control", "no-cache")
 				.POST(HttpRequest.BodyPublishers.ofString(jsonBody))
@@ -436,10 +440,16 @@ else if (contentType.contains(TEXT_EVENT_STREAM)) {
 					else if (contentType.contains(APPLICATION_JSON)) {
 						messageSink.success();
 						String data = ((ResponseSubscribers.AggregateResponseEvent) responseEvent).data();
+						if (sentMessage instanceof McpSchema.JSONRPCNotification && Utils.hasText(data)) {
+							logger.warn("Notification: {} received non-compliant response: {}", sentMessage, data);
+							return Mono.empty();
+						}
+
 						try {
 							return Mono.just(McpSchema.deserializeJsonRpcMessage(objectMapper, data));
 						}
 						catch (IOException e) {
+							// TODO: this should be a McpTransportError
 							return Mono.error(e);
 						}
 					}

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

@@ -12,6 +12,7 @@
 import org.reactivestreams.FlowAdapters;
 import org.reactivestreams.Subscription;
 
+import io.modelcontextprotocol.spec.McpError;
 import reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.FluxSink;
 
@@ -135,6 +136,7 @@ protected void hookOnSubscribe(Subscription subscription) {
 
 		@Override
 		protected void hookOnNext(String line) {
+
 			if (line.isEmpty()) {
 				// Empty line means end of event
 				if (this.eventBuilder.length() > 0) {
@@ -164,6 +166,13 @@ else if (line.startsWith("event:")) {
 						this.currentEventType.set(matcher.group(1).trim());
 					}
 				}
+				else {
+					// If the response is not successful, emit an error
+					// TODO: This should be a McpTransportError
+					this.sink.error(new McpError(
+							"Invalid SSE response. Status code: " + this.responseInfo.statusCode() + " Line: " + line));
+
+				}
 			}
 		}
 

mcp/src/main/java/io/modelcontextprotocol/spec/McpStreamableServerTransportProvider.java

@@ -41,4 +41,27 @@ public interface McpStreamableServerTransportProvider extends McpServerTransport
 	 */
 	void setSessionFactory(McpStreamableServerSession.Factory sessionFactory);
 
+	/**
+	 * Sends a notification to all connected clients.
+	 * @param method the name of the notification method to be called on the clients
+	 * @param params parameters to be sent with the notification
+	 * @return a Mono that completes when the notification has been broadcast
+	 */
+	Mono<Void> notifyClients(String method, Object params);
+
+	/**
+	 * Immediately closes all the transports with connected clients and releases any
+	 * associated resources.
+	 */
+	default void close() {
+		this.closeGracefully().subscribe();
+	}
+
+	/**
+	 * Gracefully closes all the transports with connected clients and releases any
+	 * associated resources asynchronously.
+	 * @return a {@link Mono} that completes when the connections have been closed.
+	 */
+	Mono<Void> closeGracefully();
+
 }

mcp/src/main/java/io/modelcontextprotocol/util/Utils.java

@@ -69,6 +69,9 @@ public static boolean isEmpty(@Nullable Map<?, ?> map) {
 	 * base URL or URI is malformed
 	 */
 	public static URI resolveUri(URI baseUrl, String endpointUrl) {
+		if (!Utils.hasText(endpointUrl)) {
+			return baseUrl;
+		}
 		URI endpointUri = URI.create(endpointUrl);
 		if (endpointUri.isAbsolute() && !isUnderBaseUri(baseUrl, endpointUri)) {
 			throw new IllegalArgumentException("Absolute endpoint URL does not match the base URL.");

mcp/src/test/java/io/modelcontextprotocol/server/AbstractMcpAsyncServerTests.java

@@ -64,7 +64,6 @@ void tearDown() {
 	// ---------------------------------------
 	// Server Lifecycle Tests
 	// ---------------------------------------
-
 	void testConstructorWithInvalidArguments() {
 		assertThatThrownBy(() -> McpServer.async((McpServerTransportProvider) null))
 			.isInstanceOf(IllegalArgumentException.class)