Error handling¶

This guide explains the typed exception hierarchy kneo-client raises, what each exception carries, and how to write robust catch blocks for the common operational shapes.

What "errors" mean in kneo-client¶

Every failure — at any layer — surfaces as a typed exception derived from KneoError. There is no (ok, err) tuple return, no Response[T] wrapper, no errno field. The standard Python try / except flow is the only error-handling shape.

Each exception carries enough context to:

Log the failure with traceability — every exception has .request_id (the server-assigned correlation ID) and, for POST failures, .idempotency_key.
Branch on the operational meaning — the exception class encodes "what went wrong" (auth vs. permission vs. server outage vs. network).
Read the server's reason — .body is the parsed JSON the platform returned (or raw text when the response wasn't JSON).
Decide whether to retry, escalate, or surface — combined with .status and the exception type, you can route the failure programmatically.

Hierarchy¶

KneoError
├── KneoNetworkError              # DNS / connect / TLS / read timeout — wrapped from httpx.HTTPError
├── KneoAuthError                 # HTTP 401 — missing or invalid API key
├── KneoPermissionError           # HTTP 403 — key valid but lacks the required scope
├── KneoNotFoundError             # HTTP 404 — resource does not exist
├── KneoConflictError             # HTTP 409 (generic)
│   └── KneoIdempotencyMismatchError   # HTTP 409 with payload mismatch on a replayed Idempotency-Key
├── KneoRateLimited               # HTTP 429 (carries .retry_after)
└── KneoServerError               # HTTP 5xx — server-side failure

KneoIdempotencyMismatchError is a subclass of KneoConflictError (catching KneoConflictError also catches the mismatch case). Everything else is parallel.

What every exception carries¶

class KneoError(Exception):
    status: int | None          # HTTP status, or None for transport-level failures
    body: Any                   # Parsed JSON dict, raw text, or None
    request_id: str | None      # X-Request-ID echoed by the server
    idempotency_key: str | None # Idempotency-Key sent on the failing request (POSTs)

KneoRateLimited adds one field:

class KneoRateLimited(KneoError):
    retry_after: float | None   # Seconds parsed from the Retry-After header

All other subclasses inherit from KneoError without adding fields.

Status code → exception mapping¶

