wenil/clawdmeter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

v3 M3: z.ai (GLM) provider — Anthropic-compatible poll from a pasted API key

Browse Source 
ZaiProvider replaces the zai stub. It reaches z.ai's Anthropic-compatible
endpoint (default https://api.z.ai/api/anthropic) exactly like Claude Code with
ANTHROPIC_BASE_URL pointed at z.ai: a /v1/messages POST with a Bearer key. The
key + base URL come from the config the control panel's z.ai field writes
(providers.zai.base_url/.api_key) — no OAuth, no local files.

Two z.ai-specific cares: (1) static API key, no self-refresh; (2) the GLM Coding
Plan meters by prompts, so a 60s poll would drain quota — the provider
self-throttles to one network call per poll_interval_s (default 15 min) and
serves a cached status in between. Rate-limit parsing assumes z.ai proxies
Anthropic's unified headers (5h->s/sr, 7d->w/wr); if a real response lacks them
it still reports connected and logs the limit-ish headers it did return, so the
mapping can be finalized against a live key.

tools/probe_zai.py dumps a real z.ai response's headers (key via env, never
printed) to pin the exact header names. Graceful without a key: selecting z.ai
shows the blue theme + "no key", never crashes the loop. Registry: all three
providers now real; unknown ids fall back to StubProvider. +7 z.ai tests (mocked
transport: header mapping, auth, throttle, fallback); 120 passed, 2 Linux-only.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wenil
2026-07-10 07:27:34 +03:00
co-authored by Claude Opus 4.8
parent b63a6a62ae
commit ac0742d92c
6 changed files with 347 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
daemon/providers/__init__.py
+6 -9
View File
@@ -10,12 +10,7 @@ from __future__ import annotations
from .base import Provider, ProviderStatus, StubProvider
from .anthropic import AnthropicProvider
from .openai_codex import OpenAICodexProvider
# Brand accents / labels for the not-yet-implemented providers, so the stub still
# carries the right theme hint to the watch once firmware theming lands.
_STUB_META = {
"zai": ("z.ai GLM", "3859ff"),
}
from .zai import ZaiProvider
def get_provider(pid: str) -> Provider:
@@ -23,11 +18,13 @@ def get_provider(pid: str) -> Provider:
return AnthropicProvider()
if pid == "openai":
return OpenAICodexProvider()
label, accent = _STUB_META.get(pid, (pid, "d97757"))
return StubProvider(pid, label=label, accent=accent)
if pid == "zai":
return ZaiProvider()
# Unknown id (a config from a newer version, say) — a safe placeholder.
return StubProvider(pid, label=pid, accent="d97757")
__all__ = [
"Provider", "ProviderStatus", "StubProvider",
"AnthropicProvider", "OpenAICodexProvider", "get_provider",
"AnthropicProvider", "OpenAICodexProvider", "ZaiProvider", "get_provider",
]
daemon/providers/zai.py
+135
View File
@@ -0,0 +1,135 @@
"""z.ai (GLM Coding Plan) provider (v3 M3).
z.ai exposes an Anthropic-compatible endpoint (default
``https://api.z.ai/api/anthropic``), so we reach it the same way Claude Code does
with ``ANTHROPIC_BASE_URL`` pointed at z.ai: a ``/v1/messages`` POST with a Bearer
API key. The key + base URL come from the config the control panel writes
(``providers.zai.base_url`` / ``.api_key``) — the "paste your key" field.
Two z.ai-specific cares vs. Anthropic:
* Auth is a static API key (Bearer), not the OAuth token — no self-refresh.
* The GLM Coding Plan meters by *prompts*, so polling every 60s would burn the
user's quota. We self-throttle: hit the network at most once per
``poll_interval_s`` (default 15 min) and serve a cached status in between.
Rate-limit parsing assumes z.ai proxies Anthropic's unified rate-limit headers.
If a real z.ai response doesn't carry them, poll() still reports "connected" and
logs the limit-ish headers it DID return, so the mapping can be finalized against
a live key (see tools/probe_zai.py).
"""
from __future__ import annotations
import time
import httpx
from .base import Provider, ProviderStatus
DEFAULT_BASE = "https://api.z.ai/api/anthropic"
DEFAULT_MODEL = "glm-4.6"
DEFAULT_POLL_INTERVAL_S = 900 # protect the prompt-metered plan: >= 15 min between calls
class ZaiProvider(Provider):
id = "zai"
label = "z.ai GLM"
accent = "3859ff" # brand blue
def __init__(self, transport: httpx.BaseTransport | None = None) -> None:
self._transport = transport # tests inject an httpx.MockTransport
self._cache: ProviderStatus | None = None
self._cache_at = 0.0
async def poll(self) -> ProviderStatus:
from daemon.claude_usage_daemon_windows import log
try:
from daemon.config import load_config, provider_conf
except ImportError:
from config import load_config, provider_conf
conf = provider_conf(load_config(), "zai")
key = (conf.get("api_key") or "").strip()
base = (conf.get("base_url") or "").strip().rstrip("/") or DEFAULT_BASE
model = (conf.get("model") or "").strip() or DEFAULT_MODEL
try:
interval = float(conf.get("poll_interval_s") or DEFAULT_POLL_INTERVAL_S)
except (TypeError, ValueError):
interval = DEFAULT_POLL_INTERVAL_S
if not key:
return ProviderStatus(ok=False, st="no key", auth_problem=True)
# Self-throttle: reuse the cached reading until the interval elapses, so the
# prompt-metered plan isn't drained by the 60s daemon poll.
now = time.time()
if self._cache is not None and (now - self._cache_at) < interval:
return self._cache
st = await self._fetch(base, key, model, log)
# Cache stable outcomes (fresh data or a definitive auth failure); leave a
# transient blip uncached so it retries on the next tick.
if st.ok or st.auth_problem:
self._cache, self._cache_at = st, now
return st
async def _fetch(self, base: str, key: str, model: str, log) -> ProviderStatus:
url = f"{base}/v1/messages"
headers = {
"Authorization": f"Bearer {key}",
"anthropic-version": "2023-06-01",
"Content-Type": "application/json",
}
body = {"model": model, "max_tokens": 1, "messages": [{"role": "user", "content": "hi"}]}
client_kw = {"timeout": 20.0}
if self._transport is not None:
client_kw["transport"] = self._transport
try:
async with httpx.AsyncClient(**client_kw) as http:
resp = await http.post(url, headers=headers, json=body)
except httpx.HTTPError as e:
log(f"z.ai call failed: {e}") # transient — no auth flag
return ProviderStatus(ok=False, st="offline")
if resp.status_code in (401, 403):
log(f"z.ai auth rejected: HTTP {resp.status_code}")
return ProviderStatus(ok=False, st="bad key", auth_problem=True)
if resp.status_code >= 400:
log(f"z.ai HTTP {resp.status_code}: {resp.text[:150]}")
return ProviderStatus(ok=False, st=f"http {resp.status_code}")
return self._parse(resp, log)
@staticmethod
def _parse(resp: httpx.Response, log) -> ProviderStatus:
h = resp.headers
now = time.time()
def pct(util: str) -> float:
# Anthropic reports utilization as a 0-1 fraction.
try:
return round(float(util) * 100.0, 1)
except (TypeError, ValueError):
return 0.0
def reset_min(ts: str) -> int:
try:
m = (float(ts) - now) / 60.0
except (TypeError, ValueError):
return -1
return int(round(m)) if m > 0 else -1
# Assume z.ai proxies Anthropic's unified rate-limit headers.
if h.get("anthropic-ratelimit-unified-5h-utilization") is not None:
return ProviderStatus(
s=pct(h.get("anthropic-ratelimit-unified-5h-utilization")),
sr=reset_min(h.get("anthropic-ratelimit-unified-5h-reset")),
w=pct(h.get("anthropic-ratelimit-unified-7d-utilization")),
wr=reset_min(h.get("anthropic-ratelimit-unified-7d-reset")),
st=h.get("anthropic-ratelimit-unified-5h-status", "allowed"),
ok=True,
)
# Connected but no known limit headers — surface what z.ai DID return so the
# mapping can be finalized against a live response (tools/probe_zai.py).
limitish = sorted(k for k in h.keys()
if any(w in k.lower() for w in ("limit", "rate", "quota", "usage")))
log(f"z.ai: no unified rate-limit headers; limit-ish headers = {limitish or '<none>'}")
return ProviderStatus(ok=True, st="allowed")