Python SDK

Already using the openai or anthropic package? Keep it — point base_url at https://api.ai.ml/v1 and use an aiml key. That is the fastest path and it is tested in CI on every commit.

Reach for aiml when you want what a compatibility dialect cannot express: the IR's typed blocks, the routing controls, dry_run, and the cost that comes back on every response.

pip install aiml

A generation, streamed

from aiml import AIML, AIMLError

BASE = os.environ.get("AIML_BASE_URL", "https://api.ai.ml")
request = {
    "model": os.environ.get("AIML_MODEL", "openai/gpt-5-mini"),
    "messages": [{"role": "user", "content": [{"type": "text", "text": "Write one sentence about gateways."}]}],
    "generation": {"max_tokens": 128},
}

with AIML(base_url=BASE) as aiml:
    # what will this cost, and where will it go? Nothing is sent upstream.
    size = aiml.count_tokens(request)
    plan = aiml.dry_run(request)
    print(f"input_tokens={size['input_tokens']} source={size['source']}")
    print(f"plan={','.join(p['endpoint_id'] for p in plan['plan'])} reserve_micro={plan['reserve_micro']}")

    # now send it, streaming; leaving the block closes the socket
    with aiml.stream(request) as s:
        for text in s.text_deltas():
            print(text, end="", flush=True)
    print()

    final = aiml.generate(request)
    usage = final.get("usage") or {}
    cost = final.get("cost") or {}
    print(f"request_id={final['id']} tokens={usage.get('output_tokens', '?')} cost_micro={cost.get('total_micro_usd', '?')}")

The stream is a context manager on purpose. Leaving the block closes the socket, and closing the socket is what actually stops the generation upstream — an iterator you simply stop reading keeps the provider generating, and keeps you paying for it until the next checkpoint.

s.final() assembles the response from the events rather than asking the server again: the event sequence is a prefix-closed rendering of the response, so the two are the same answer and one of them costs nothing.

Async

Every method has an awaited twin on AsyncAIML, with the same name.

import asyncio
from aiml import AsyncAIML

async def main():
    async with AsyncAIML() as aiml:
        s = await aiml.stream(request)
        async with s:
            async for text in s.text_deltas():
                print(text, end="", flush=True)

asyncio.run(main())

Retries and idempotency

Every call carries an Idempotency-Key, generated per call unless you pass idempotency_key=, and every retry reuses it. That is what makes retrying a billed operation safe: the gateway bills the request once however many times it arrives.

Retries happen only before the first byte, and only for connection failures, 408, 409, 429 and 5xx. Nothing retries after the first byte: a half-delivered stream cannot be replayed without either double-billing or lying about what the model said. A stream that fails midway gives you the partial answer and an AIMLError.

AIML(max_retries=4, initial_delay=0.25)
aiml.generate(request, max_retries=0)   # per call

Errors carry the request id

from aiml import AIMLError

try:
    aiml.generate(request)
except AIMLError as e:
    print(e.status, e.code, e.request_id)

The id is on the response header even when the body is not JSON — a proxy's 502 page still tells you which request it was. Look it up on the request lookup page or with aiml.get_request(id).

Before you send

aiml.count_tokens(request)   # {"input_tokens": …, "source": "exact" | "estimated", "tokenizer": …}
aiml.dry_run(request)        # the plan, the capability warnings and the reserve

dry_run runs the real router over the real catalog and calls no provider.