MeasyAI Python SDK

Python client for the MeasyAI API.

pip install measyai

Requires Python 3.9+. One dependency: httpx.

Chat

from measyai import MeasyAI

client = MeasyAI()  # reads MEASYAI_API_KEY

completion = client.chat.create(
    model="measyai/meridian",
    messages=[{"role": "user", "content": "Explain HTTP/3 briefly."}],
)
print(completion.text)

completion.text is the first choice's content. The full shape is there when you need it — completion.choices[0].finish_reason, completion.usage.total_tokens.

Streaming

stream() opens the connection before it returns, so a rejected request raises there rather than on the first frame. Iterating yields chunks; text() yields just the deltas.

with client.chat.stream(
    messages=[{"role": "user", "content": "Why is Postgres MVCC useful?"}],
) as stream:
    for delta in stream.text():
        print(delta, end="", flush=True)

Use it as a context manager. Iterating to the end releases the connection on its own, but breaking out early only does so once the iterator is collected — with makes that immediate.

For finish reasons or the closing usage block, iterate the chunks:

with client.chat.stream(messages=messages) as stream:
    for chunk in stream:
        if chunk.usage:
            print(f"\n{chunk.usage.total_tokens} tokens")

Async

AsyncMeasyAI has the same surface with await on the calls and async for on the streams.

import asyncio
from measyai import AsyncMeasyAI

async def main() -> None:
    async with AsyncMeasyAI() as client:
        stream = await client.chat.stream(
            messages=[{"role": "user", "content": "Explain QUIC."}],
        )
        async with stream:
            async for delta in stream.text():
                print(delta, end="", flush=True)

asyncio.run(main())

Unmodelled parameters

Anything this package does not model is passed as a keyword argument and forwarded to the provider unchanged:

client.chat.create(
    messages=messages,
    temperature=0.2,
    tools=my_tools,
    response_format={"type": "json_object"},
)

model, messages and stream are owned by the method signature and are refused rather than silently dropped — otherwise stream=True would hand create() an SSE body it then tries to decode as JSON.

Response fields this package does not model are kept too: every object carries the payload it was built from in .raw.

Batches

from measyai import BatchRequest

batch = client.batches.create([
    BatchRequest(custom_id="row-1", body={"messages": messages}),
])

done = client.batches.wait(batch.id, interval=5)
print(f"{done.completed} completed, {done.failed} failed")

for item in done.items:
    if item.response:
        print(item.custom_id, item.response.text)

wait() polls until the batch is terminal. It waits indefinitely by default — a large batch legitimately runs for hours — so pass timeout= if you want it to give up.

Webhooks

Verify against the raw body. Re-encoding a payload you already decoded will not reproduce the same bytes, and the check will fail on a delivery that was perfectly valid.

import measyai

@app.post("/webhooks/measyai")
async def receive(request: Request):
    raw = await request.body()

    if not measyai.verify_webhook(
        secret,
        request.headers[measyai.HEADER_SIGNATURE],
        request.headers[measyai.HEADER_TIMESTAMP],
        raw,
    ):
        raise HTTPException(400)
    ...

The timestamp is signed inside the message and checked against a tolerance, so a captured delivery cannot be replayed. The comparison is constant-time.

Errors

Everything raised inherits from MeasyAIError. Below it the tree forks on why the call failed: APIError when the API rejected the request, APIConnectionError when there was never a response.

from measyai import APIError, MeasyAIError

try:
    completion = client.chat.create(messages=messages)
except APIError as err:
    if err.retryable:  # 429 or 5xx — worth backing off
        ...
    print(err.code, err.request_id)
except MeasyAIError:
    ...  # connection, timeout, malformed body

APIError also subclasses by status — AuthenticationError, RateLimitError, NotFoundError and the rest — so you can catch just the case you handle. Branch on err.code, which is stable; err.message is written for humans and may be reworded.

Nothing is retried for you. retryable tells you when a retry could succeed; the backoff policy is yours.

Configuration

client = MeasyAI(
    api_key="msy_...",              # or MEASYAI_API_KEY
    base_url="http://localhost:8080",  # or MEASYAI_BASE_URL
    user_agent="acme-crm/3.1",      # appended, for server logs
    timeout=httpx.Timeout(connect=10.0, read=60.0),
)

The default timeouts bound stalls, not total duration — a streamed generation legitimately runs for minutes, and an overall deadline would cut it off mid-answer.

Pass http_client= to supply your own httpx.Client for a proxy, a custom transport or a test double. You keep ownership of it: closing the MeasyAI client leaves yours open.

Reuse one client for the life of the process. It is safe to share between threads, and a client per call throws away the connection pool.

Using an OpenAI client instead

The /v1 surface is OpenAI-compatible, so the official OpenAI SDK works by changing its base URL:

from openai import OpenAI

client = OpenAI(api_key="msy_...", base_url="https://api.measyai.com/v1")

Use this package when you also want batches and webhook verification typed.

Links

• Documentation
• Status
• Go SDK
• TypeScript SDK

MIT licensed.