List-method results¶

Every .list()-style method on kneo_client returns a typed wrapper around the server response. There are two wrapper classes — chosen by the underlying server-response shape, not by "does this paginate":

Page[T] for list-shaped responses (runs, checkpoints, trace events, human tasks, audit events, the skill catalog).
Map[K, V] for dict-keyed responses (credentials inventory, per-environment policies).

Both live in kneo_client.core.results and are exported from kneo_client.core. The free-function iterate_all() async iterator is also there and walks pages for Page returns whose endpoint supports it.

Per-method return shapes¶

Method	Returns	Server-side pagination
`client.platform.runs.list()`	`Page[Run]`	Full `limit` / `offset` / `total`
`client.platform.runs.checkpoints(run_id)`	`Page[Checkpoint]`	Full `limit` / `offset` / `total`
`client.platform.runs.trace(run_id)`	`Page[TraceEvent]`	Full `limit` / `offset` / `total`
`client.platform.human_tasks.list()`	`Page[HumanTask]`	Full `limit` / `offset` / `total`
`client.platform.audit.list()`	`Page[AuditEvent]`	Full `limit` / `offset` / `total` (since `kneo_serv` 0.6.0; see below)
`client.agent.skills.list()`	`Page[Skill]`	Full `limit` / `offset` / `total` (endpoint requires `kneo_serv` ≥ 0.8.0)
`client.platform.credentials.list()`	`Map[str, Credential]`	None — whole map returned in one call
`client.platform.policies.environment_list()`	`Map[str, EnvironmentPolicy]`	None — whole map returned in one call

The item / value types (Run, Checkpoint, Credential, etc.) are the generated models re-exported from kneo_client.types — import them from there (kneo_client._generated is private and unsupported). Several are open objects in the pinned spec (no declared fields), in which case fields are reached via bracket lookup — run["run_id"] rather than run.run_id. The API Reference lists which per-endpoint item models are typed vs open.

`Page[T]`¶

@dataclass(frozen=True)
class Page(Generic[T]):
    items: list[T]
    total: int | None = None
    limit: int | None = None
    offset: int | None = None
    sort_by: str | None = None
    sort_order: str | None = None
    window: int | None = None
    complete: bool | None = None  # runs.trace only (kneo_serv >= 1.2.0)
    dropped: int | None = None    # runs.trace only: events dropped by the collector

    @property
    def count(self) -> int: ...        # = len(items)

    @property
    def has_more(self) -> bool: ...    # False when offset/total are None; clamps to window

    def __iter__(self) -> Iterator[T]: ...
    def __len__(self) -> int: ...
    def __getitem__(self, index: int) -> T: ...

Metadata fields are Optional — a server that doesn't echo a particular field (e.g. a pre-0.6.0 kneo_serv answering audit.list(), or a pre-0.9.0 one for window) leaves it as None rather than carrying a synthesized value. has_more derives sensibly: False whenever offset or total is None, and clamped to window when the server discloses one (see below).

Iterating one page¶

Page is a sequence; iterate it directly:

page = await client.platform.runs.list(status="running", limit=50)
print(f"got {page.count} of {page.total} runs")
for run in page:
    print(run["run_id"])

Indexing and len work as expected:

first_run = page[0]
total_on_this_page = len(page)

Walking all pages: `iterate_all()`¶

For the fully-paginating endpoints, iterate_all() walks pages transparently. Pass a fetch_page(limit, offset) callable that returns the next Page:

from kneo_client.core import iterate_all

async def fetch(limit: int, offset: int):
    return await client.platform.runs.list(limit=limit, offset=offset, status="running")

async for run in iterate_all(fetch, page_size=200):
    print(run["run_id"])

iterate_all() does three things on your behalf:

