How do I implement API pagination with MCP servers?
API pagination is a critical technique when working with MCP (Model Context Protocol) servers, especially when dealing with large datasets that cannot be retrieved in a single request. MCP servers often expose paginated endpoints to efficiently handle data retrieval, and understanding how to implement pagination correctly ensures optimal performance and complete data collection.
Understanding API Pagination in MCP Context
The Model Context Protocol allows AI assistants to interact with external data sources through standardized server implementations. When MCP servers expose APIs that return large datasets—such as search results, database records, or scraped content—pagination becomes essential to manage memory, reduce response times, and comply with rate limits.
There are three primary pagination strategies you'll encounter when working with MCP servers:
- Offset-based pagination - Uses offset and limit parameters
- Cursor-based pagination - Uses cursor tokens for navigation
- Page-based pagination - Uses page numbers and page size
Implementing Offset-Based Pagination
Offset-based pagination is the most straightforward approach, using offset and limit (or skip and take) parameters to control which subset of results to retrieve.
Python Implementation
Here's how to implement offset-based pagination when calling an MCP server API:
import requests
from typing import List, Dict, Any
class MCPPaginatedClient:
def __init__(self, base_url: str, api_key: str):
self.base_url = base_url
self.headers = {"Authorization": f"Bearer {api_key}"}
def fetch_all_records(self, endpoint: str, limit: int = 100) -> List[Dict[str, Any]]:
"""
Fetch all records from a paginated MCP server endpoint.
Args:
endpoint: The API endpoint to query
limit: Number of records per request
Returns:
List of all records from all pages
"""
all_records = []
offset = 0
while True:
params = {
"offset": offset,
"limit": limit
}
response = requests.get(
f"{self.base_url}/{endpoint}",
headers=self.headers,
params=params
)
response.raise_for_status()
data = response.json()
records = data.get("records", [])
if not records:
break
all_records.extend(records)
offset += limit
# Check if we've reached the end
total = data.get("total")
if total and offset >= total:
break
return all_records
# Usage example
client = MCPPaginatedClient(
base_url="https://api.example.com",
api_key="your_api_key_here"
)
all_data = client.fetch_all_records("scraped-data", limit=50)
print(f"Retrieved {len(all_data)} total records")
JavaScript Implementation
Here's the equivalent implementation in JavaScript using async/await:
class MCPPaginatedClient {
constructor(baseUrl, apiKey) {
this.baseUrl = baseUrl;
this.apiKey = apiKey;
}
async fetchAllRecords(endpoint, limit = 100) {
const allRecords = [];
let offset = 0;
while (true) {
const url = new URL(`${this.baseUrl}/${endpoint}`);
url.searchParams.append('offset', offset);
url.searchParams.append('limit', limit);
const response = await fetch(url, {
headers: {
'Authorization': `Bearer ${this.apiKey}`
}
});
if (!response.ok) {
throw new Error(`API error: ${response.status}`);
}
const data = await response.json();
const records = data.records || [];
if (records.length === 0) {
break;
}
allRecords.push(...records);
offset += limit;
// Check if we've reached the end
if (data.total && offset >= data.total) {
break;
}
}
return allRecords;
}
}
// Usage example
const client = new MCPPaginatedClient(
'https://api.example.com',
'your_api_key_here'
);
const allData = await client.fetchAllRecords('scraped-data', 50);
console.log(`Retrieved ${allData.length} total records`);
Implementing Cursor-Based Pagination
Cursor-based pagination is more efficient for large datasets and provides consistent results even when data changes between requests. This approach is commonly used by modern APIs, including many MCP server implementations.
Python Implementation with Cursors
class MCPCursorClient:
def __init__(self, base_url: str, api_key: str):
self.base_url = base_url
self.headers = {"Authorization": f"Bearer {api_key}"}
def fetch_all_with_cursor(self, endpoint: str, page_size: int = 100) -> List[Dict[str, Any]]:
"""
Fetch all records using cursor-based pagination.
Args:
endpoint: The API endpoint to query
page_size: Number of records per page
Returns:
List of all records from all pages
"""
all_records = []
cursor = None
while True:
params = {"page_size": page_size}
if cursor:
params["cursor"] = cursor
response = requests.get(
f"{self.base_url}/{endpoint}",
headers=self.headers,
params=params
)
response.raise_for_status()
data = response.json()
records = data.get("data", [])
if not records:
break
all_records.extend(records)
# Get the next cursor
cursor = data.get("next_cursor")
if not cursor:
break
return all_records
# Usage
client = MCPCursorClient(
base_url="https://api.example.com",
api_key="your_api_key_here"
)
results = client.fetch_all_with_cursor("search-results", page_size=100)
JavaScript Cursor Implementation
class MCPCursorClient {
constructor(baseUrl, apiKey) {
this.baseUrl = baseUrl;
this.apiKey = apiKey;
}
async fetchAllWithCursor(endpoint, pageSize = 100) {
const allRecords = [];
let cursor = null;
while (true) {
const url = new URL(`${this.baseUrl}/${endpoint}`);
url.searchParams.append('page_size', pageSize);
if (cursor) {
url.searchParams.append('cursor', cursor);
}
const response = await fetch(url, {
headers: {
'Authorization': `Bearer ${this.apiKey}`
}
});
if (!response.ok) {
throw new Error(`API error: ${response.status}`);
}
const data = await response.json();
const records = data.data || [];
if (records.length === 0) {
break;
}
allRecords.push(...records);
// Get the next cursor
cursor = data.next_cursor;
if (!cursor) {
break;
}
}
return allRecords;
}
}
Implementing Page-Based Pagination
Page-based pagination uses page numbers and is intuitive for developers familiar with traditional web pagination patterns.
Python Page-Based Implementation
def fetch_paginated_data(base_url: str, endpoint: str, api_key: str, per_page: int = 50):
"""Fetch all pages of data from an MCP server endpoint."""
headers = {"Authorization": f"Bearer {api_key}"}
all_data = []
page = 1
while True:
params = {
"page": page,
"per_page": per_page
}
response = requests.get(
f"{base_url}/{endpoint}",
headers=headers,
params=params
)
response.raise_for_status()
data = response.json()
items = data.get("items", [])
if not items:
break
all_data.extend(items)
# Check if there are more pages
if not data.get("has_next_page", False):
break
page += 1
return all_data
Advanced Pagination Techniques
Rate Limiting and Retry Logic
When working with MCP servers, you should implement rate limiting and retry logic to handle API throttling:
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session_with_retries():
"""Create a requests session with automatic retry logic."""
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
class RateLimitedMCPClient:
def __init__(self, base_url: str, api_key: str, requests_per_second: float = 2):
self.base_url = base_url
self.headers = {"Authorization": f"Bearer {api_key}"}
self.session = create_session_with_retries()
self.min_interval = 1.0 / requests_per_second
self.last_request_time = 0
def _rate_limit(self):
"""Ensure we don't exceed the rate limit."""
elapsed = time.time() - self.last_request_time
if elapsed < self.min_interval:
time.sleep(self.min_interval - elapsed)
self.last_request_time = time.time()
def fetch_page(self, endpoint: str, params: dict) -> dict:
"""Fetch a single page with rate limiting."""
self._rate_limit()
response = self.session.get(
f"{self.base_url}/{endpoint}",
headers=self.headers,
params=params,
timeout=30
)
response.raise_for_status()
return response.json()
Parallel Pagination for Known Page Counts
When you know the total number of pages in advance, you can fetch multiple pages in parallel to improve performance:
import asyncio
import aiohttp
from typing import List
async def fetch_page_async(
session: aiohttp.ClientSession,
url: str,
headers: dict,
page: int,
per_page: int
) -> List[dict]:
"""Fetch a single page asynchronously."""
params = {"page": page, "per_page": per_page}
async with session.get(url, headers=headers, params=params) as response:
response.raise_for_status()
data = await response.json()
return data.get("items", [])
async def fetch_all_pages_parallel(
base_url: str,
endpoint: str,
api_key: str,
total_pages: int,
per_page: int = 50
) -> List[dict]:
"""Fetch all pages in parallel."""
headers = {"Authorization": f"Bearer {api_key}"}
url = f"{base_url}/{endpoint}"
async with aiohttp.ClientSession() as session:
tasks = [
fetch_page_async(session, url, headers, page, per_page)
for page in range(1, total_pages + 1)
]
results = await asyncio.gather(*tasks)
# Flatten the list of lists
all_items = []
for items in results:
all_items.extend(items)
return all_items
# Usage
all_data = asyncio.run(
fetch_all_pages_parallel(
base_url="https://api.example.com",
endpoint="data",
api_key="your_api_key",
total_pages=10
)
)
Handling Browser-Based Pagination with MCP
When using MCP servers that integrate with browser automation tools like Puppeteer or Playwright, you'll need to handle AJAX requests using Puppeteer to properly wait for paginated content to load. This is particularly important when scraping single-page applications that load data dynamically.
For browser-based pagination, you can combine MCP server capabilities with navigation techniques. Learning how to navigate to different pages using Puppeteer will help you implement pagination in scenarios where clicking "Next" buttons or infinite scroll is required.
Best Practices for MCP Server Pagination
Always implement error handling: API requests can fail due to network issues, rate limits, or server errors. Use try-except blocks and implement retry logic.
Respect rate limits: MCP servers often have rate limits. Implement exponential backoff and respect the
Retry-Afterheader when present.Cache pagination tokens: When using cursor-based pagination, store cursors to enable resuming interrupted requests.
Monitor memory usage: When fetching large datasets, consider processing records in batches rather than loading everything into memory.
Log pagination progress: Track which pages have been fetched to enable debugging and resume functionality.
Use appropriate page sizes: Balance between fewer requests (larger pages) and memory usage (smaller pages). Typical values range from 50 to 1000 records per page.
Handle edge cases: Account for empty results, single-page results, and API changes that might affect pagination structure.
Testing Your Pagination Implementation
Here's a simple test script to verify your pagination logic:
def test_pagination(client, endpoint, expected_total):
"""Test that pagination retrieves all expected records."""
records = client.fetch_all_records(endpoint)
assert len(records) == expected_total, \
f"Expected {expected_total} records, got {len(records)}"
# Verify no duplicates
ids = [r['id'] for r in records]
assert len(ids) == len(set(ids)), "Found duplicate records"
print(f"✓ Successfully retrieved all {len(records)} records")
print(f"✓ No duplicates found")
Conclusion
Implementing API pagination with MCP servers requires understanding the pagination strategy used by the server and writing robust code that handles edge cases, rate limits, and errors. Whether you're using offset-based, cursor-based, or page-based pagination, the key principles remain the same: iterate through all pages systematically, handle errors gracefully, and respect API limitations.
By following the patterns and best practices outlined in this guide, you'll be able to efficiently retrieve complete datasets from MCP servers while maintaining good performance and reliability. Remember to always consult the specific MCP server's API documentation for details about its pagination implementation and any special requirements or limitations.
