Source code for aidputils.agents.tools.mcp.pagination_helper

from typing import Any, Dict, Optional

PAGINATION_ENABLED: bool = False



def set_pagination_enabled(enabled: bool) -> None:
    global PAGINATION_ENABLED
    PAGINATION_ENABLED = bool(enabled)




def normalize_paged_result(op: str, raw: Any, page: int, limit: int, paginate: Optional[bool] = None) -> Dict[str, Any]:
    """
    Normalize various possible client responses into a standard pagination envelope.
    preserving semantics to avoid regressions.
    """
    items = []
    total = None
    has_next = None
    # Determine whether to paginate this response. Default is disabled.
    do_paginate = PAGINATION_ENABLED if paginate is None else bool(paginate)

    if isinstance(raw, dict):
        # Try common keys in descending priority
        if isinstance(raw.get("items"), list):
            items = raw.get("items") or []
        elif op == "list_tools" and isinstance(raw.get("tools"), list):
            items = raw.get("tools") or []
        elif op == "list_resources" and isinstance(raw.get("resources"), list):
            items = raw.get("resources") or []
        elif isinstance(raw.get("data"), list):
            items = raw.get("data") or []
        elif isinstance(raw.get("result"), list):
            items = raw.get("result") or []
        else:
            # Unknown structure; attempt to extract any list-like payload
            for _, v in raw.items():
                if isinstance(v, list):
                    items = v
                    break

        # Derive totals/next if provided by server
        total = raw.get("total") or raw.get("totalCount") or raw.get("count")
        if "hasNextPage" in raw:
            has_next = bool(raw["hasNextPage"])
        elif "nextPage" in raw:
            has_next = raw["nextPage"] is not None
        else:
            # If server didn't include pagination info, either return full (no pagination) or emulate by slicing
            total = len(items)
            if do_paginate:
                start = max((page - 1), 0) * limit
                end = start + limit
                items = items[start:end]
                has_next = end < total
            else:
                has_next = False
                page = 0

    elif isinstance(raw, list):
        total = len(raw)
        if do_paginate:
            start = max((page - 1), 0) * limit
            end = start + limit
            items = raw[start:end]
            has_next = end < total
        else:
            items = raw
            has_next = False
    else:
        # Unknown payload
        items = []

    result: Dict[str, Any] = {
        "items": items,
        "page": page,
        "limit": limit,
        "total": len(items),
    }
    if has_next is None and total is not None:
        has_next = (page * limit) < int(total)
    if has_next is not None:
        result["nextPage"] = (page + 1) if has_next else None

    return result