> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.milliseconds.ai/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.milliseconds.ai/_mcp/server.

# Python SDK

```bash
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`.

```python
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.

```python
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

```python
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

```python
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.

```python
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.

```python
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](/capabilities/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.

> **Warning**
>
> **Declare every field `| None`.** A missing value comes back as `None`.

```python
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:

```python
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:

```python
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](/capabilities/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`.

```python
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'
```

> **Note**
>
> 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](/reference/limits).

## Errors and retries

```python
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.

| Exception             | Codes it carries                                    |
| --------------------- | --------------------------------------------------- |
| `InvalidRequestError` | `invalid_request`, `invalid_schema`, `client_error` |
| `AuthenticationError` | `missing_api_key`, `invalid_api_key`                |
| `RateLimitError`      | `rate_limit_exceeded`                               |
| `QuotaExceededError`  | `insufficient_quota`                                |
| `RunnerError`         | `runner_error`                                      |
| `OverloadedError`     | `overloaded`                                        |
| `ConnectionError`     | `connection_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.

```python
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](/reference/errors) lists every code.

## Async, and your own pool

```python
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

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

## Gotchas

* Every endpoint needs exactly one of `text`, `texts` or `image`. A body with none of the three returns `400` `invalid_request`. The SDK always sends one, 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.