Skip to content

listXyz() methods return re-subscribable Monos #1175

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Kehrlann merged 1 commit into from
Oct 13, 2026
Merged

listXyz() methods return re-subscribable Monos #1175

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions mcp-core/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -734,7 +734,7 @@ public Mono<McpSchema.ListToolsResult> listTools() {
return this.listTools(McpSchema.FIRST_PAGE).expand(result -> {
String next = result.nextCursor();
return (next != null && !next.isEmpty()) ? this.listTools(next) : Mono.empty();
}).reduce(new ArrayList<McpSchema.Tool>(), (accumulated, result) -> {
}).reduceWith(ArrayList<McpSchema.Tool>::new, (accumulated, result) -> {
accumulated.addAll(result.tools());
return accumulated;
}).map(all -> McpSchema.ListToolsResult.builder(Collections.unmodifiableList(all)).build());
Expand Down Expand Up @@ -821,7 +821,7 @@ public Mono<McpSchema.ListResourcesResult> listResources() {
return this.listResources(McpSchema.FIRST_PAGE).expand(result -> {
String next = result.nextCursor();
return (next != null && !next.isEmpty()) ? this.listResources(next) : Mono.empty();
}).reduce(new ArrayList<McpSchema.Resource>(), (accumulated, result) -> {
}).reduceWith(ArrayList<McpSchema.Resource>::new, (accumulated, result) -> {
accumulated.addAll(result.resources());
return accumulated;
}).map(all -> McpSchema.ListResourcesResult.builder(Collections.unmodifiableList(all)).build());
Expand Down Expand Up @@ -907,7 +907,7 @@ public Mono<McpSchema.ListResourceTemplatesResult> listResourceTemplates() {
return this.listResourceTemplates(McpSchema.FIRST_PAGE).expand(result -> {
String next = result.nextCursor();
return (next != null && !next.isEmpty()) ? this.listResourceTemplates(next) : Mono.empty();
}).reduce(new ArrayList<McpSchema.ResourceTemplate>(), (accumulated, result) -> {
}).reduceWith(ArrayList<McpSchema.ResourceTemplate>::new, (accumulated, result) -> {
accumulated.addAll(result.resourceTemplates());
return accumulated;
}).map(all -> McpSchema.ListResourceTemplatesResult.builder(Collections.unmodifiableList(all)).build());
Expand Down Expand Up @@ -1026,7 +1026,7 @@ public Mono<ListPromptsResult> listPrompts() {
return this.listPrompts(McpSchema.FIRST_PAGE).expand(result -> {
String next = result.nextCursor();
return (next != null && !next.isEmpty()) ? this.listPrompts(next) : Mono.empty();
}).reduce(new ArrayList<McpSchema.Prompt>(), (accumulated, result) -> {
}).reduceWith(ArrayList<McpSchema.Prompt>::new, (accumulated, result) -> {
accumulated.addAll(result.prompts());
return accumulated;
}).map(all -> McpSchema.ListPromptsResult.builder(Collections.unmodifiableList(all)).build());
Expand Down
2 changes: 1 addition & 1 deletion mcp-core/src/main/java/io/modelcontextprotocol/server/McpAsyncServerExchange.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,7 +209,7 @@ public Mono<McpSchema.ListRootsResult> listRoots() {
return this.listRoots(McpSchema.FIRST_PAGE)
.expand(result -> (result.nextCursor() != null) ?
this.listRoots(result.nextCursor()) : Mono.empty())
.reduce(McpSchema.ListRootsResult.builder(new ArrayList<>()).build(),
.reduceWith(() -> McpSchema.ListRootsResult.builder(new ArrayList<>()).build(),
(allRootsResult, result) -> {
allRootsResult.roots().addAll(result.roots());
return allRootsResult;
Expand Down
21 changes: 21 additions & 0 deletions mcp-core/src/test/java/io/modelcontextprotocol/server/McpAsyncServerExchangeTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,27 @@ void testListRootsWithSinglePage() {
}).verifyComplete();
}

@Test
void testListRootsCanBeResubscribed() {

List<McpSchema.Root> roots = Arrays
.asList(McpSchema.Root.builder("file:///home/user/project1").name("Project 1").build());
McpSchema.ListRootsResult singlePageResult = McpSchema.ListRootsResult.builder(roots).build();

when(mockSession.sendRequest(eq(McpSchema.METHOD_ROOTS_LIST), any(McpSchema.PaginatedRequest.class),
any(TypeRef.class)))
.thenReturn(Mono.just(singlePageResult));

Mono<McpSchema.ListRootsResult> listRoots = exchange.listRoots();

// The same Mono must be safe to subscribe more than once: the accumulator used to
// aggregate the pages has to be created per subscription, not once when
// listRoots()
// is called.
StepVerifier.create(listRoots).assertNext(result -> assertThat(result.roots()).hasSize(1)).verifyComplete();
StepVerifier.create(listRoots).assertNext(result -> assertThat(result.roots()).hasSize(1)).verifyComplete();
}

@Test
void testListRootsWithMultiplePages() {

Expand Down
80 changes: 80 additions & 0 deletions mcp-test/src/main/java/io/modelcontextprotocol/client/AbstractMcpAsyncClientTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,6 +206,26 @@ void testListAllTools() {
});
}

@Test
void testListAllToolsCanBeResubscribed() {
withClient(createMcpTransport(), mcpAsyncClient -> {
Mono<McpSchema.ListToolsResult> listTools = mcpAsyncClient.listTools();

AtomicInteger firstSubscriptionSize = new AtomicInteger();
StepVerifier.create(mcpAsyncClient.initialize().then(listTools)).consumeNextWith(result -> {
firstSubscriptionSize.set(result.tools().size());
assertThat(result.tools()).isNotEmpty();
}).verifyComplete();

// The same Mono must be safe to subscribe again: every subscription has to
// start from an empty accumulator instead of appending to the previous
// result.
StepVerifier.create(listTools)
.consumeNextWith(result -> assertThat(result.tools()).hasSize(firstSubscriptionSize.get()))
.verifyComplete();
});
}

@Test
void testListAllToolsReturnsImmutableList() {
withClient(createMcpTransport(), mcpAsyncClient -> {
Expand Down Expand Up @@ -378,6 +398,26 @@ void testListAllResources() {
});
}

@Test
void testListAllResourcesCanBeResubscribed() {
withClient(createMcpTransport(), mcpAsyncClient -> {
Mono<McpSchema.ListResourcesResult> listResources = mcpAsyncClient.listResources();

AtomicInteger firstSubscriptionSize = new AtomicInteger();
StepVerifier.create(mcpAsyncClient.initialize().then(listResources)).consumeNextWith(result -> {
firstSubscriptionSize.set(result.resources().size());
assertThat(result.resources()).isNotEmpty();
}).verifyComplete();

// The same Mono must be safe to subscribe again: every subscription has to
// start from an empty accumulator instead of appending to the previous
// result.
StepVerifier.create(listResources)
.consumeNextWith(result -> assertThat(result.resources()).hasSize(firstSubscriptionSize.get()))
.verifyComplete();
});
}

@Test
void testListAllResourcesReturnsImmutableList() {
withClient(createMcpTransport(), mcpAsyncClient -> {
Expand Down Expand Up @@ -443,6 +483,26 @@ void testListAllPrompts() {
});
}

@Test
void testListAllPromptsCanBeResubscribed() {
withClient(createMcpTransport(), mcpAsyncClient -> {
Mono<McpSchema.ListPromptsResult> listPrompts = mcpAsyncClient.listPrompts();

AtomicInteger firstSubscriptionSize = new AtomicInteger();
StepVerifier.create(mcpAsyncClient.initialize().then(listPrompts)).consumeNextWith(result -> {
firstSubscriptionSize.set(result.prompts().size());
assertThat(result.prompts()).isNotEmpty();
}).verifyComplete();

// The same Mono must be safe to subscribe again: every subscription has to
// start from an empty accumulator instead of appending to the previous
// result.
StepVerifier.create(listPrompts)
.consumeNextWith(result -> assertThat(result.prompts()).hasSize(firstSubscriptionSize.get()))
.verifyComplete();
});
}

@Test
void testListAllPromptsReturnsImmutableList() {
withClient(createMcpTransport(), mcpAsyncClient -> {
Expand Down Expand Up @@ -636,6 +696,26 @@ void testListAllResourceTemplates() {
});
}

@Test
void testListAllResourceTemplatesCanBeResubscribed() {
withClient(createMcpTransport(), mcpAsyncClient -> {
Mono<McpSchema.ListResourceTemplatesResult> listResourceTemplates = mcpAsyncClient.listResourceTemplates();

AtomicInteger firstSubscriptionSize = new AtomicInteger();
StepVerifier.create(mcpAsyncClient.initialize().then(listResourceTemplates)).consumeNextWith(result -> {
firstSubscriptionSize.set(result.resourceTemplates().size());
assertThat(result.resourceTemplates()).isNotEmpty();
}).verifyComplete();

// The same Mono must be safe to subscribe again: every subscription has to
// start from an empty accumulator instead of appending to the previous
// result.
StepVerifier.create(listResourceTemplates)
.consumeNextWith(result -> assertThat(result.resourceTemplates()).hasSize(firstSubscriptionSize.get()))
.verifyComplete();
});
}

@Test
void testListAllResourceTemplatesReturnsImmutableList() {
withClient(createMcpTransport(), mcpAsyncClient -> {
Expand Down
Loading