Get All Assistant Session Messages

import os

import requests

BASE_URL = "https://api.caylex.ai/api/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['CAYLEX_PLATFORM_TOKEN']}"}

PROJECT_NAME = "Acme Corp"
TIMESTAMP_START = "2026-06-01T00:00:00Z"
TIMESTAMP_END = "2026-06-30T23:59:59Z"


def get_items(path: str, params: dict | None = None) -> list[dict]:
    """Fetch a cursor-paginated list endpoint (returns all items)."""
    params = dict(params or {})
    params.setdefault("size", 100)
    items: list[dict] = []
    cursor: str | None = None
    while True:
        page_params = {**params}
        if cursor:
            page_params["cursor"] = cursor
        resp = requests.get(f"{BASE_URL}{path}", headers=HEADERS, params=page_params)
        resp.raise_for_status()
        page = resp.json()
        items.extend(page["items"])
        if not page["meta"]["has_next"]:
            break
        cursor = page["meta"]["next_cursor"]
    return items


def project_id_by_name(name: str) -> str | None:
    return next((p["id"] for p in get_items("/projects") if p["name"] == name), None)


def list_sessions_for_project(
    project_id: str,
    *,
    timestamp_start: str,
    timestamp_end: str,
) -> list[dict]:
    """Page through GET /assistants/sessions with limit/offset."""
    sessions: list[dict] = []
    limit = 200
    offset = 0
    while True:
        resp = requests.get(
            f"{BASE_URL}/assistants/sessions",
            headers=HEADERS,
            params={
                "project_id": project_id,
                "timestamp_start": timestamp_start,
                "timestamp_end": timestamp_end,
                "limit": limit,
                "offset": offset,
            },
        )
        resp.raise_for_status()
        page = resp.json()
        sessions.extend(page["sessions"])
        total = page["count"]
        offset += len(page["sessions"])
        if offset >= total or not page["sessions"]:
            break
    return sessions


def get_all_messages(session_id: str, fmt: str = "raw") -> list[dict]:
    """Page through the whole timeline and return every message item."""
    items: list[dict] = []
    cursor: str | None = None
    while True:
        params = {"view": fmt, "size": 100}
        if cursor:
            params["cursor"] = cursor
        resp = requests.get(
            f"{BASE_URL}/assistants/sessions/{session_id}/messages",
            headers=HEADERS,
            params=params,
        )
        resp.raise_for_status()
        page = resp.json()
        items.extend(page["items"])
        if not page["meta"]["has_next"]:
            break
        cursor = page["meta"]["next_cursor"]
    return items


if __name__ == "__main__":
    project_id = project_id_by_name(PROJECT_NAME)
    if not project_id:
        raise SystemExit(f"project not found: {PROJECT_NAME!r}")

    sessions = list_sessions_for_project(
        project_id,
        timestamp_start=TIMESTAMP_START,
        timestamp_end=TIMESTAMP_END,
    )
    print(f"{len(sessions)} sessions in range for {PROJECT_NAME}")

    for s in sessions:
        session_id = s["session_id"]
        messages = get_all_messages(session_id, fmt="resolved")
        print(f"\n{s['session_name']!r} ({session_id}): {len(messages)} messages")
        for m in messages:
            for e in m["events"]:
                if e["type"] in ("user_message", "assistant_message"):
                    print(f"  [{m['role']}] {e['text'][:80]}")
                elif e["type"] == "tool_call":
                    print(f"  [tool_call] {e['name']}")
                elif e["type"] == "tool_result":
                    status = "error" if e.get("is_error") else "ok"
                    print(f"  [tool_result] {e.get('tool_name') or ''} ({status})")