Python SDK

Install and use krun-ai, the official Python SDK for the Krun API: sync and async clients, errors, timeouts and retries.

The official Python SDK for the Krun API.

• Package: krun-ai on PyPI (import name krun)
• Source: github.com/krun-ai/krun-python
• Version: 0.1.0, Python 3.10 or later, Apache-2.0
• Clients: Krun (sync) and AsyncKrun (async), with the same methods and types

Install

pip install krun-ai

Set your API key:

export KRUN_API_KEY=krun_live_...

Basic decision

from krun import Krun

client = Krun()  # reads KRUN_API_KEY

result = client.decide(
    context="Customer wants to return an item.",
    questions={
        "department": {
            "type": "choice",
            "options": {
                "shipping": "Shipping and delivery issues",
                "returns": "Returns and refunds",
                "billing": "Billing and payment issues",
            },
        }
    },
)

answer = result.answers["department"]
answer.choice             # "returns", or None if the model abstains
answer.confidence         # 0.967536: top-1 minus top-2 probability
answer.probabilities      # {"shipping": 0.011351, "returns": 0.978886, "billing": 0.009763}
answer.abstain            # False
answer.abstention_status  # "advisory"
result.usage.input_tokens # 49
result.request_id         # "req_...", from the X-Request-ID header

decide() returns a DecisionResult:

Attribute	Type	Description
model	str	Model that answered, krun-one-v0.
answers	dict[str, ChoiceAnswer]	One answer per question id, in request order.
usage	Usage	usage.input_tokens: int, or None if not reported.
request_id	str	The request id. Pass it to feedback().

Each ChoiceAnswer has type, choice (str | None), confidence, probabilities, abstain and abstention_status. See Questions and answers.

When the model abstains, choice is None. The SDK never replaces it with the most likely option. Read probabilities for a best guess. See Abstention.

decide() arguments:

Argument	Description
context	The text to decide on, 1 to 8,000 characters.
questions	Question id to question, 1 to 16 questions.
model	Optional. Defaults to krun-one-v0.
request_id	Optional X-Request-ID to send: 1 to 128 characters of [A-Za-z0-9._:-].
timeout	Optional per-call timeout in seconds.

Multiple questions

Ask several questions about the same context in one call:

result = client.decide(
    context="My package never arrived and I need it for a wedding this Saturday.",
    questions={
        "department": {
            "type": "choice",
            "options": {"shipping": "", "returns": "", "billing": ""},
        },
        "priority": {
            "type": "choice",
            "options": {"low": "Can wait", "normal": "Normal priority", "high": "Needs quick attention"},
        },
        "sentiment": {
            "type": "choice",
            "options": {"positive": "", "neutral": "", "negative": ""},
        },
    },
)

for question_id, answer in result.answers.items():
    print(question_id, answer.choice, answer.confidence)

See Multiple questions.

Tool routing

Set task_type="tool" on the question:

result = client.decide(
    context="Find my meetings tomorrow.",
    questions={
        "tool": {
            "type": "choice",
            "task_type": "tool",
            "options": {
                "calendar_search": "Search calendar events",
                "send_email": "Send an email",
            },
        }
    },
)

result.answers["tool"].choice             # "calendar_search"
result.answers["tool"].abstention_status  # "advisory"

Questions can also be written as objects with ChoiceQuestion:

from krun import ChoiceQuestion

questions = {
    "tool": ChoiceQuestion(
        options={"calendar_search": "Search calendar events", "send_email": "Send an email"},
        task_type="tool",
    )
}

See Tool routing.

Feedback

client.feedback(
    request_id=result.request_id,
    question_id="department",
    correct=False,
    expected_decision="billing",
    metadata={"ticket": "T-1234"},  # optional JSON object, up to 8 KiB, no personal data
)

feedback() returns a Feedback object with id, request_id, question_id and created_at. It raises NotFoundError if the request id was not decided by your project. See Feedback.

Models

for model in client.models():
    print(model.id, model.status)  # krun-one-v0 available

Async client

AsyncKrun has the same arguments, methods and return types as Krun, as coroutines:

import asyncio

from krun import AsyncKrun

QUESTIONS = {
    "department": {
        "type": "choice",
        "options": {
            "shipping": "Shipping and delivery issues",
            "returns": "Returns and refunds",
            "billing": "Billing and payment issues",
        },
    }
}