Clamps page_size to MAX_PAGE_SIZE = 1000 (the platform's hard upper bound). Asking for more silently downsizes.
Walks offset automatically — each page's offset plus its count becomes the next page's starting position.
Stops when has_more is False — either because the server has no more fetchable data (offset + count ≥ total, clamped to the paging window — see next section) or because the server doesn't echo enough metadata to know (pre-0.6.0 audit, see below).

`window` and the true `total`¶

Since kneo_serv 0.9.0, list responses disclose window — the deepest paging offset the server will reach — and total is the true store-side count (COUNT(*)), which can exceed it. When both are present, has_more reports whether more items are fetchable, not merely whether they exist: it clamps total to window, so pagination loops (including iterate_all()) stop honestly at the window edge instead of fetching a guaranteed-empty page. To detect items beyond the reachable window, compare page.total against page.window:

page = await client.platform.runs.list(limit=100)
if page.window is not None and page.total is not None and page.total > page.window:
    print(f"{page.total - page.window} runs exist beyond the paging window")

Pre-0.9.0 servers don't echo window; it stays None and has_more falls back to the plain offset + count < total derivation.

Audit is fully paginated¶

audit.list() has been fully paginated since kneo_serv 0.6.0 / kneo-client 0.6.0: the endpoint accepts limit / offset / sort_by / sort_order, the response echoes total / offset and sort metadata, and iterate_all() walks it like any other endpoint:

page = await client.platform.audit.list(event_type="run.created", limit=200)
print(f"got {page.count} of {page.total} matching audit events")

async def fetch(limit: int, offset: int):
    return await client.platform.audit.list(
        event_type="run.created", limit=limit, offset=offset
    )

async for event in iterate_all(fetch, page_size=200):
    print(event)

One back-compat note: pre-0.6.0 servers omit the audit pagination metadata, in which case the fields degrade to None, has_more is False, and iterate_all() fetches once and exits.

Choosing a `page_size`¶

Rules of thumb for the fully-paginating endpoints:

100–200 for most interactive flows — snappy first-byte latency.
500–1000 for back-end exports — fewer requests, higher per-request cost on retry.
< 50 when per-item downstream processing is slow and you want to start producing output sooner.

The hard ceiling is MAX_PAGE_SIZE = 1000; iterate_all() clamps for you.

`Map[K, V]`¶

@dataclass(frozen=True)
class Map(Generic[K, V]):
    items: Mapping[K, V]

    @property
    def count(self) -> int: ...        # = len(items)

    def __getitem__(self, key: K) -> V: ...
    def __iter__(self) -> Iterator[V]: ...    # iterates values
    def __len__(self) -> int: ...
    def __contains__(self, key: object) -> bool: ...

    def get(self, key: K, default: V | None = None) -> V | None: ...
    def keys(self) -> KeysView[K]: ...
    def values(self) -> ValuesView[V]: ...

Map covers the two endpoints whose responses are keyed maps rather than item lists. The whole collection comes back in one call — there is no limit / offset to think about.

inv = await client.platform.credentials.list()
print(f"{inv.count} credential references")
for credential in inv:                            # iterates values
    print(credential["provider"])

aws = inv["cred-aws"]                             # bracket lookup
if "cred-missing" in inv:                         # membership test
    ...
for cred_id in inv.keys():                        # explicit keys view
    ...

# Pair iteration uses the underlying Mapping when needed:
for cred_id, body in inv.items.items():
    print(cred_id, body["provider"])

Iteration yields values, not keys — consistent with Page iterating items. Use .keys() or .items.items() if you need keys.

Filters + paging¶

Resource-specific filter kwargs compose with paging kwargs:

# Just the failures, walked in pages of 200
async def fetch(limit, offset):
    return await client.platform.runs.list(
        status="failed", limit=limit, offset=offset
    )

async for run in iterate_all(fetch, page_size=200):
    ...

# Audit events for one run, first 500 (paginate onward with offset=)
events = await client.platform.audit.list(run_id="r1", limit=500)

Drop to the raw response¶

If you need a field the wrapper doesn't expose — for example the run_id echo on CheckpointListResponse / TraceResponse, or a field the platform adds in a newer release — bypass the wrapper with the public client.request() escape hatch and parse the body yourself:

resp = await client.request(
    "GET", f"/v1/runs/{run_id}/checkpoints", params={"limit": 100}
)
raw = resp.json()
print(raw["run_id"], raw["checkpoints"], raw["total"])

client.request(...) is the supported forward-compat path (full pipeline, raw httpx.Response); prefer a typed .platform / .agent wrapper whenever one exists.

client.request() is part of the stable 1.x contract; you parse the raw httpx.Response yourself, so anything you build on top of an unwrapped endpoint tracks the /v1 payload directly and won't get a typed model — re-evaluate at each kneo_serv pin bump, and switch to a typed .platform / .agent wrapper once one exists. (The older client._transport.request(...) still works but is internal and unsupported — prefer client.request(...).)

Auto-walking is not on the roadmap¶

client.platform.runs.list_all() returning an async iterator — i.e. auto-paged at the adapter layer — is intentionally not planned. The current shape (list methods return one Page; iterate_all() walks pages explicitly) keeps the per-call cost transparent and lets the caller decide when to stop. An accidental wide filter shouldn't walk millions of items behind the caller's back.

If you want a one-liner in your own code, write a small helper around iterate_all() and the relevant .list() callable — the pattern is identical for every fully-paginating endpoint.