SDKs

Python SDK

A sync client, an async client, one exception tree, and typed results from your own constants.
pip install cloudraker-milliseconds
export MS_API_KEY=test_sk-...

Python 3.10 or later. The only runtime dependency is httpx, plus typing-extensions on Python 3.10. The package ships py.typed.

from milliseconds import DecisionMachine
dm = DecisionMachine()  # reads MS_API_KEY
r = dm.classify("I was charged twice.", {"billing": "payments and refunds", "shipping": "delivery"})
r.label

What Python can and cannot infer

Python has no mapped types and no conditional return types. It cannot read your label names out of a dict display. It can solve a TypeVar from an annotated constant. One annotation buys the whole chain.

from typing import Final, Literal, Mapping
from milliseconds import DecisionMachine
Intent = Literal["billing", "shipping", "account"]
LABELS: Final[Mapping[Intent, str]] = {
    "billing": "payments, invoices, charges and refunds",
    "shipping": "delivery, tracking and packages",
    "account": "login, passwords and profile settings",
}
dm = DecisionMachine()  # reads MS_API_KEY
r = dm.classify("I was charged twice.", LABELS)
r.label  # Intent. A match statement over it is exhaustive.
r.scores["billing"]  # ok. r.scores["refunds"] is a type error.
r.probability  # float
r.confidence  # float. 1 = one clear winner, 0 = flat.

Without the annotation you get ClassifyResult[str]. Nothing breaks. You lose only the names. Keep every label set in one constants file, and the annotation lands where the constants already live.

classify_tree returns label: str. Python has no expression that reads literals out of a nested dict.

Client options

dm = DecisionMachine(
    api_key=None,                            # falls back to MS_API_KEY
    base_url="https://api.milliseconds.ai",
    timeout=60.0,                            # per attempt, in seconds
    max_retries=2,
    headers=None,                            # merged into every request
    http_client=None,                        # your own httpx.Client
)

timeout, max_retries and headers also work per call: dm.classify(text, LABELS, max_retries=5, timeout=10.0).

The client is a context manager. close() closes only a pool the SDK opened itself.

The eight capabilities

statement = dm.yes_no(
    "Fix this today.",
    "The customer expresses urgency.",
    when_true="Time pressure, ASAP, losing money",
)
statement.answer  # bool
statement.probability  # float
label = dm.classify("I was charged twice.", LABELS)
tree = dm.classify_tree(
    "I want my money back.",
    {
        "billing": {
            "description": "payments, invoices, charges, refunds and subscriptions",
            "labels": {
                "refund_request": "the customer asks for money back",
                "subscription_change": "the customer wants to upgrade or cancel a plan",
            },
        },
        "shipping": "delivery, tracking, lost or damaged parcels",
    },
)
tree.path  # winning label per level, top to bottom
tree.label  # the deepest label
mood = dm.rate(
    "I am done with this company.",
    ["Calm", "Annoyed", "Angry", "Threatening to leave"],
)
mood.score  # 0 to len(scale) - 1. Route on this.
mood.level  # the most likely index. It flips on 0.001.
who = dm.answer("Apple announced the M5 today.", "Who announced the product?")
if who.span is not None:
    start, end = who.span
people = dm.entities("Ada met Grace in Paris.", {"person": "a human name", "place": "a city"})
[e.text for e in people]
check = dm.verify("Invoice 4471, total 120.00 EUR.", "invoice_number", 4471)
check.matches  # bool
check.found  # what the text actually says

Describe every label. The label text is the instruction, and the model reads it literally. Described labels score measurably better than bare names.

Annotate a tree you keep in a constant, as you annotate a label set. A bare TAXONOMY = {...} infers a wider type, and classify_tree then refuses it.

from typing import Final
from milliseconds import Tree
TAXONOMY: Final[Tree] = {
    "billing": {
        "description": "payments, invoices, charges, refunds and subscriptions",
        "labels": {
            "refund_request": "the customer asks for money back",
            "subscription_change": "the customer wants to upgrade or cancel a plan",
        },
    },
    "shipping": "delivery, tracking, lost or damaged parcels",
}
walked = dm.classify_tree("I want my money back.", TAXONOMY)
walked.label

Batching

Pass a list of texts for a batch. The reply follows your request, never the other way.

tickets = ["I was charged twice.", "Where is my parcel?"]
many = dm.classify(tickets, LABELS)  # Results[ClassifyResult[Intent]]
many[0].label
many.usage.input_tokens  # the usage of the one call
grid = dm.yes_no(tickets, ["The text mentions a price.", "The customer is angry."])
grid[0][1].answer  # text 0, statement 1

Results is a list with a usage attribute. The limits are 32 texts per call, and 20,000 characters per text. The SDK never splits a batch for you. Splitting costs money and changes failure modes, so you decide. See Batching.

Extraction

Four schema shapes work: a plain dict JSON Schema, a TypedDict, a dataclass, and a pydantic v2 model. The SDK never imports pydantic. It calls model_json_schema() by duck typing.

