chore: update sync variants of callToolStream etc. to actually return Stream

Author: LucaButBoring

mcp-core/src/main/java/io/modelcontextprotocol/client/McpSyncClient.java

@@ -7,6 +7,7 @@
 import java.time.Duration;
 import java.util.List;
 import java.util.function.Supplier;
+import java.util.stream.Stream;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -310,8 +311,7 @@ public McpSchema.CreateTaskResult callToolTask(McpSchema.CallToolRequest callToo
 	 * var request = new CallToolRequest("my-tool", Map.of("arg", "value"),
 	 *     new TaskMetadata(60000L), null);  // Optional task metadata
 	 *
-	 * List<ResponseMessage<CallToolResult>> messages = client.callToolStream(request);
-	 * for (var message : messages) {
+	 * client.callToolStream(request).forEach(message -> {
 	 *     switch (message) {
 	 *         case TaskCreatedMessage<CallToolResult> tc ->
 	 *             System.out.println("Task created: " + tc.task().taskId());
@@ -322,21 +322,21 @@ public McpSchema.CreateTaskResult callToolTask(McpSchema.CallToolRequest callToo
 	 *         case ErrorMessage<CallToolResult> e ->
 	 *             System.err.println("Error: " + e.error().getMessage());
 	 *     }
-	 * }
+	 * });
 	 * }</pre>
 	 * @param callToolRequest The request containing the tool name and arguments. If the
 	 * {@code task} field is set, the call will be task-augmented.
-	 * @return A list of {@link McpSchema.ResponseMessage} instances representing the
+	 * @return A stream of {@link McpSchema.ResponseMessage} instances representing the
 	 * progress and result of the tool call.
 	 * @see McpSchema.ResponseMessage
 	 * @see McpSchema.TaskCreatedMessage
 	 * @see McpSchema.TaskStatusMessage
 	 * @see McpSchema.ResultMessage
 	 * @see McpSchema.ErrorMessage
 	 */
-	public List<McpSchema.ResponseMessage<McpSchema.CallToolResult>> callToolStream(
+	public Stream<McpSchema.ResponseMessage<McpSchema.CallToolResult>> callToolStream(
 			McpSchema.CallToolRequest callToolRequest) {
-		return withProvidedContextFlux(this.delegate.callToolStream(callToolRequest)).collectList().block();
+		return withProvidedContextFlux(this.delegate.callToolStream(callToolRequest)).toStream();
 	}
 
 	/**

mcp-core/src/main/java/io/modelcontextprotocol/server/McpSyncServerExchange.java

@@ -11,6 +11,7 @@
 import io.modelcontextprotocol.spec.McpSchema.LoggingMessageNotification;
 
 import java.util.List;
+import java.util.stream.Stream;
 
 /**
  * Represents a synchronous exchange with a Model Context Protocol (MCP) client. The
@@ -457,12 +458,12 @@ public McpSchema.CreateTaskResult createMessageTask(McpSchema.CreateMessageReque
 	 * releases.
 	 * @param createMessageRequest The request containing the sampling parameters. If the
 	 * {@code task} field is set, the call will be task-augmented.
-	 * @return A list of {@link McpSchema.ResponseMessage} instances representing the
+	 * @return A stream of {@link McpSchema.ResponseMessage} instances representing the
 	 * progress and result
 	 */
-	public List<McpSchema.ResponseMessage<McpSchema.CreateMessageResult>> createMessageStream(
+	public Stream<McpSchema.ResponseMessage<McpSchema.CreateMessageResult>> createMessageStream(
 			McpSchema.CreateMessageRequest createMessageRequest) {
-		return this.exchange.createMessageStream(createMessageRequest).collectList().block();
+		return this.exchange.createMessageStream(createMessageRequest).toStream();
 	}
 
 	// --------------------------
@@ -506,12 +507,12 @@ public McpSchema.CreateTaskResult createElicitationTask(McpSchema.ElicitRequest
 	 * releases.
 	 * @param elicitRequest The request containing the elicitation parameters. If the
 	 * {@code task} field is set, the call will be task-augmented.
-	 * @return A list of {@link McpSchema.ResponseMessage} instances representing the
+	 * @return A stream of {@link McpSchema.ResponseMessage} instances representing the
 	 * progress and result
 	 */
-	public List<McpSchema.ResponseMessage<McpSchema.ElicitResult>> createElicitationStream(
+	public Stream<McpSchema.ResponseMessage<McpSchema.ElicitResult>> createElicitationStream(
 			McpSchema.ElicitRequest elicitRequest) {
-		return this.exchange.createElicitationStream(elicitRequest).collectList().block();
+		return this.exchange.createElicitationStream(elicitRequest).toStream();
 	}
 
 }

mcp-core/src/test/java/io/modelcontextprotocol/server/AbstractMcpClientServerIntegrationTests.java

@@ -21,6 +21,7 @@
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import io.modelcontextprotocol.client.McpClient;
 import io.modelcontextprotocol.client.McpSyncClient;
@@ -1882,7 +1883,8 @@ protected McpSyncClient createTaskClient(String clientType, String name, ClientC
 	}
 
 	/** Extracts the task ID from a list of response messages. */
-	protected String extractTaskId(List<ResponseMessage<CallToolResult>> messages) {
+	protected String extractTaskId(Stream<ResponseMessage<CallToolResult>> messageStream) {
+		List<ResponseMessage<CallToolResult>> messages = messageStream.toList();
 		for (var msg : messages) {
 			if (msg instanceof TaskCreatedMessage<CallToolResult> tcm) {
 				return tcm.task().taskId();
@@ -1892,7 +1894,8 @@ protected String extractTaskId(List<ResponseMessage<CallToolResult>> messages) {
 	}
 
 	/** Extracts all task statuses from a list of response messages. */
-	protected List<TaskStatus> extractTaskStatuses(List<ResponseMessage<CallToolResult>> messages) {
+	protected List<TaskStatus> extractTaskStatuses(Stream<ResponseMessage<CallToolResult>> messageStream) {
+		List<ResponseMessage<CallToolResult>> messages = messageStream.toList();
 		List<TaskStatus> statuses = new ArrayList<>();
 		for (var msg : messages) {
 			if (msg instanceof TaskCreatedMessage<CallToolResult> tcm) {
@@ -2093,7 +2096,7 @@ void testAutomaticPollingShimWithCreateTaskHandler(String clientType) {
 
 			// Call tool WITHOUT task metadata - should trigger automatic polling shim
 			var request = new McpSchema.CallToolRequest("auto-polling-tool", Map.of("input", "test-value"), null, null);
-			var messages = client.callToolStream(request);
+			var messages = client.callToolStream(request).toList();
 
 			// The automatic polling shim should poll and return the final result
 			assertThat(messages).as("Should have response messages").isNotEmpty();

mcp-test/src/main/java/io/modelcontextprotocol/AbstractMcpClientServerIntegrationTests.java

@@ -22,6 +22,7 @@
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import io.modelcontextprotocol.client.McpClient;
 import io.modelcontextprotocol.client.McpSyncClient;
@@ -1845,7 +1846,8 @@ protected McpSyncClient createTaskClient(String clientType, String name, ClientC
 	}
 
 	/** Extracts the task ID from a list of response messages. */
-	protected String extractTaskId(List<ResponseMessage<CallToolResult>> messages) {
+	protected String extractTaskId(Stream<ResponseMessage<CallToolResult>> messageStream) {
+		List<ResponseMessage<CallToolResult>> messages = messageStream.toList();
 		for (var msg : messages) {
 			if (msg instanceof TaskCreatedMessage<CallToolResult> tcm) {
 				return tcm.task().taskId();
@@ -1855,7 +1857,8 @@ protected String extractTaskId(List<ResponseMessage<CallToolResult>> messages) {
 	}
 
 	/** Extracts all task statuses from a list of response messages. */
-	protected List<TaskStatus> extractTaskStatuses(List<ResponseMessage<CallToolResult>> messages) {
+	protected List<TaskStatus> extractTaskStatuses(Stream<ResponseMessage<CallToolResult>> messageStream) {
+		List<ResponseMessage<CallToolResult>> messages = messageStream.toList();
 		List<TaskStatus> statuses = new ArrayList<>();
 		for (var msg : messages) {
 			if (msg instanceof TaskCreatedMessage<CallToolResult> tcm) {
@@ -2181,7 +2184,7 @@ void testAutomaticPollingShimWithCreateTaskHandler(String clientType) {
 
 			// Call tool WITHOUT task metadata - should trigger automatic polling shim
 			var request = new McpSchema.CallToolRequest("auto-polling-tool", Map.of("input", "test-value"), null, null);
-			var messages = client.callToolStream(request);
+			var messages = client.callToolStream(request).toList();
 
 			// The automatic polling shim should poll and return the final result
 			assertThat(messages).as("Should have response messages").isNotEmpty();