1010import java .util .ArrayList ;
1111import java .util .Collections ;
1212import java .util .HashMap ;
13+ import java .util .HashSet ;
1314import java .util .List ;
1415import java .util .Map ;
16+ import java .util .Set ;
1517import java .util .concurrent .ConcurrentHashMap ;
18+ import java .util .function .BiFunction ;
1619import java .util .function .Function ;
20+ import java .util .function .Supplier ;
1721
1822import io .modelcontextprotocol .client .LifecycleInitializer .Initialization ;
1923import io .modelcontextprotocol .json .TypeRef ;
@@ -185,6 +189,12 @@ public class McpAsyncClient {
185189
186190 private final boolean applyElicitationDefaults ;
187191
192+ /**
193+ * Bounds applied to the no-arg list operations to protect against unbounded
194+ * pagination from misbehaving servers.
195+ */
196+ private final PaginationConfig paginationConfig ;
197+
188198 /**
189199 * Create a new McpAsyncClient with the given transport and session request-response
190200 * timeout.
@@ -196,7 +206,8 @@ public class McpAsyncClient {
196206 * schemas.
197207 */
198208 McpAsyncClient (McpClientTransport transport , Duration requestTimeout , Duration initializationTimeout ,
199- JsonSchemaValidator jsonSchemaValidator , McpClientFeatures .Async features ) {
209+ JsonSchemaValidator jsonSchemaValidator , McpClientFeatures .Async features ,
210+ PaginationConfig paginationConfig ) {
200211
201212 Assert .notNull (transport , "Transport must not be null" );
202213 Assert .notNull (requestTimeout , "Request timeout must not be null" );
@@ -210,6 +221,7 @@ public class McpAsyncClient {
210221 this .toolsOutputSchemaCache = new ConcurrentHashMap <>();
211222 this .enableCallToolSchemaCaching = features .enableCallToolSchemaCaching ();
212223 this .applyElicitationDefaults = features .applyElicitationDefaults ();
224+ this .paginationConfig = paginationConfig != null ? paginationConfig : PaginationConfig .DEFAULT ;
213225
214226 // Request Handlers
215227 Map <String , RequestHandler <?>> requestHandlers = new HashMap <>();
@@ -731,13 +743,11 @@ private McpSchema.CallToolResult validateToolResult(String toolName, McpSchema.C
731743 * @return A Mono that emits the list of all tools result
732744 */
733745 public Mono <McpSchema .ListToolsResult > listTools () {
734- return this .listTools (McpSchema .FIRST_PAGE ).expand (result -> {
735- String next = result .nextCursor ();
736- return (next != null && !next .isEmpty ()) ? this .listTools (next ) : Mono .empty ();
737- }).reduce (new ArrayList <McpSchema .Tool >(), (accumulated , result ) -> {
738- accumulated .addAll (result .tools ());
739- return accumulated ;
740- }).map (all -> McpSchema .ListToolsResult .builder (Collections .unmodifiableList (all )).build ());
746+ return paginate (this ::listTools , McpSchema .ListToolsResult ::nextCursor , ArrayList <McpSchema .Tool >::new ,
747+ (all , result ) -> {
748+ all .addAll (result .tools ());
749+ return all ;
750+ }, all -> McpSchema .ListToolsResult .builder (Collections .unmodifiableList (all )).build ());
741751 }
742752
743753 /**
@@ -818,13 +828,11 @@ private NotificationHandler asyncToolsChangeNotificationHandler(
818828 * @see #readResource(McpSchema.Resource)
819829 */
820830 public Mono <McpSchema .ListResourcesResult > listResources () {
821- return this .listResources (McpSchema .FIRST_PAGE ).expand (result -> {
822- String next = result .nextCursor ();
823- return (next != null && !next .isEmpty ()) ? this .listResources (next ) : Mono .empty ();
824- }).reduce (new ArrayList <McpSchema .Resource >(), (accumulated , result ) -> {
825- accumulated .addAll (result .resources ());
826- return accumulated ;
827- }).map (all -> McpSchema .ListResourcesResult .builder (Collections .unmodifiableList (all )).build ());
831+ return paginate (this ::listResources , McpSchema .ListResourcesResult ::nextCursor ,
832+ ArrayList <McpSchema .Resource >::new , (all , result ) -> {
833+ all .addAll (result .resources ());
834+ return all ;
835+ }, all -> McpSchema .ListResourcesResult .builder (Collections .unmodifiableList (all )).build ());
828836 }
829837
830838 /**
@@ -904,13 +912,11 @@ public Mono<McpSchema.ReadResourceResult> readResource(McpSchema.ReadResourceReq
904912 * @see McpSchema.ListResourceTemplatesResult
905913 */
906914 public Mono <McpSchema .ListResourceTemplatesResult > listResourceTemplates () {
907- return this .listResourceTemplates (McpSchema .FIRST_PAGE ).expand (result -> {
908- String next = result .nextCursor ();
909- return (next != null && !next .isEmpty ()) ? this .listResourceTemplates (next ) : Mono .empty ();
910- }).reduce (new ArrayList <McpSchema .ResourceTemplate >(), (accumulated , result ) -> {
911- accumulated .addAll (result .resourceTemplates ());
912- return accumulated ;
913- }).map (all -> McpSchema .ListResourceTemplatesResult .builder (Collections .unmodifiableList (all )).build ());
915+ return paginate (this ::listResourceTemplates , McpSchema .ListResourceTemplatesResult ::nextCursor ,
916+ ArrayList <McpSchema .ResourceTemplate >::new , (all , result ) -> {
917+ all .addAll (result .resourceTemplates ());
918+ return all ;
919+ }, all -> McpSchema .ListResourceTemplatesResult .builder (Collections .unmodifiableList (all )).build ());
914920 }
915921
916922 /**
@@ -1023,13 +1029,85 @@ private NotificationHandler asyncResourcesUpdatedNotificationHandler(
10231029 * @see #getPrompt(GetPromptRequest)
10241030 */
10251031 public Mono <ListPromptsResult > listPrompts () {
1026- return this .listPrompts (McpSchema .FIRST_PAGE ).expand (result -> {
1027- String next = result .nextCursor ();
1028- return (next != null && !next .isEmpty ()) ? this .listPrompts (next ) : Mono .empty ();
1029- }).reduce (new ArrayList <McpSchema .Prompt >(), (accumulated , result ) -> {
1030- accumulated .addAll (result .prompts ());
1031- return accumulated ;
1032- }).map (all -> McpSchema .ListPromptsResult .builder (Collections .unmodifiableList (all )).build ());
1032+ return paginate (this ::listPrompts , ListPromptsResult ::nextCursor , ArrayList <McpSchema .Prompt >::new ,
1033+ (all , result ) -> {
1034+ all .addAll (result .prompts ());
1035+ return all ;
1036+ }, all -> McpSchema .ListPromptsResult .builder (Collections .unmodifiableList (all )).build ());
1037+ }
1038+
1039+ /**
1040+ * Fetches every page of a paginated list operation, accumulating the pages into a
1041+ * single result, while enforcing the client's pagination bounds. A server that
1042+ * returns an endless stream of non-empty cursors is stopped with an
1043+ * {@link McpPaginationException} once the configured page limit, cursor-repetition
1044+ * guard or total timeout is hit.
1045+ * @param pageFetcher fetches a single page for a given cursor
1046+ * @param nextCursorOf extracts the next cursor from a page result
1047+ * @param initialAccumulator supplies the accumulator for the aggregated result
1048+ * @param accumulate merges one page into the accumulator
1049+ * @param finalize converts the accumulated pages into the final result
1050+ * @param <R> the page/result type
1051+ * @param <A> the accumulator type
1052+ * @return a Mono that emits the aggregated result of all pages
1053+ */
1054+ private <R , A > Mono <R > paginate (Function <String , Mono <R >> pageFetcher , Function <R , String > nextCursorOf ,
1055+ Supplier <A > initialAccumulator , BiFunction <A , R , A > accumulate , Function <A , R > finalize ) {
1056+ return Mono .defer (() -> {
1057+ PaginationGuard guard = new PaginationGuard (this .paginationConfig );
1058+ return pageFetcher .apply (McpSchema .FIRST_PAGE ).expand (page -> {
1059+ String next = nextCursorOf .apply (page );
1060+ if (next == null || next .isEmpty ()) {
1061+ return Mono .empty ();
1062+ }
1063+ guard .beforeNextPage (next );
1064+ return pageFetcher .apply (next );
1065+ }).reduce (initialAccumulator .get (), accumulate ).map (finalize );
1066+ });
1067+ }
1068+
1069+ /**
1070+ * Tracks pagination state for a single list operation and enforces the configured
1071+ * bounds. Fresh state is created per subscription so that a shared {@link Mono} can
1072+ * be subscribed multiple times without carrying stale guards.
1073+ */
1074+ private static final class PaginationGuard {
1075+
1076+ private final Set <String > visitedCursors = new HashSet <>();
1077+
1078+ private final PaginationConfig config ;
1079+
1080+ private final long startNanos = System .nanoTime ();
1081+
1082+ private int pagesFetched = 1 ;
1083+
1084+ PaginationGuard (PaginationConfig config ) {
1085+ this .config = config ;
1086+ }
1087+
1088+ /**
1089+ * Validates that the next page may be fetched, throwing an
1090+ * {@link McpPaginationException} when a bound is exceeded.
1091+ * @param cursor the next cursor the server asked the client to follow
1092+ */
1093+ void beforeNextPage (String cursor ) {
1094+ if (!this .visitedCursors .add (cursor )) {
1095+ throw new McpPaginationException ("Pagination loop detected: the server returned cursor '" + cursor
1096+ + "' more than once. Aborting the list operation to avoid an endless request loop." );
1097+ }
1098+ if (this .config .maxPages () > 0 && this .pagesFetched >= this .config .maxPages ()) {
1099+ throw new McpPaginationException (
1100+ "Pagination limit exceeded: the server returned more than " + this .config .maxPages ()
1101+ + " pages. Increase maxPaginationPages if this is expected for the server." );
1102+ }
1103+ if (this .config .timeout () != null
1104+ && Duration .ofNanos (System .nanoTime () - this .startNanos ).compareTo (this .config .timeout ()) > 0 ) {
1105+ throw new McpPaginationException ("Pagination timed out after " + this .config .timeout ()
1106+ + ". Increase paginationTimeout if this is expected for the server." );
1107+ }
1108+ this .pagesFetched ++;
1109+ }
1110+
10331111 }
10341112
10351113 /**
0 commit comments