feat(client): Set a limit on the size of the API responses passed back to the MCP client to control LLM context bloat

Issue: APPAI-176

Author: timothedelion

packages/gg_api_core/src/gg_api_core/client.py

@@ -75,6 +75,21 @@ class ListResponse(TypedDict):
     has_more: bool
 
 
+# Default limit for paginate_all to prevent context bloat
+DEFAULT_PAGINATION_MAX_BYTES = 20_000
+
+# Default HTTP timeout in seconds (to handle slow pagination)
+DEFAULT_HTTP_TIMEOUT = 20
+
+
+class PaginatedResult(TypedDict):
+    """Result from paginate_all with size limit protection."""
+
+    data: list[dict[str, Any]]
+    cursor: str | None  # Use this cursor to fetch more results if has_more is True
+    has_more: bool  # True if results were capped by size limit OR more pages exist
+
+
 def is_oauth_enabled() -> bool:
     """
     Check if OAuth authentication is enabled via environment variable.
@@ -432,7 +447,7 @@ async def _request(self, method: str, endpoint: str, **kwargs) -> Any:
 
         while retry_count <= max_retries:
             try:
-                async with httpx.AsyncClient(follow_redirects=True) as client:
+                async with httpx.AsyncClient(follow_redirects=True, timeout=DEFAULT_HTTP_TIMEOUT) as client:
                     logger.debug(f"Sending {method} request to {url}")
                     response = await client.request(method, url, headers=headers, **kwargs)
 
@@ -652,7 +667,7 @@ async def _request_list(self, endpoint: str, **kwargs) -> ListResponse:
         }
         headers.update(kwargs.pop("headers", {}))
 
-        async with httpx.AsyncClient(follow_redirects=True) as client:
+        async with httpx.AsyncClient(follow_redirects=True, timeout=DEFAULT_HTTP_TIMEOUT) as client:
             response = await client.get(url, headers=headers, **kwargs)
             response.raise_for_status()
 
@@ -677,19 +692,31 @@ async def _request_list(self, endpoint: str, **kwargs) -> ListResponse:
             "has_more": cursor is not None,
         }
 
-    async def paginate_all(self, endpoint: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]:
+    async def paginate_all(
+        self,
+        endpoint: str,
+        params: dict[str, Any] | None = None,
+        max_bytes: int = DEFAULT_PAGINATION_MAX_BYTES,
+    ) -> PaginatedResult:
         """Fetch all pages of results using cursor-based pagination.
 
+        Pagination stops when either all data is fetched or the size limit is reached.
+
         Args:
             endpoint: API endpoint path
             params: Query parameters to include in the request
+            max_bytes: Maximum total bytes of JSON data to accumulate (default: 20KB).
+                      When this limit is reached, pagination stops and truncated=True is returned.
 
         Returns:
-            List of all items from all pages
+            PaginatedResult with data, cursor, has_more, truncated info, and total_bytes
         """
         params = params or {}
-        all_items = []
-        cursor = None
+        all_items: list[dict[str, Any]] = []
+        total_bytes = 0
+        cursor: str | None = None
+        truncated = False
+        has_more = False
 
         logger.debug(f"Starting pagination for endpoint '{endpoint}' with initial params: {params}")
 
@@ -724,9 +751,26 @@ async def paginate_all(self, endpoint: str, params: dict[str, Any] | None = None
                 logger.debug("Received empty response data, stopping pagination")
                 break
 
-            logger.debug(f"Received page with {len(response['data'])} items")
+            # Calculate size of this page's data
+            page_bytes = len(json.dumps(response["data"]).encode("utf-8"))
+
+            # Check if adding this page would exceed the limit
+            if total_bytes + page_bytes > max_bytes and all_items:
+                # We already have some data, stop here to avoid exceeding limit
+                logger.warning(
+                    f"Pagination stopped due to size limit: {total_bytes} bytes accumulated, "
+                    f"next page would add {page_bytes} bytes (limit: {max_bytes} bytes)"
+                )
+                truncated = True
+                has_more = True
+                # Keep the cursor so caller can continue if needed
+                cursor = response["cursor"]
+                break
+
+            logger.debug(f"Received page with {len(response['data'])} items ({page_bytes} bytes)")
             all_items.extend(response["data"])
-            logger.debug(f"Total items collected so far: {len(all_items)}")
+            total_bytes += page_bytes
+            logger.debug(f"Total items collected so far: {len(all_items)} ({total_bytes} bytes)")
 
             # Check for next cursor
             cursor = response["cursor"]
@@ -736,8 +780,15 @@ async def paginate_all(self, endpoint: str, params: dict[str, Any] | None = None
                 logger.debug("No next cursor found, pagination complete")
                 break
 
-        logger.info(f"Pagination complete for {endpoint}: collected {len(all_items)} total items")
-        return all_items
+        logger.info(
+            f"Pagination complete for {endpoint}: collected {len(all_items)} items "
+            f"({total_bytes} bytes, capped={truncated})"
+        )
+        return {
+            "data": all_items,
+            "cursor": cursor,
+            "has_more": has_more or (cursor is not None),
+        }
 
     async def create_honeytoken(
         self, name: str, description: str = "", custom_tags: list | None = None
@@ -908,9 +959,8 @@ async def list_incidents(
             endpoint = "/incidents/secrets"
 
         if get_all:
-            # When get_all=True, return all items without cursor
-            all_items = await self.paginate_all(endpoint, params)
-            return {"data": all_items, "cursor": None, "has_more": False}
+            # When get_all=True, return paginated result with truncation metadata
+            return await self.paginate_all(endpoint, params)
 
         query_string = "&".join([f"{k}={v}" for k, v in params.items()])
         if query_string:
@@ -1043,9 +1093,8 @@ async def list_honeytokens(
         endpoint = "/honeytokens"
 
         if get_all:
-            # When get_all=True, return all items without cursor
-            all_items = await self.paginate_all(endpoint, params)
-            return {"data": all_items, "cursor": None, "has_more": False}
+            # When get_all=True, return paginated result with truncation metadata
+            return await self.paginate_all(endpoint, params)
 
         query_string = "&".join([f"{k}={v}" for k, v in params.items()])
         if query_string:
@@ -1499,11 +1548,10 @@ async def list_occurrences(
         if with_sources is not None:
             params["with_sources"] = str(with_sources).lower()
 
-        # If get_all is True, use paginate_all to get all results
+        # If get_all is True, use paginate_all to get all results with truncation metadata
         if get_all:
             logger.info("Getting all occurrences using cursor-based pagination")
-            all_items = await self.paginate_all("occurrences/secrets", params)
-            return {"data": all_items, "cursor": None, "has_more": False}
+            return await self.paginate_all("occurrences/secrets", params)
 
         # Otherwise, get a single page
         logger.info(f"Getting occurrences with params: {params}")
@@ -1613,9 +1661,8 @@ async def list_sources(
         endpoint = "/sources"
 
         if get_all:
-            # When get_all=True, return all items without cursor
-            all_items = await self.paginate_all(endpoint, params)
-            return {"data": all_items, "cursor": None, "has_more": False}
+            # When get_all=True, return paginated result with truncation metadata
+            return await self.paginate_all(endpoint, params)
 
         return await self._request_list(endpoint, params=params)
 

packages/gg_api_core/src/gg_api_core/tools/list_honeytokens.py

@@ -4,6 +4,7 @@
 from fastmcp.exceptions import ToolError
 from pydantic import BaseModel, Field
 
+from gg_api_core.client import DEFAULT_PAGINATION_MAX_BYTES
 from gg_api_core.utils import get_client
 
 logger = logging.getLogger(__name__)
@@ -26,7 +27,10 @@ class ListHoneytokensParams(BaseModel):
     creator_api_token_id: str | int | None = Field(default=None, description="Filter by creator API token ID")
     per_page: int = Field(default=20, description="Number of results per page (default: 20, min: 1, max: 100)")
     cursor: str | None = Field(default=None, description="Pagination cursor from a previous response")
-    get_all: bool = Field(default=False, description="If True, fetch all results using cursor-based pagination")
+    get_all: bool = Field(
+        default=False,
+        description=f"If True, fetch all pages (capped at ~{DEFAULT_PAGINATION_MAX_BYTES / 1000}KB; check 'has_more' and use cursor to continue)",
+    )
 
 
 class ListHoneytokensResult(BaseModel):
@@ -36,6 +40,7 @@ class ListHoneytokensResult(BaseModel):
     next_cursor: str | None = Field(
         default=None, description="Cursor for fetching the next page (null if no more results)"
     )
+    has_more: bool = Field(default=False, description="True if more results exist (use next_cursor to fetch)")
 
 
 async def list_honeytokens(params: ListHoneytokensParams) -> ListHoneytokensResult:
@@ -90,7 +95,11 @@ async def list_honeytokens(params: ListHoneytokensParams) -> ListHoneytokensResu
         next_cursor = response["cursor"]
 
         logger.debug(f"Found {len(honeytokens_data)} honeytokens")
-        return ListHoneytokensResult(honeytokens=honeytokens_data, next_cursor=next_cursor)
+        return ListHoneytokensResult(
+            honeytokens=honeytokens_data,
+            next_cursor=next_cursor,
+            has_more=response.get("has_more", False),
+        )
     except Exception as e:
         logger.error(f"Error listing honeytokens: {str(e)}")
         raise ToolError(str(e))

packages/gg_api_core/src/gg_api_core/tools/list_incidents.py

@@ -3,7 +3,13 @@
 
 from pydantic import BaseModel, Field
 
-from gg_api_core.client import IncidentSeverity, IncidentStatus, IncidentValidity, TagNames
+from gg_api_core.client import (
+    DEFAULT_PAGINATION_MAX_BYTES,
+    IncidentSeverity,
+    IncidentStatus,
+    IncidentValidity,
+    TagNames,
+)
 from gg_api_core.tools.find_current_source_id import find_current_source_id
 from gg_api_core.utils import get_client
 
@@ -115,7 +121,10 @@ class ListIncidentsParams(BaseModel):
     ordering: str | None = Field(default=None, description="Sort field (e.g., 'date', '-date' for descending)")
     per_page: int = Field(default=20, description="Number of results per page (default: 20, min: 1, max: 100)")
     cursor: str | None = Field(default=None, description="Pagination cursor for fetching next page of results")
-    get_all: bool = Field(default=False, description="If True, fetch all results using cursor-based pagination")
+    get_all: bool = Field(
+        default=False,
+        description=f"If True, fetch all pages (capped at ~{DEFAULT_PAGINATION_MAX_BYTES / 1000}; check 'has_more' and use cursor to continue)",
+    )
 
     # Filters
     from_date: str | None = Field(
@@ -164,6 +173,7 @@ class ListIncidentsResult(BaseModel):
     next_cursor: str | None = Field(default=None, description="Pagination cursor for next page")
     applied_filters: dict[str, Any] = Field(default_factory=dict, description="Filters that were applied to the query")
     suggestion: str = Field(default="", description="Suggestions for interpreting or modifying the results")
+    has_more: bool = Field(default=False, description="True if more results exist (use next_cursor to fetch)")
 
 
 class ListIncidentsError(BaseModel):
@@ -248,7 +258,7 @@ async def list_incidents(params: ListIncidentsParams) -> ListIncidentsResult | L
         if params.get_all:
             api_params["get_all"] = params.get_all
 
-        # Get incidents using list_incidents which returns ListResponse
+        # Get incidents using list_incidents which returns ListResponse or PaginatedResult
         response = await client.list_incidents(**api_params)
         incidents_data = response["data"]
         next_cursor = response["cursor"]
@@ -261,6 +271,7 @@ async def list_incidents(params: ListIncidentsParams) -> ListIncidentsResult | L
             next_cursor=next_cursor,
             applied_filters=_build_filter_info(params),
             suggestion=_build_suggestion(params, count),
+            has_more=response.get("has_more", False),
         )
 
     except Exception as e:

packages/gg_api_core/src/gg_api_core/tools/list_repo_occurrences.py

@@ -3,7 +3,13 @@
 
 from pydantic import BaseModel, Field
 
-from gg_api_core.client import IncidentSeverity, IncidentStatus, IncidentValidity, TagNames
+from gg_api_core.client import (
+    DEFAULT_PAGINATION_MAX_BYTES,
+    IncidentSeverity,
+    IncidentStatus,
+    IncidentValidity,
+    TagNames,
+)
 from gg_api_core.utils import get_client
 
 logger = logging.getLogger(__name__)
@@ -74,7 +80,10 @@ class ListRepoOccurrencesBaseParams(BaseModel):
     ordering: str | None = Field(default=None, description="Sort field (e.g., 'date', '-date' for descending)")
     per_page: int = Field(default=20, description="Number of results per page (default: 20, min: 1, max: 100)")
     cursor: str | None = Field(default=None, description="Pagination cursor for fetching next page of results")
-    get_all: bool = Field(default=False, description="If True, fetch all results using cursor-based pagination")
+    get_all: bool = Field(
+        default=False,
+        description=f"If True, fetch all pages (capped at ~{DEFAULT_PAGINATION_MAX_BYTES / 1000}KB; check 'has_more' and use cursor to continue)",
+    )
 
 
 class ListRepoOccurrencesParams(ListRepoOccurrencesFilters, ListRepoOccurrencesBaseParams):
@@ -88,7 +97,7 @@ class ListRepoOccurrencesResult(BaseModel):
     occurrences_count: int = Field(description="Number of occurrences returned")
     occurrences: list[dict[str, Any]] = Field(default_factory=list, description="List of occurrence objects")
     cursor: str | None = Field(default=None, description="Pagination cursor for next page")
-    has_more: bool = Field(default=False, description="Whether more results are available")
+    has_more: bool = Field(default=False, description="True if more results exist (use cursor to fetch)")
     applied_filters: dict[str, Any] = Field(default_factory=dict, description="Filters that were applied to the query")
     suggestion: str = Field(default="", description="Suggestions for interpreting or modifying the results")
 
@@ -237,10 +246,10 @@ async def list_repo_occurrences(
                 with_sources=False,
             )
 
-        # Extract data from ListResponse
+        # Extract data from ListResponse or PaginatedResult
         occurrences_data = result["data"]
         next_cursor = result["cursor"]
-        has_more = result["has_more"]
+        has_more = result.get("has_more", False)
 
         count = len(occurrences_data)
         return ListRepoOccurrencesResult(

packages/gg_api_core/src/gg_api_core/tools/list_users.py

@@ -4,6 +4,7 @@
 
 from pydantic import BaseModel, Field
 
+from gg_api_core.client import DEFAULT_PAGINATION_MAX_BYTES
 from gg_api_core.utils import get_client
 
 logger = logging.getLogger(__name__)
@@ -27,7 +28,10 @@ class ListUsersParams(BaseModel):
         default=None,
         description="Sort results by field (created_at, -created_at, last_login, -last_login). Use '-' prefix for descending order",
     )
-    get_all: bool = Field(default=False, description="If True, fetch all results using cursor-based pagination")
+    get_all: bool = Field(
+        default=False,
+        description=f"If True, fetch all pages (capped at ~{DEFAULT_PAGINATION_MAX_BYTES / 1000}KB; check 'has_more' and use cursor to continue)",
+    )
 
 
 class ListUsersResult(BaseModel):
@@ -36,6 +40,7 @@ class ListUsersResult(BaseModel):
     members: list[dict[str, Any]] = Field(description="List of workspace member objects")
     total_count: int = Field(description="Total number of members returned")
     next_cursor: str | None = Field(default=None, description="Pagination cursor for next page (if applicable)")
+    has_more: bool = Field(default=False, description="True if more results exist (use next_cursor to fetch)")
 
 
 async def list_users(params: ListUsersParams) -> ListUsersResult:
@@ -81,10 +86,15 @@ async def list_users(params: ListUsersParams) -> ListUsersResult:
     logger.debug(f"Query parameters: {json.dumps(query_params)}")
 
     if params.get_all:
-        # Use paginate_all for fetching all results
-        members_list = await client.paginate_all("/members", query_params)
-        logger.debug(f"Retrieved all {len(members_list)} members using pagination")
-        return ListUsersResult(members=members_list, total_count=len(members_list), next_cursor=None)
+        # Use paginate_all for fetching all results (capped at DEFAULT_PAGINATION_MAX_BYTES)
+        result = await client.paginate_all("/members", query_params)
+        logger.debug(f"Retrieved {len(result['data'])} members using pagination (has_more={result['has_more']})")
+        return ListUsersResult(
+            members=result["data"],
+            total_count=len(result["data"]),
+            next_cursor=result["cursor"],
+            has_more=result["has_more"],
+        )
     else:
         # Single page request
         result = await client.list_members(params=query_params)

packages/secops_mcp_server/src/secops_mcp_server/server.py

@@ -6,6 +6,7 @@
 from developer_mcp_server.add_health_check import add_health_check
 from developer_mcp_server.register_tools import register_developer_tools
 from fastmcp.exceptions import ToolError
+from gg_api_core.client import DEFAULT_PAGINATION_MAX_BYTES
 from gg_api_core.mcp_server import get_mcp_server, register_common_tools
 from gg_api_core.scopes import set_secops_scopes
 from gg_api_core.tools.assign_incident import assign_incident
@@ -56,7 +57,10 @@ class ListIncidentsParams(BaseModel):
         description="Sort field and direction (prefix with '-' for descending order). If you need to get the latest incidents, use '-date'.",
     )
     per_page: int = Field(default=20, description="Number of results per page (1-100)")
-    get_all: bool = Field(default=False, description="If True, fetch all results using cursor-based pagination")
+    get_all: bool = Field(
+        default=False,
+        description=f"If True, fetch all pages (capped at ~{DEFAULT_PAGINATION_MAX_BYTES / 1000}KB; check 'has_more' and use cursor to continue)",
+    )
     mine: bool = Field(default=False, description="If True, fetch incidents assigned to the current user")
 
 
@@ -72,7 +76,10 @@ class ListHoneytokensParams(BaseModel):
     creator_id: str | int | None = Field(default=None, description="Filter by creator ID")
     creator_api_token_id: str | int | None = Field(default=None, description="Filter by creator API token ID")
     per_page: int = Field(default=20, description="Number of results per page (default: 20, min: 1, max: 100)")
-    get_all: bool = Field(default=False, description="If True, fetch all results using cursor-based pagination")
+    get_all: bool = Field(
+        default=False,
+        description=f"If True, fetch all pages (capped at ~{DEFAULT_PAGINATION_MAX_BYTES / 1000}KB; check 'has_more' and use cursor to continue)",
+    )
     mine: bool = Field(default=False, description="If True, fetch honeytokens created by the current user")