Declare every field | None. A missing value comes back as None.

from typing import Literal, TypedDict
class Invoice(TypedDict):
    invoice_number: str | None
    total: float | None
    currency: Literal["USD", "EUR"] | None
data = dm.extract("Invoice 4471, total 120.00 EUR.", Invoice)
data["total"]  # float | None

A dataclass and a pydantic v2 model work the same way, and return your own type:

from dataclasses import dataclass
@dataclass
class Vendor:
    name: str | None
    country: str | None
vendor = dm.extract("Acme SA, Paris, France.", Vendor)
vendor.name

A plain dict carries descriptions, which raise accuracy:

schema = {
    "type": "object",
    "properties": {
        "invoice_number": {"description": "the identifier printed on the invoice"},
        "total": {"type": "number", "description": "the amount due including tax"},
        "tags": {"type": "array", "items": {"type": "string"}},
    },
}
invoice = dm.extract("Invoice 4471, total 120.00 EUR.", schema)

Four degradations are real, and no Python annotation can hide them:

  • a missing value is None;
  • an array of objects always comes back [];
  • an array of scalars comes back as a list of strings;
  • an enum is not checked on the server, so a value outside your Literal can arrive.

See Extract.

Usage and rate limits

Request-limit values count inference request units, not HTTP calls. A batch with 24 texts consumes 24 units; text and statement or question counts multiply.

Every result carries the usage of the call that produced it. A single-text extract and post() are the two exceptions. Both return your own object, which has no place for the usage. Send a one-text batch to reach it: dm.extract([text], Invoice).usage.

r = dm.classify("I was charged twice.", LABELS)
r.usage.input_chars
r.usage.input_tokens  # what this call bills
r.usage.inference_ms  # model time, not wall clock
r.usage.headers["x-input-tokens"]  # every response header stays reachable
limits = r.usage.rate_limit  # RateLimit | None
if limits is not None:
    limits.remaining_requests
    limits.reset_requests  # OpenAI style, for example '1s'

The API reserves units before processing the call. Rate-limit headers show a snapshot after admission; concurrent calls can change the remaining balance.

Honor retry-after on rate-limit refusals. See Limits and rate limits.

Errors and retries

from milliseconds import (
    AuthenticationError,
    InvalidRequestError,
    MillisecondsError,
    OverloadedError,
    QuotaExceededError,
    RateLimitError,
)
try:
    r = dm.classify("I was charged twice.", LABELS)
except RateLimitError as e:
    print(e.retry_after, e.attempts)
except QuotaExceededError:
    print("add credits at https://console.milliseconds.ai")
except MillisecondsError as e:
    print(e.code, e.status, e.api_message)

Every exception subclasses MillisecondsError, so one except catches the lot.

ExceptionCodes it carries
InvalidRequestErrorinvalid_request, invalid_schema, client_error
AuthenticationErrormissing_api_key, invalid_api_key
RateLimitErrorrate_limit_exceeded
QuotaExceededErrorinsufficient_quota
RunnerErrorrunner_error
OverloadedErroroverloaded
ConnectionErrorconnection_error, timeout

The SDK retries 429 rate_limit_exceeded, 502 runner_error, 529 overloaded, and transport failures. Every capability is a pure function, so a retry is always safe. It never retries 400, 401, 403 or 429 insufficient_quota. A timer retry cannot fix a spent quota.

Pass max_retries=0 to turn retries off.

Some checks run before any HTTP call. They raise InvalidRequestError with code client_error and status 0. Nothing was sent, so no token was billed.

try:
    dm.classify("I was charged twice.", ["billing"])
except InvalidRequestError as e:
    print(e.code)  # client_error
    print(e.api_message)  # labels has 1 entry. classify needs 2 to 64.

Errors lists every code.

Async, and your own pool

import asyncio
from milliseconds import AsyncDecisionMachine
async def main() -> None:
    async with AsyncDecisionMachine() as adm:
        r = await adm.classify("I was charged twice.", LABELS)
        print(r.label)
asyncio.run(main())

The async client has the same methods and the same options. Both clients share one transport module, so the retries, the error parsing and the header parsing cannot drift. Pass http_client= to bring your own httpx.Client or httpx.AsyncClient. The SDK closes only a pool it opened itself.

Any path, any body

raw = dm.post("/v1/decision-machine-1/classify", {"text": "hi", "labels": ["a", "b"]})
raw["label"]

Gotchas

  • yes-no and answer are the two endpoints with no text refinement on the server. A body with neither text nor texts returns 200 and {"results": []}. The SDK always sends one of the two, so that body cannot reach the API.
  • A labels or types dict has no size limit on the server. The 2-to-64 rule binds the list form only, and the SDK checks the same way.
  • classify_tree re-sends the text at every level. The per-level input_chars therefore do not sum to usage.input_chars, which counts one pass over the body.
  • The SDK reports x-input-tokens. It never estimates a cost.