feat: add pagination support for large card execution results

- Add _estimate_tokens helper function for rough token counting
- Enhance execute_card to detect large results and suggest pagination
- Add execute_card_paginated tool with server-side and client-side pagination fallback
- Implement token-based result trimming to stay within context limits
- Support configurable page sizes and token limits

Author: kmlx

src/server.py

@@ -5,6 +5,7 @@
 execute queries, manage cards, and work with collections.
 """
 
+import json
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
@@ -86,6 +87,18 @@ async def request(self, method: str, path: str, **kwargs: Any) -> Any:
                 raise Exception(error_message)
 
 
+def _estimate_tokens(obj: Any) -> int:
+    """
+    Very rough token estimate: ~4 chars per token.
+    (Good enough to avoid giant pages; tune as needed.)
+    """
+    try:
+        s = json.dumps(obj, ensure_ascii=False, separators=(",", ":"))
+    except Exception:
+        s = str(obj)
+    return max(1, len(s) // 4)
+
+
 @dataclass
 class AppContext:
     """Type-safe application context for FastMCP lifespan."""
@@ -442,8 +455,10 @@ async def list_cards_by_collection(ctx: Context, collection_id: int) -> Any:
 
 
 @mcp.tool()
-async def execute_card(ctx: Context, card_id: int, parameters: dict[str, Any] | None = None) -> Any:
-    """Execute a Metabase question/card and get results"""
+async def execute_card(
+    ctx: Context, card_id: int, parameters: dict[str, Any] | None = None
+) -> dict[str, Any]:
+    """Execute a Metabase question/card and get results. For large results, use execute_card_paginated instead."""
     try:
         # Access type-safe lifespan context
         metabase_client = ctx.request_context.lifespan_context.metabase_client
@@ -453,12 +468,143 @@ async def execute_card(ctx: Context, card_id: int, parameters: dict[str, Any] |
             payload["parameters"] = parameters
 
         result = await metabase_client.request("POST", f"/card/{card_id}/query", json=payload)
+
+        # Check if result is too large (>20k tokens)
+        if _estimate_tokens(result) > 20_000:
+            return {
+                "error": "Result too large for single response",
+                "message": "This card returns a large dataset that exceeds the 20,000 token limit. Please use the 'execute_card_paginated' tool instead to get results in manageable chunks. For futher runs extract the sql from paginated results and use it directly",
+                "estimated_tokens": _estimate_tokens(result),
+                "suggested_action": f"Use execute_card_paginated({card_id}) to get paginated results",
+                "card_id": card_id,
+            }
+
         return result
     except Exception as e:
         logger.error(f"Error executing card {card_id}: {e}")
         raise
 
 
+@mcp.tool()
+async def execute_card_paginated(
+    ctx: Context,
+    card_id: int,
+    parameters: dict[str, Any] | None = None,
+    page: int = 0,
+    *,
+    server_page_size_rows: int = 5_000,
+    client_page_size_rows: int = 2_000,
+    max_tokens_per_page: int = 20_000,
+) -> dict[str, Any]:
+    """
+    Execute a Metabase question/card and return a specific page of results.
+
+    Args:
+        card_id: The ID of the card to execute
+        parameters: Parameters to pass to the card
+        page: Page number to retrieve (0-based)
+        server_page_size_rows: Rows per page for server-side pagination
+        client_page_size_rows: Rows per page for client-side chunking
+        max_tokens_per_page: Maximum tokens per page to avoid context limits
+
+    Returns:
+        A dictionary containing the page data and pagination info
+    """
+    try:
+        # Access type-safe lifespan context
+        metabase_client = ctx.request_context.lifespan_context.metabase_client
+
+        base_params = dict(parameters or {})
+
+        # Try server-side pagination first
+        payload = {"parameters": dict(base_params)}
+        payload["parameters"]["limit"] = server_page_size_rows
+        payload["parameters"]["offset"] = page * server_page_size_rows
+
+        try:
+            result = await metabase_client.request("POST", f"/card/{card_id}/query", json=payload)
+            data = result.get("data", {})
+            rows = data.get("rows", []) or data.get("results", [])
+
+            # If server-side pagination worked (got reasonable number of rows)
+            if len(rows) <= server_page_size_rows:
+                page_result = dict(result)
+                if rows and _estimate_tokens(page_result) > max_tokens_per_page:
+                    # Trim rows to fit token budget
+                    trimmed_rows = []
+                    for row in rows:
+                        test_page = dict(result)
+                        test_page["data"] = dict(data, rows=trimmed_rows + [row])
+                        if _estimate_tokens(test_page) > max_tokens_per_page:
+                            break
+                        trimmed_rows.append(row)
+                    page_result["data"] = dict(data, rows=trimmed_rows)
+
+                has_more = len(rows) == server_page_size_rows
+                page_result["pagination"] = {
+                    "page": page,
+                    "page_size": len(rows),
+                    "has_more": has_more,
+                    "pagination_type": "server_side",
+                }
+                return page_result
+
+        except Exception:
+            # Server-side pagination failed, fall back to client-side
+            pass
+
+        # Fall back to client-side pagination
+        payload = {}
+        if parameters:
+            payload["parameters"] = parameters
+
+        full_result = await metabase_client.request("POST", f"/card/{card_id}/query", json=payload)
+        data = full_result.get("data", {})
+        all_rows = data.get("rows", []) or data.get("results", [])
+
+        # Calculate pagination bounds
+        start_idx = page * client_page_size_rows
+        end_idx = start_idx + client_page_size_rows
+        page_rows = all_rows[start_idx:end_idx]
+
+        # Create page result
+        page_result = dict(full_result)
+        page_data = dict(data)
+        page_data["rows"] = page_rows
+        if "cols" in data:
+            page_data["cols"] = data["cols"]
+        page_result["data"] = page_data
+
+        # Trim by token budget if needed
+        if page_rows and _estimate_tokens(page_result) > max_tokens_per_page:
+            trimmed_rows = []
+            for row in page_rows:
+                test_page = dict(page_result)
+                test_page["data"] = dict(page_data, rows=trimmed_rows + [row])
+                if _estimate_tokens(test_page) > max_tokens_per_page:
+                    break
+                trimmed_rows.append(row)
+            page_result["data"] = dict(page_data, rows=trimmed_rows)
+
+        # Add pagination metadata
+        total_rows = len(all_rows)
+        total_pages = (total_rows + client_page_size_rows - 1) // client_page_size_rows
+        page_result["pagination"] = {
+            "page": page,
+            "page_size": len(page_rows),
+            "total_rows": total_rows,
+            "total_pages": total_pages,
+            "has_more": page < total_pages - 1,
+            "pagination_type": "client_side",
+        }
+
+        return page_result
+
+    except Exception as e:
+        logger.error(f"Error executing card {card_id}: {e}")
+        raise
+
+
 @mcp.tool()
 async def execute_query(
     ctx: Context,