Quickstart¶

This guide walks you from a fresh shell to your first authenticated call against a running Kneo Agent Platform instance, then to creating a run and inspecting its trace.

What "quickstart" means in kneo-client¶

kneo-client is a typed Python SDK and adapter toolkit for the Kneo Agent Platform's /v1 HTTP API. Two products use it as their shared client layer — Kneo Agent Dashboard for operations and Kneo Agent Studio for development — but you can use it directly from any Python 3.12+ codebase that needs to talk to a platform instance.

A working setup needs three things:

The kneo-client package installed.
A profile — a (url, api_key, auth_scheme, timeout) tuple resolved from a TOML config file, environment variables, or explicit kwargs. See Profiles and auth for the full resolution model.
A reachable platform instance.

The rest of this page assumes you have all three.

Install¶

python -m pip install kneo-client

Requires Python ≥ 3.12 (3.12 / 3.13 / 3.14 tested in CI). The runtime closure is small: httpx, pydantic, anyio, platformdirs, attrs. No CLI is installed — kneo-client is a library, not a service.

Verify the install:

python -c "from kneo_client import KneoClient, __version__; print(__version__)"
# → 1.2.0

Configure a profile¶

The client reads connection details in this order (later sources override earlier):

A TOML config file at ~/.config/kneo/client.toml (XDG-style location via platformdirs).
Environment variables: KNEO_PROFILE, KNEO_URL, KNEO_API_KEY, KNEO_AUTH_SCHEME, KNEO_TIMEOUT.
Explicit keyword arguments to KneoClient.from_profile() (or the underlying load_profile()).

The minimum-friction setup is env vars — useful for one-off scripts and CI:

export KNEO_URL=https://kneo.example.com
export KNEO_API_KEY=your-api-key

For named multi-profile setups (e.g. default for prod, staging for non-prod), use the TOML file:

# ~/.config/kneo/client.toml
[default]
url = "https://kneo.example.com"
api_key = "prod-key"
auth_scheme = "bearer"
timeout = 30.0

[staging]
url = "https://staging-kneo.example.com"
api_key = "staging-key"

Switch profiles by name (KneoClient.from_profile("staging")) or by env (KNEO_PROFILE=staging).

See Profiles and auth for the full resolution semantics, both auth schemes, and the multi-profile workflow.

Ping the platform¶

The smallest possible script — verifies that auth flows correctly and the platform is reachable:

import asyncio
from kneo_client import KneoClient


async def main():
    async with KneoClient.from_profile() as client:
        ready = await client.platform.health.readyz()
        print(f"platform ready: ok={ready.ok} version={ready.version!r}")


asyncio.run(main())

What this exercises:

Profile resolution (TOML / env / kwargs).
API-key injection on the outgoing request (Authorization: Bearer … or X-Kneo-Api-Key: … depending on scheme).
The transport's retry loop (will retry on transient network failures).
Error mapping (a missing key surfaces as KneoAuthError, an unreachable URL as KneoNetworkError).

If you get a KneoAuthError, your API key isn't reaching the platform — re-check the profile resolution chain. If you get a KneoNetworkError, the URL is unreachable; see Error handling § Transport errors and network troubleshooting for timeouts, TLS, proxies, and how to inspect the underlying httpx cause.

Create your first run¶

A run is the unit of work the platform executes — it instantiates a spec (an agent definition) and tracks its lifecycle through queued → running → terminal {completed, failed, cancelled, timed_out, expired}. To create one:

async with KneoClient.from_profile() as client:
    # A run body needs an `input` (the task) plus a spec: a server-side
    # `spec_path` (shown here) or an inline `spec={...}` object.
    created = await client.platform.runs.create(
        {"input": "Summarize the latest activity.", "spec_path": "your-spec.yaml"}
    )
    print(f"run_id={created.run_id} status={created.status}")

    terminal = await client.platform.runs.wait_for_completion(
        created.run_id, poll_interval=2.0, timeout=600
    )
    print(f"final status: {terminal.status}")

    trace = await client.platform.runs.trace(created.run_id, limit=20)
    for event in trace:  # trace is a Page — iterate it directly (or use trace.items)
        print(event)

The interesting parts:

runs.create(...) auto-injects an Idempotency-Key header (UUID4). Re-running the same payload with the same key is safe — the platform replays the original response. See Idempotency and retries.
runs.wait_for_completion(...) polls runs.get(run_id) until the run reaches a terminal status. Default terminal set is {"completed", "failed", "cancelled", "timed_out", "expired"} — the platform's canonical terminal set; pass terminal_statuses={"blocked", …} to also stop when a run pauses for human review.
runs.trace(...) returns events the platform recorded during the run (tool calls, model calls, middleware decisions, policy checks, etc.).

This is the operational core of the platform adapter. Every other endpoint follows the same shape: client.platform.<resource>.<method>(...) returning a typed model.

Skills catalog and per-request overlays¶

The agent surface (client.agent.*) mirrors the same shape. One pairing worth knowing early: the platform publishes a skill catalog — the declared + default skills a spec may reference by name — and runs.create accepts a per-request skills overlay that adds or disables skills for just that run:

async with KneoClient.from_profile() as client:
    catalog = await client.agent.skills.list()  # Page; requires kneo_serv >= 0.8.0
    for skill in catalog:
        print(skill["name"], skill["description"])

    created = await client.platform.runs.create({
        "input": "Summarize the latest activity.",
        "spec_path": "your-spec.yaml",
        "skills": {"add": ["web_search"], "disable": ["code_exec"]},
    })

The skills overlay on runs.create requires kneo_serv >= 0.9.0, and every name in add must be declared in the spec's skills block — the server rejects undeclared names. Use agent.skills.list() to discover the valid targets.

Sync facade¶

If your caller can't run an event loop (a script, a notebook, a sync framework), wrap the transport with SyncTransport:

from kneo_client.core import load_profile
from kneo_client.core import SyncTransport

with SyncTransport(load_profile()) as transport:
    response = transport.request("GET", "/v1/healthz")
    print(response.json())

Under the hood SyncTransport runs Transport inside an anyio.from_thread.start_blocking_portal() — same retry / idempotency / error flows, called synchronously. The async surface is the recommended path; the sync facade is for ergonomics.

SyncTransport does not mount .platform / .agent adapters — those are async-only. Two patterns work for sync consumers who want the wrapped endpoints:

One-shot: `asyncio.run()`¶

If you're calling once and not in an existing event loop:

import asyncio
from kneo_client import KneoClient

async def fetch() -> str:
    async with KneoClient.from_profile() as client:
        run = await client.platform.runs.create(
            {"input": "Summarize the latest activity.", "spec_path": "s1.yaml"}
        )
        return run.run_id

run_id = asyncio.run(fetch())
print(run_id)

Repeated calls: `anyio.from_thread.start_blocking_portal()`¶

If you're calling many times from sync code (e.g., inside a synchronous CLI loop) and want to avoid the startup cost of asyncio.run() per call, hold open a portal — the same machinery SyncTransport itself uses:

from anyio.from_thread import start_blocking_portal
from kneo_client import KneoClient

with start_blocking_portal() as portal:
    client = portal.call(KneoClient.from_profile().__aenter__)
    try:
        run = portal.call(
            client.platform.runs.create,
            {"input": "Summarize the latest activity.", "spec_path": "s1.yaml"},
        )
        status = portal.call(client.platform.runs.get, run.run_id)
        print(status.status)
    finally:
        portal.call(client.__aexit__, None, None, None)

The portal runs an event loop on a background thread; portal.call(async_fn, *args) blocks the calling thread until the coroutine completes. This is the recommended pattern for long-lived sync code that talks to the platform repeatedly.

Where to go next¶

By topic:

Profiles and auth — multi-profile workflows, the two header schemes the platform supports, environment-variable precedence.
Idempotency and retries — when keys are auto-injected, how 409 mismatches surface, customizing the retry policy.
List-method results — Page[T] and Map[K, V] wrappers, walking pages with iterate_all() (audit included — fully paginated since kneo_serv 0.6.0), the paging window / true-total semantics.
Error handling — the full exception hierarchy and what to catch.
Compatibility matrix — which kneo-client releases support which kneo_serv versions.

For the comprehensive API reference (every class, method, exception), see the API Reference HTML or the PDF version.

For runnable end-to-end scripts: examples/.