HTTP status	Exception	When
401	`KneoAuthError`	Missing or invalid API key. Re-check the profile resolution chain.
403	`KneoPermissionError`	API key is valid but the platform won't authorize this operation. Check the key's scopes / role.
404	`KneoNotFoundError`	The resource (run, spec, environment, etc.) doesn't exist. Often a stale ID.
409	`KneoConflictError`	Generic conflict — resource state, optimistic-lock failure, etc.
409 with idempotency key	`KneoIdempotencyMismatchError`	Same key reused with a different payload — see idempotency.
429	`KneoRateLimited`	Rate limit hit. `.retry_after` carries the server's hint.
5xx	`KneoServerError`	Server-side failure. The transport already retried within its policy (for 502/503/504); a `KneoServerError` reaching your code means retries exhausted.
Other (1xx / 3xx / 4xx that aren't above)	`KneoError` (base)	Catch-all for unmodeled statuses.
Connection / DNS / TLS / read timeout	`KneoNetworkError`	Transport-level failure. The transport already retried for transient errors; a `KneoNetworkError` reaching your code means retries exhausted.

Catching patterns¶

Catch broadly, log richly¶

The most common pattern — log the full context, then decide whether to re-raise:

from kneo_client.core.errors import KneoError

try:
    run = await client.platform.runs.create(payload)
except KneoError as exc:
    log.error(
        "create_run failed status=%s request_id=%s idempotency_key=%s body=%r",
        exc.status,
        exc.request_id,
        exc.idempotency_key,
        exc.body,
    )
    raise

The request_id is the link to the platform's audit events — pass it along when reporting a problem to the platform operators.

Branch on specific status¶

When the operational meaning matters (auth flow, retry decision, user-visible error message):

from kneo_client.core.errors import (
    KneoAuthError,
    KneoNotFoundError,
    KneoRateLimited,
    KneoServerError,
)

try:
    run = await client.platform.runs.get(run_id)
except KneoAuthError:
    print("API key is missing, invalid, or revoked.")
    raise
except KneoNotFoundError:
    print(f"run {run_id!r} does not exist.")
    return None
except KneoRateLimited as exc:
    print(f"rate-limited; server suggests waiting {exc.retry_after}s")
    await asyncio.sleep(exc.retry_after or 10)
    raise
except KneoServerError as exc:
    log.error("platform 5xx (after retries): %s", exc.body)
    raise

Order matters: catch more specific exceptions first (KneoNotFoundError before KneoError).

Handle idempotency-key mismatches loudly¶

A 409 with an idempotency key set means the same key was reused with a different payload. This is almost always a caller bug — surface it explicitly:

from kneo_client.core.errors import KneoIdempotencyMismatchError

try:
    await client.platform.runs.create(payload, idempotency_key=key)
except KneoIdempotencyMismatchError as exc:
    raise RuntimeError(
        f"Idempotency-Key {exc.idempotency_key!r} was reused with a different payload. "
        f"Either generate a new key for the new request or fix the payload drift."
    ) from exc

See Idempotency and retries for the full story on how / when this happens.

Don't catch successful-status branches¶

Methods on the platform / agent clients return parsed response models on success and raise on failure. There is no "ok / err" branching at the call site. Wrap the call in try / except, not the return value:

# Right
try:
    run = await client.platform.runs.create(payload)
    process(run)
except KneoError:
    ...

# Wrong — runs.create never returns None / False on failure; it raises
result = await client.platform.runs.create(payload)
if result is None:  # never happens
    ...

Transport errors specifically¶

KneoNetworkError covers everything below the HTTP layer: DNS resolution, TCP connect failures, TLS handshakes, read timeouts. It wraps the underlying httpx.HTTPError as the cause — exc.__cause__ is the original httpx exception if you need to inspect it.

The transport retries these automatically within RetryPolicy.max_attempts for transport errors and for HTTP 429 / 502 / 503 / 504. So a KneoNetworkError reaching your code means all retries exhausted:

from kneo_client.core.errors import KneoNetworkError

try:
    health = await client.platform.health.readyz()
except KneoNetworkError as exc:
    print(f"could not reach the platform: {exc}")
    # Treat as a hard dependency outage; don't pretend the call succeeded.
    raise

If you want to see the retry behavior in your logs, set the kneo_client.transport logger to INFO:

import logging
logging.getLogger("kneo_client.transport").setLevel(logging.INFO)
# → INFO kneo_client.transport: transport error on attempt 1; sleeping 0.20s: ...

Building error messages for users¶

The exceptions are designed for internal error handling, not for user-facing messages. If you're surfacing platform errors to end users (in a dashboard UI, a CLI prompt, etc.), build a friendly message from the exception's attributes rather than printing str(exc):

def user_message(exc: KneoError) -> str:
    if isinstance(exc, KneoAuthError):
        return "Your API key is invalid. Please check your credentials."
    if isinstance(exc, KneoPermissionError):
        return "You don't have permission to perform this action."
    if isinstance(exc, KneoNotFoundError):
        return "The requested item could not be found."
    if isinstance(exc, KneoRateLimited):
        wait = exc.retry_after or 60
        return f"Too many requests. Please try again in {int(wait)}s."
    if isinstance(exc, KneoServerError):
        return f"Server error (request {exc.request_id}). Please report this."
    if isinstance(exc, KneoNetworkError):
        return "Could not reach the server. Check your network connection."
    return f"Unexpected error: {exc}"

Why a typed hierarchy¶

A single KneoError would force callers to inspect .status everywhere they want to branch. A flat enum of error codes would lose the natural isinstance ergonomics. The hierarchy lets you catch broadly (except KneoError) for logging and narrowly (except KneoAuthError) for recovery — without losing the underlying response context, which stays attached to the exception instance.

Subclasses are added when the platform introduces a new HTTP status that warrants its own catch site, and not before. The set above is sufficient for /v1 as it stands at kneo_serv 0.4.0.

Reference¶

Helper	Where	What it does
`from_response(response, *, idempotency_key=None)`	`kneo_client.core.errors`	Maps an `httpx.Response` to the appropriate `KneoError` subclass. Used internally by `Transport`; rarely called directly.
`KneoError(message, *, status, body, request_id, idempotency_key)`	`kneo_client.core.errors`	The base. All subclasses share this constructor signature (except `KneoRateLimited`, which adds `retry_after`).