kindgi.client

The Kindgi API client.

from kindgi.client import Kindgi

client = Kindgi()  # KINDGI_API_URL, KINDGI_API_TOKEN
run = client.runs.start(flow="acme.ledger.record-flow", input={"vendor": "Acme"})
print(run.status, run.output)

Every operation of the API (@kindgi/api/openapi.json) is a method: the operation id approvals.reviewers.list is client.approvals.reviewers.list(). Request bodies are a model from kindgi.client.models, a mapping, or their fields as keywords; answers are models. Errors are typed (NotFoundError, GuardrailViolationError, …). AsyncKindgi is the same with asyncio.

AsyncKindgi

class AsyncKindgi(
    base_url: str | None = None,
    *,
    token: str | None = None,
    timeout: float = 60.0,
    max_retries: int = 2,
    default_headers: Mapping[str, str] | None = None,
    http_client: httpx.AsyncClient | None = None,
)

The Kindgi API, with asyncio.

async with AsyncKindgi() as client: run = await client.runs.start(flow="acme.ledger.record-flow", input={...}) async for event in client.runs.stream(run.id): ...

AuthError

class AuthError(
    message: str,
    *,
    reason: Literal['unauthenticated', 'forbidden', 'token-expired'],
    **kw: Any,
)

ConflictError

class ConflictError(
    message: str,
    *,
    status: int | None = None,
    server_code: str | None = None,
    details: Mapping[str, Any] | None = None,
    request_id: str | None = None,
)

GuardrailViolationError

class GuardrailViolationError(
    message: str,
    *,
    violations: list[Mapping[str, Any]],
    evaluation_errors: list[Mapping[str, Any]],
    **kw: Any,
)

InvalidRequestError

class InvalidRequestError(message: str, *, issues: list[Mapping[str, Any]], **kw: Any)

Kindgi

class Kindgi(
    base_url: str | None = None,
    *,
    token: str | None = None,
    timeout: float = 60.0,
    max_retries: int = 2,
    default_headers: Mapping[str, str] | None = None,
    http_client: httpx.Client | None = None,
)

The Kindgi API, synchronously.

client = Kindgi("http://127.0.0.1:4000", token="kgi_bt_…")
client = Kindgi()  # KINDGI_API_URL, KINDGI_API_TOKEN
run = client.runs.start(agent="acme.bookkeeper", input={"userMessage": "hi"})

Resources follow the API's operation ids (client.approvals.reviewers.list()). Use it as a context manager, or close() it, to release connections.

KindgiApiError

class KindgiApiError(
    message: str,
    *,
    status: int | None = None,
    server_code: str | None = None,
    details: Mapping[str, Any] | None = None,
    request_id: str | None = None,
)

A Kindgi API call failed.

NetworkError

class NetworkError(
    message: str,
    *,
    status: int | None = None,
    server_code: str | None = None,
    details: Mapping[str, Any] | None = None,
    request_id: str | None = None,
)

The request didn't get an answer (connection, timeout, a body that isn't JSON).

NotFoundError

class NotFoundError(message: str, *, kind: str, id: str, **kw: Any)

RateLimitedError

class RateLimitedError(
    message: str,
    *,
    retry_after_seconds: float | None = None,
    **kw: Any,
)

ServerError

class ServerError(
    message: str,
    *,
    status: int | None = None,
    server_code: str | None = None,
    details: Mapping[str, Any] | None = None,
    request_id: str | None = None,
)

apaginate

apaginate(*args: Any, **kwargs: Any) -> AsyncIterator[T]

paginate for AsyncKindgi: async for run in apaginate(client.runs.list): ….

paginate

paginate(*args: Any, **kwargs: Any) -> Iterator[T]

Every item of a list operation, page after page: paginate(client.runs.list, limit=50).