MESSAGES = [
    "Where is my package? It was due on Monday.",
    "I want to send these shoes back.",
    "Why is there a second charge on my card?",
]


async def main() -> None:
    async with AsyncKrun() as client:
        results = await asyncio.gather(
            *(client.decide(context=m, questions=QUESTIONS) for m in MESSAGES)
        )
        for message, result in zip(MESSAGES, results):
            print(result.answers["department"].choice, "<-", message)


asyncio.run(main())

Concurrent requests count toward your per-minute rate limit. To ask several questions about the same text, use one request with multiple questions instead.

Configuration

from krun import Krun

client = Krun(
    api_key="krun_live_...",   # default: the KRUN_API_KEY environment variable
    base_url="https://api.krun.ai",  # default
    timeout=70.0,              # seconds per attempt (default 70)
    max_retries=1,             # decide() and models() only (default 1)
    http_client=None,          # optional httpx.Client, for proxies or custom transports
)

Use the client as a context manager, or call client.close(), to release connections:

with Krun() as client:
    result = client.decide(context="...", questions={...})

If no API key is passed and KRUN_API_KEY is not set, Krun() raises KrunError. The key never appears in repr(client), error messages or logs.

Timeouts

The default timeout is 70 seconds per attempt, because a serverless cold start can use most of the API’s 60-second deadline. The timeout can’t be disabled: None, 0 and infinity are rejected. When it elapses, the SDK raises APITimeoutError.

client = Krun(timeout=30.0)                    # for all calls
client.decide(..., timeout=10.0)               # for one call

See Timeouts and cold starts.

Retries

The API already retries its model backend, so the SDK retries only a little:

Method	Retried on	Default
decide(), models()	Connection errors, HTTP 502, 503, 504	1 retry
feedback()	Never: it writes a record and the API has no idempotency key	–

• The wait follows Retry-After when the API sends it, up to 10 seconds. Otherwise it is 0.5 s, then 1 s, 2 s, and so on, with jitter.
• The SDK’s own timeout (APITimeoutError) is not retried.
• Other 4xx errors and 500 are not retried.
• A retried decide() can count one extra decision against usage if the first attempt reached the model.
• Set max_retries=0 to disable retries.

Errors

All errors inherit from krun.KrunError and expose message, request_id, status_code and error_code when available.

KrunError
├── APIError                     the API answered with an error (status_code always set)
│   ├── InvalidRequestError      400/413  INVALID_REQUEST, INVALID_OPTIONS, PAYLOAD_TOO_LARGE
│   ├── AuthenticationError      401      UNAUTHORIZED
│   ├── NotFoundError            404      NOT_FOUND
│   ├── RateLimitError           429      RATE_LIMITED           (.retry_after)
│   ├── QuotaExceededError       429      QUOTA_EXCEEDED
│   ├── InferenceFailedError     502      INFERENCE_FAILED
│   ├── ServiceUnavailableError  503      UPSTREAM_UNAVAILABLE   (.retry_after)
│   ├── UpstreamTimeoutError     504      UPSTREAM_TIMEOUT
│   └── InternalServerError      500      INTERNAL_ERROR
├── APIConnectionError           no HTTP response (DNS, refused, reset, TLS)
│   └── APITimeoutError          the SDK timeout elapsed
└── APIResponseValidationError   a 2xx response did not match the contract

import time

import krun

try:
    result = client.decide(context="...", questions={...})
except krun.InvalidRequestError as e:
    print(e.error_code, e.message, e.request_id)  # fix the request
except krun.RateLimitError as e:
    time.sleep(e.retry_after or 1)
except krun.QuotaExceededError:
    ...  # the monthly quota is exhausted: retrying will not help
except krun.KrunError as e:
    print("Krun request failed:", e)

Arguments of the wrong type, like context=None, raise TypeError before any request is sent. See Errors for the API error codes.

Logging

The SDK is silent by default and has no telemetry. It logs to the krun logger at DEBUG level: method, path, status, request id and retries only. It never logs the API key, context, options or answers.

import logging

logging.getLogger("krun").setLevel(logging.DEBUG)

Types

The krun package exports typed request and response types: ChoiceQuestion, ChoiceQuestionParam, QuestionsParam, DecisionResult, ChoiceAnswer, Usage, Feedback, Model, TaskType and AbstentionStatus. The package ships with py.typed, so type checkers like mypy and pyright use them.