wehub-resource-sync
49b9bb6724
Deploy Docs / deploy-docs (push) Failing after 1s
Conformance Tests / client-conformance (push) Failing after 3s
Conformance Tests / server-conformance (push) Failing after 1s
GitHub Actions Security Analysis / zizmor (push) Failing after 1s
CI / checks (push) Failing after 59m20s
CI / all-green (push) Waiting to run
41 lines
1.2 KiB
Python
41 lines
1.2 KiB
Python
"""Example of consuming paginated MCP endpoints from a client."""
|
|
|
|
import asyncio
|
|
|
|
from mcp_types import PaginatedRequestParams, Resource
|
|
|
|
from mcp.client.session import ClientSession
|
|
from mcp.client.stdio import StdioServerParameters, stdio_client
|
|
|
|
|
|
async def list_all_resources() -> None:
|
|
"""Fetch all resources using pagination."""
|
|
async with stdio_client(StdioServerParameters(command="uv", args=["run", "mcp-simple-pagination"])) as (
|
|
read,
|
|
write,
|
|
):
|
|
async with ClientSession(read, write) as session:
|
|
await session.initialize()
|
|
|
|
all_resources: list[Resource] = []
|
|
cursor = None
|
|
|
|
while True:
|
|
# Fetch a page of resources
|
|
result = await session.list_resources(params=PaginatedRequestParams(cursor=cursor))
|
|
all_resources.extend(result.resources)
|
|
|
|
print(f"Fetched {len(result.resources)} resources")
|
|
|
|
# Check if there are more pages
|
|
if result.next_cursor:
|
|
cursor = result.next_cursor
|
|
else:
|
|
break
|
|
|
|
print(f"Total resources: {len(all_resources)}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(list_all_resources())
|