hanzoai/python-sdk
2
0
Fork 0
mirror of https://github.com/hanzoai/python-sdk.git synced 2026-10-23 20:23:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(hanzoai): typed usage-limit errors and read_usage

Browse Source
This commit is contained in:
zeekay
2026-10-04 16:37:35 -07:00
parent 259f4d2fa4
commit 4240afb736
5 changed files with 724 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
LLM.md
+26
View File
@@ -416,6 +416,32 @@ runs them against `conformance/fake.py`, and `uv run python tests/live_conforman
against api.hanzo.ai and prints pass or fail per rule. Release: tag `hanzo-kai-v<version>`, and
`https://pypi.org/pypi/hanzo-kai/json` is the proof, not a green run.
## Plan usage — `hanzoai.usage`
`read_usage(headers)` → `Usage(usage, usage_class, paid_by, fallback, served, reason)` from
the `X-Hanzo-*` headers, case-insensitive; an absent header is `None`. `UsageLimitError`
(a `Fault`) with one subclass per AI-router refusal code — `PlanAllowanceUsedError`,
`PaidPlanRequiredError`, `FreePlanCapError`, `ModelCapError`, `UsageCapExceededError`,
`InsufficientBalanceError`, the JS and Go names. `refusal(reply)` reads only the router's
nested `{"error": {code, ...}}` body; the money gate's flat body stays `Denied`.
`Client.send` and `Client.response_deserialize` raise it, so every generated operation run
through `hanzoai.Client` does; a bare generated `ApiClient` still raises `ApiException`.
The rest is generated, typed, and needs nothing hand-written: `AiApi.ai_limits` /
`ai_set_limits`, `SyncApi.get_sync` / `get_sync_by_id` / `post_sync` /
`post_sync_by_id_run`, `AiApi.post_decisions`, `AiApi.get_models` (`ModelInfo.var_class`
is the wire's `class`; `pricing.variable`). `/v1/models` ignores `?class=`, so filter
client-side. Headers: the `*_with_http_info` variant's `.headers` into `read_usage`.
`ai.DecisionsChoice.criteria` is an object in cloud's document, so the `[label, ...]` form
the server also takes cannot be sent through the generated client.
**Hand-written root modules ship with the next cloud fanout.** The root version is
cloud's; the fanout's `v8.5.N` tag publishes whatever `pkg/hanzoai` holds at that commit.
Never cut a root version by hand.
**Commit with `LEFTHOOK=0`.** The global hook falls through to `uv run lefthook -h`, which
re-locks `uv.lock` with the local uv and syncs a `.venv` on every checkout and commit.
## Key entry points
- `pkg/hanzoai/` — the client (`ApiClient`, `Configuration`, `*Api`) under `cloud/`, plus
seven hand-written modules beside it (`config`, `mcp`, `protocols`, `session`, `zap`, …).
pkg/hanzoai/__init__.py
+35 -1
View File
@@ -73,10 +73,33 @@ _HOME = {
"Triple": "hanzoai.graph",
"Vocabulary": "hanzoai.graph",
"Source": "hanzoai.graph",
"Usage": "hanzoai.usage",
"read_usage": "hanzoai.usage",
"UsageLimitError": "hanzoai.usage",
"PlanAllowanceUsedError": "hanzoai.usage",
"PaidPlanRequiredError": "hanzoai.usage",
"FreePlanCapError": "hanzoai.usage",
"ModelCapError": "hanzoai.usage",
"UsageCapExceededError": "hanzoai.usage",
"InsufficientBalanceError": "hanzoai.usage",
}
#: The capability modules, reachable as `hanzoai.budget` and so on.
_MODULES = ("wire", "page", "answer", "token", "grant", "client", "budget", "policy", "audit", "search", "kb", "graph")
_MODULES = (
"wire",
"page",
"answer",
"token",
"grant",
"client",
"budget",
"policy",
"audit",
"search",
"kb",
"graph",
"usage",
)
__all__ = ["__version__", *_MODULES, *_HOME]
@@ -120,6 +143,17 @@ if TYPE_CHECKING: # what a type checker and an IDE see, without the import cost
Vocabulary as Vocabulary,
)
from hanzoai.token import Token as Token
from hanzoai.usage import (
Usage as Usage,
ModelCapError as ModelCapError,
UsageLimitError as UsageLimitError,
FreePlanCapError as FreePlanCapError,
PaidPlanRequiredError as PaidPlanRequiredError,
UsageCapExceededError as UsageCapExceededError,
PlanAllowanceUsedError as PlanAllowanceUsedError,
InsufficientBalanceError as InsufficientBalanceError,
read_usage as read_usage,
)
from hanzoai.answer import (
Ok as Ok,
Cure as Cure,
pkg/hanzoai/client.py
+16 -3
View File
@@ -27,8 +27,8 @@ from typing import Any, Dict, Optional
from email.utils import parsedate_to_datetime
from urllib.parse import urlencode
from urllib3.util.retry import Retry
from urllib3.exceptions import MaxRetryError, ResponseError
from urllib3.util.retry import Retry
from hanzoai import wire
from hanzoai.kb import Kb
@@ -36,6 +36,7 @@ from hanzoai.audit import Audit
from hanzoai.grant import Grant
from hanzoai.graph import Graph
from hanzoai.token import BASE, ISSUER, Token
from hanzoai.usage import refusal
from hanzoai.answer import Held, held, value
from hanzoai.budget import Budget
from hanzoai.policy import Policy
@@ -166,14 +167,19 @@ class Client(ApiClient):
member nobody set — a request carries what the caller determined, and a
JSON null is not that. A `datetime` is stamped RFC 3339. A 401 re-mints
once and replays, so a rotated token costs a round trip rather than an
error the caller has to handle.
error the caller has to handle. A priced call the plan refused raises
its :class:`hanzoai.usage.UsageLimitError`.
"""
headers = {"Accept": "application/json", "Authorization": "Bearer " + self.credential.token()}
if body is not None:
headers["Content-Type"] = media
response = self.call_api(method, self.base + path + _query(query), headers, _body(body))
response.read()
return _reply(response)
reply = _reply(response)
stop = refusal(reply)
if stop:
raise stop
return reply
def read(self, method: str, path: str, **call: Any) -> Any:
"""A call no gate refuses: the decoded body, or the reason there is none."""
@@ -215,10 +221,17 @@ class Client(ApiClient):
with `None` and a queued call reads as one that succeeded and returned
nothing. The six capabilities answer the same hold as
:class:`hanzoai.Held`, which is the arm this raises.
A priced call the plan refused raises its
:class:`hanzoai.usage.UsageLimitError` rather than the generated
``ApiException``, the same class :meth:`send` raises for it.
"""
reply = _reply(response_data)
if held(reply):
raise Held.read(reply.body, reply.request)
stop = refusal(reply)
if stop:
raise stop
return super().response_deserialize(response_data, response_types_map)
pkg/hanzoai/usage.py
+245
View File
@@ -0,0 +1,245 @@
"""Plan usage: where a priced call stands, and the refusals it can meet.
Every answer from a priced model carries the standing in headers::
r = AiApi(c).post_chat_completions_with_http_info(req)
u = read_usage(r.headers)
u.paid_by # "plan", "credits" or "free"; None when the answer named none
A refused call raises one class per refusal code, all under
:class:`UsageLimitError`, from :meth:`hanzoai.Client.send` and from every
generated operation the client runs::
try:
AiApi(c).post_decisions(req)
except FreePlanCapError as e:
e.resets_at, [a.url for a in e.actions]
The refusal states shares and ways on, never an amount: no field here is a
price, a count or a cap. The body is the AI router's, ``{"error": {message,
type, code, class, model, fallback, resets_at, upgrade_url, actions}}``; the
money gate's flat refusal is :class:`hanzoai.Denied`, and a body with neither
shape stays the fault it was. The class names are the ones the JavaScript and
Go SDKs use.
"""
from __future__ import annotations
from typing import Any, Dict, Type, Tuple, Optional
from datetime import datetime
from dataclasses import dataclass
from hanzoai.wire import Reply, instant
from hanzoai.answer import Fault
__all__ = [
"Usage",
"Action",
"read_usage",
"refusal",
"UsageLimitError",
"PlanAllowanceUsedError",
"PaidPlanRequiredError",
"FreePlanCapError",
"ModelCapError",
"UsageCapExceededError",
"InsufficientBalanceError",
]
@dataclass(frozen=True)
class Usage:
"""The standing one answer reported, read from its ``X-Hanzo-*`` headers.
A header the answer did not carry reads `None`. A free model's answer
carries only `served`, so its `paid_by` is unknown, not ``""``.
"""
#: ``X-Hanzo-Usage``: ok, near or limited, where the class stands.
usage: Optional[str] = None
#: ``X-Hanzo-Usage-Class``: premium, ours or free.
usage_class: Optional[str] = None
#: ``X-Hanzo-Paid-By``: plan, credits or free.
paid_by: Optional[str] = None
#: ``X-Hanzo-Fallback``: the model that answered instead of the one asked for.
fallback: Optional[str] = None
#: ``X-Hanzo-Served``: the model that answered.
served: Optional[str] = None
#: ``X-Hanzo-Usage-Reason``: the refusal code that sent the call to `fallback`.
reason: Optional[str] = None
#: Header name, lowercased, to the field it fills.
HEADERS = {
"x-hanzo-usage": "usage",
"x-hanzo-usage-class": "usage_class",
"x-hanzo-paid-by": "paid_by",
"x-hanzo-fallback": "fallback",
"x-hanzo-served": "served",
"x-hanzo-usage-reason": "reason",
}
def read_usage(headers: Any) -> Usage:
"""The :class:`Usage` in `headers`, matched without regard to case.
`headers` is any mapping (a dict, urllib3's or httpx's headers, a generated
``ApiResponse.headers``) or a sequence of name, value pairs. `None` reads
as no headers. An empty value is absent.
"""
pairs = headers.items() if hasattr(headers, "items") else (headers or ())
found: Dict[str, str] = {}
for name, v in pairs:
field = HEADERS.get(str(name).lower())
if field and isinstance(v, str) and v.strip():
found[field] = v.strip()
return Usage(**found)
@dataclass(frozen=True)
class Action:
"""One way on from a refusal.
`kind` is upgrade (`url`, `plan`), switch (`model`), credits (`url` is
``/v1/ai/limits``; a PUT with ``creditsAfterAllowance`` turns it on) or
topup (`url` is the pay page). A field the action did not carry is `None`.
"""
kind: str
label: Optional[str] = None
url: Optional[str] = None
plan: Optional[str] = None
model: Optional[str] = None
@classmethod
def read(cls, body: Any) -> "Action":
return cls(kind=_text(body, "kind") or "", **{k: _text(body, k) for k in ("label", "url", "plan", "model")})
class UsageLimitError(Fault):
"""A priced call the plan, the wallet or a cap refused, and the ways on.
`status` is the HTTP status and `code` the refusal. `usage_class` is the
model class it concerns (the body's ``class``); `model` and `fallback` name
the model refused and the one that would answer instead; `resets_at` is
when the refusal lifts; `actions` are the ways on, in the order the server
gave them. It is a :class:`hanzoai.Fault`, so it also carries `request`
and `retry_after`, and ``except Fault`` still catches it.
"""
#: The refusal code a subclass answers; the base class names none.
CODE = ""
def __init__(
self,
status: int,
code: str,
message: str,
*,
type: Optional[str] = None,
usage_class: Optional[str] = None,
model: Optional[str] = None,
fallback: Optional[str] = None,
resets_at: Optional[datetime] = None,
upgrade_url: Optional[str] = None,
actions: Tuple[Action, ...] = (),
request: str = "",
retry_after: float = 0,
) -> None:
super().__init__(status, code, message, request, retry_after)
self.message = message
self.type = type
self.usage_class = usage_class
self.model = model
self.fallback = fallback
self.resets_at = resets_at
self.upgrade_url = upgrade_url
self.actions = actions
class PlanAllowanceUsedError(UsageLimitError):
"""402: the plan's included usage of the class is used and nothing else may pay."""
CODE = "plan_allowance_used"
class PaidPlanRequiredError(UsageLimitError):
"""402: the model needs a paid plan or prepaid balance."""
CODE = "paid_plan_required"
class FreePlanCapError(UsageLimitError):
"""429: the free plan's daily share of the model is used; `resets_at` says when it returns."""
CODE = "free_plan_cap"
class ModelCapError(UsageLimitError):
"""402: the model used its share of the plan; `fallback` is the model that answers instead."""
CODE = "model_cap"
class UsageCapExceededError(UsageLimitError):
"""429: a session or day request window is spent."""
CODE = "usage_cap_exceeded"
class InsufficientBalanceError(UsageLimitError):
"""402: the wallet's known balance cannot cover the call."""
CODE = "insufficient_balance"
#: Refusal code to the class raised for it.
CODES: Dict[str, Type[UsageLimitError]] = {
c.CODE: c
for c in (
PlanAllowanceUsedError,
PaidPlanRequiredError,
FreePlanCapError,
ModelCapError,
UsageCapExceededError,
InsufficientBalanceError,
)
}
def refusal(reply: Reply) -> Optional[UsageLimitError]:
"""The :class:`UsageLimitError` `reply` carries, or `None` when it carries none.
Only a non-2xx whose body is ``{"error": {"code": ...}}`` with one of the
six codes is one; every other answer is left to the rule that already reads
it.
"""
if 200 <= reply.status <= 299 or not isinstance(reply.body, dict):
return None
e = reply.body.get("error")
if not isinstance(e, dict):
return None
cls = CODES.get(_text(e, "code") or "")
if cls is None:
return None
acts = e.get("actions")
return cls(
reply.status,
cls.CODE,
_text(e, "message") or "",
type=_text(e, "type"),
usage_class=_text(e, "class"),
model=_text(e, "model"),
fallback=_text(e, "fallback"),
resets_at=instant(e.get("resets_at")),
upgrade_url=_text(e, "upgrade_url"),
actions=tuple(Action.read(a) for a in acts if isinstance(a, dict)) if isinstance(acts, list) else (),
request=reply.request,
retry_after=reply.retry_after,
)
def _text(body: Any, key: str) -> Optional[str]:
"""The non-empty string at `key`, or `None`."""
v = body.get(key) if isinstance(body, dict) else None
return v if isinstance(v, str) and v else None
tests/test_usage.py
+402
View File
@@ -0,0 +1,402 @@
"""Plan usage: the refusal classes, the header reader, and where the client raises them.
The bodies are the AI router's own, as api.hanzo.ai answers them. The client
tests answer over a real socket, so the refusal is raised from the same code
path a caller runs: :meth:`Client.send` and a generated operation.
"""
import json
import threading
from datetime import datetime, timezone
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
import pytest
from urllib3 import HTTPHeaderDict
from hanzoai import (
Fault,
Usage,
ModelCapError,
UsageLimitError,
FreePlanCapError,
PaidPlanRequiredError,
UsageCapExceededError,
PlanAllowanceUsedError,
InsufficientBalanceError,
read_usage,
)
from hanzoai.wire import Reply
from hanzoai.usage import Action, refusal
#: Measured: POST /v1/decisions, model kai, org on the free plan.
FREE_PLAN_CAP = {
"error": {
"message": "Free plan: today's Kai requests are used. Upgrade for more: https://hanzo.ai/pay",
"type": "rate_limit_error",
"code": "free_plan_cap",
"class": "ours",
"resets_at": "2026-10-05T00:00:00Z",
"upgrade_url": "https://hanzo.ai/pay/cart?plan=dev",
"actions": [
{
"kind": "upgrade",
"label": "Upgrade your plan",
"url": "https://hanzo.ai/pay/cart?plan=dev",
"plan": "dev",
},
{"kind": "topup", "label": "Add prepaid credit", "url": "https://hanzo.ai/pay"},
],
}
}
MODEL_CAP = {
"error": {
"message": "This model has used its share of your plan for now. Enso answers until it resets.",
"type": "billing_error",
"code": "model_cap",
"class": "premium",
"model": "anthropic/claude-opus-4.1",
"fallback": "enso",
"resets_at": "2026-10-05T05:00:00Z",
"actions": [
{"kind": "switch", "label": "Try Enso", "model": "enso"},
{
"kind": "upgrade",
"label": "Upgrade to Max",
"url": "https://hanzo.ai/pay/cart?plan=max-5x",
"plan": "max-5x",
},
],
}
}
PLAN_ALLOWANCE_USED = {
"error": {
"message": "Your plan's included premium usage is used for this period.",
"type": "billing_error",
"code": "plan_allowance_used",
"class": "premium",
"resets_at": "2026-11-01T00:00:00Z",
"actions": [
{"kind": "credits", "label": "Continue with credits", "url": "/v1/ai/limits"},
{"kind": "topup", "label": "Add prepaid credit", "url": "https://hanzo.ai/pay"},
],
}
}
PAID_PLAN_REQUIRED = {
"error": {
"message": "This model needs a paid plan or prepaid credit.",
"type": "billing_error",
"code": "paid_plan_required",
"class": "premium",
"model": "anthropic/claude-sonnet-4.5",
"upgrade_url": "https://hanzo.ai/pay/cart?plan=dev",
"actions": [
{"kind": "upgrade", "label": "Upgrade to Pro", "url": "https://hanzo.ai/pay/cart?plan=dev", "plan": "dev"}
],
}
}
USAGE_CAP_EXCEEDED = {
"error": {
"message": "This session's requests are spent. It resets shortly.",
"type": "rate_limit_error",
"code": "usage_cap_exceeded",
"resets_at": "2026-10-04T23:30:00Z",
}
}
#: The wallet refusal carries only the three members.
INSUFFICIENT_BALANCE = {
"error": {
"message": "Insufficient balance for this request.",
"type": "billing_error",
"code": "insufficient_balance",
}
}
def reply(status, body, request="req_9", retry_after=0):
return Reply(status=status, body=body, request=request, retry_after=retry_after)
# --------------------------------------------------------------------------
# One class per code
# --------------------------------------------------------------------------
@pytest.mark.parametrize(
"status, body, cls",
[
(402, PLAN_ALLOWANCE_USED, PlanAllowanceUsedError),
(402, PAID_PLAN_REQUIRED, PaidPlanRequiredError),
(429, FREE_PLAN_CAP, FreePlanCapError),
(402, MODEL_CAP, ModelCapError),
(429, USAGE_CAP_EXCEEDED, UsageCapExceededError),
(402, INSUFFICIENT_BALANCE, InsufficientBalanceError),
],
)
def test_each_code_raises_its_own_class_under_one_base(status, body, cls):
e = refusal(reply(status, body))
assert type(e) is cls
assert isinstance(e, UsageLimitError) and isinstance(e, Fault)
assert (e.status, e.code, e.message, e.type) == (
status,
body["error"]["code"],
body["error"]["message"],
body["error"]["type"],
)
assert e.code == cls.CODE
assert e.request == "req_9"
def test_free_plan_cap_reads_every_field_the_router_sends():
e = refusal(reply(429, FREE_PLAN_CAP, retry_after=3032))
assert e.usage_class == "ours"
assert e.resets_at == datetime(2026, 10, 5, tzinfo=timezone.utc)
assert e.upgrade_url == "https://hanzo.ai/pay/cart?plan=dev"
assert e.model is None and e.fallback is None
assert e.retry_after == 3032
assert e.actions == (
Action(kind="upgrade", label="Upgrade your plan", url="https://hanzo.ai/pay/cart?plan=dev", plan="dev"),
Action(kind="topup", label="Add prepaid credit", url="https://hanzo.ai/pay"),
)
assert "429 free_plan_cap" in str(e) and "today's Kai requests are used" in str(e)
def test_model_cap_names_the_model_its_fallback_and_the_switch():
e = refusal(reply(402, MODEL_CAP))
assert (e.model, e.fallback, e.usage_class) == ("anthropic/claude-opus-4.1", "enso", "premium")
assert e.actions[0] == Action(kind="switch", label="Try Enso", model="enso")
def test_a_field_the_refusal_did_not_carry_is_none_not_empty():
e = refusal(reply(402, INSUFFICIENT_BALANCE))
assert (e.usage_class, e.model, e.fallback, e.resets_at, e.upgrade_url) == (None, None, None, None, None)
assert e.actions == ()
@pytest.mark.parametrize(
"status, body",
[
# The money gate's flat refusal is Denied's, not this family's.
(402, {"type": "about:blank", "status": 402, "detail": "the wallet is empty", "code": "insufficient_balance"}),
(402, {"error": "payment_required", "product": "search", "message": "top up"}),
# The retryable sibling and the free lane are not usage-limit refusals.
(503, {"error": {"code": "balance_unavailable", "message": "try again"}}),
(429, {"error": {"code": "pool_busy", "message": "busy"}}),
(500, "upstream exploded"),
(402, None),
# A success never is one, whatever it carries.
(200, FREE_PLAN_CAP),
],
)
def test_anything_else_is_left_to_the_rule_that_reads_it(status, body):
assert refusal(reply(status, body)) is None
# --------------------------------------------------------------------------
# The headers
# --------------------------------------------------------------------------
def test_read_usage_matches_without_regard_to_case():
u = read_usage(
{
"X-Hanzo-Usage": "ok",
"x-hanzo-usage-class": "premium",
"X-HANZO-PAID-BY": "credits",
"x-Hanzo-Served": "anthropic/claude-sonnet-4.5",
"Content-Type": "application/json",
}
)
assert u == Usage(usage="ok", usage_class="premium", paid_by="credits", served="anthropic/claude-sonnet-4.5")
def test_an_absent_header_reads_none_never_empty():
u = read_usage({"x-hanzo-served": "enso", "x-hanzo-paid-by": "", "x-hanzo-fallback": " "})
assert u.served == "enso"
assert (u.usage, u.usage_class, u.paid_by, u.fallback, u.reason) == (None, None, None, None, None)
assert read_usage(None) == Usage()
assert read_usage({}) == Usage()
def test_read_usage_takes_pairs_and_every_header_mapping():
pairs = [("X-Hanzo-Fallback", "enso"), ("X-Hanzo-Usage-Reason", "model_cap"), ("X-Hanzo-Usage", "limited")]
assert read_usage(pairs) == Usage(usage="limited", fallback="enso", reason="model_cap")
assert read_usage(HTTPHeaderDict(pairs)) == read_usage(pairs)
# --------------------------------------------------------------------------
# Where the client raises them
# --------------------------------------------------------------------------
class Minted:
"""A credential that already holds its token, so no call goes to IAM."""
def token(self):
return "tok"
def invalidate(self):
pass
@pytest.fixture
def server():
"""A server on a free port: the base URL and the staged answers."""
staged = []
class Handler(BaseHTTPRequestHandler):
def answer(self):
n = int(self.headers.get("Content-Length") or 0)
if n:
self.rfile.read(n)
status, headers, body = staged.pop(0)
data = json.dumps(body).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(data)))
for name, v in headers.items():
self.send_header(name, v)
self.end_headers()
self.wfile.write(data)
do_GET = do_POST = answer
def log_message(self, *args):
pass
httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=httpd.serve_forever, daemon=True).start()
try:
yield "http://127.0.0.1:{0}".format(httpd.server_address[1]), staged
finally:
httpd.shutdown()
httpd.server_close()
def test_send_raises_the_refusal_with_the_call_s_request_id(server):
from hanzoai import Client
base, staged = server
staged.append((429, {"x-request-id": "req_kai", "Retry-After": "3032"}, FREE_PLAN_CAP))
c = Client(base=base, credential=Minted())
with pytest.raises(FreePlanCapError) as raised:
c.send("POST", "/v1/decisions", body={"model": "kai", "state": "s", "questions": {}})
assert (raised.value.request, raised.value.retry_after, raised.value.usage_class) == ("req_kai", 3032, "ours")
def test_a_generated_operation_raises_the_same_class(server):
from hanzoai import Client
from hanzoai.cloud import AiApi
base, staged = server
staged.append((402, {}, PLAN_ALLOWANCE_USED))
with pytest.raises(PlanAllowanceUsedError) as raised:
AiApi(Client(base=base, credential=Minted())).ai_limits()
assert raised.value.actions[0].url == "/v1/ai/limits"
def test_a_generated_answer_s_headers_read_as_usage(server):
from hanzoai import Client
from hanzoai.cloud import AiApi
base, staged = server
headers = {"X-Hanzo-Usage": "ok", "X-Hanzo-Usage-Class": "premium", "X-Hanzo-Paid-By": "credits"}
staged.append((200, headers, {"plan": "dev", "state": "ok"}))
r = AiApi(Client(base=base, credential=Minted())).ai_limits_with_http_info()
assert r.data.plan == "dev"
assert read_usage(r.headers) == Usage(usage="ok", usage_class="premium", paid_by="credits")
# --------------------------------------------------------------------------
# Live, against api.hanzo.ai, as the signed-in `hanzo` CLI user
# --------------------------------------------------------------------------
class Bearer:
"""The `hanzo` CLI's live IAM session token, asked for once."""
def __init__(self):
import subprocess
try:
out = subprocess.run(["hanzo", "auth", "token"], capture_output=True, text=True, timeout=30)
except (OSError, subprocess.SubprocessError):
out = None
self.value = out.stdout.strip() if out and out.returncode == 0 else ""
def token(self):
return self.value
def invalidate(self):
pass
@pytest.fixture(scope="module")
def live():
from hanzoai import Client
bearer = Bearer()
if not bearer.value:
pytest.skip("no `hanzo auth token`: sign in with `hanzo auth login`")
return Client(base="https://api.hanzo.ai", credential=bearer)
@pytest.mark.integration
def test_live_limits_models_and_a_premium_chat_paid_by_credits(live):
from hanzoai.cloud import AiApi
from hanzoai.cloud.models import OpenaiChatCompletionRequest
ai = AiApi(live)
limits = ai.ai_limits()
assert limits.state in ("ok", "near", "limited")
assert limits.plan
rows = ai.get_models().data or []
ours = [m for m in rows if m.var_class == "ours"]
assert {"kai"} <= {m.id for m in ours}
assert all(m.var_class in ("premium", "ours", "free") for m in rows)
assert any(m.pricing and m.pricing.variable for m in rows)
r = ai.post_chat_completions_with_http_info(
OpenaiChatCompletionRequest.from_dict(
{
"model": "anthropic/claude-sonnet-4.5",
"messages": [{"role": "user", "content": "Say ok"}],
"max_tokens": 5,
}
)
)
u = read_usage(r.headers)
assert (u.usage_class, u.paid_by, u.served) == ("premium", "credits", "anthropic/claude-sonnet-4.5")
assert u.usage in ("ok", "near")
@pytest.mark.integration
def test_live_a_decision_answers_or_raises_its_typed_refusal(live):
from hanzoai.cloud import AiApi
from hanzoai.cloud.models import AiDecisionsRequest
req = AiDecisionsRequest.from_dict(
{
"model": "kai",
"state": "refund for a double charge",
"questions": {
"team": {
"type": "choice",
"criteria": {"payments": "money moved wrong", "account": "sign-in or profile"},
}
},
}
)
try:
d = AiApi(live).post_decisions(req)
except UsageLimitError as e:
# The free plan's daily Kai share, used: the refusal arrives typed.
assert isinstance(e, FreePlanCapError) and e.usage_class == "ours" and e.resets_at is not None
else:
assert "team" in d.answers