skynet/small-scripts
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

agent-subscriptions: use /api/oauth/usage and profile for Anthropic usage and plan info

Browse Source 
Response rate-limit headers are no longer returned. Read the access token
from the Claude Code profile credentials (user:profile scope), report
5h/7d usage and resets, and add plan info with an inferred monthly renewal.

Claude-Session: https://claude.ai/code/session_01Gz3N3q53Bx9pbjB9tZT9dt
This commit is contained in:
Paul O'Reilly
2026-10-14 16:41:22 +13:00
parent 6c69456261
commit 8935d58bd2
2 changed files with 90 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

153
scripts/agent-subscriptions
View File

@@ -10,18 +10,20 @@ import urllib.request
from datetime import datetime, timezone from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
ANTHROPIC_TOKEN_PATH = Path("~/dev/claude/secrets/anthropic/api_key").expanduser() # Interactive-login credentials (carry the user:profile scope that the usage
# endpoint needs; the setup-token key in secrets/ does not). Read-only: never refreshed here.
ANTHROPIC_PROFILE_DIR = Path(
os.environ.get("AGENT_SUBSCRIPTIONS_CLAUDE_DIR", "~/.claude-oreillyit-anthropic")
).expanduser()
ANTHROPIC_TOKEN_PATH = ANTHROPIC_PROFILE_DIR / ".credentials.json"
ANTHROPIC_API = "https://api.anthropic.com/api/oauth"
MINIMAX_SOPS_PATH = Path( MINIMAX_SOPS_PATH = Path(
"~/dev/claude/projects/agent-runtime-secrets/providers/minimax/v1/provider.sops.env" "~/dev/claude/projects/agent-runtime-secrets/providers/minimax/v1/provider.sops.env"
).expanduser() ).expanduser()
MINIMAX_SOPS_KEY = Path("~/dev/claude/secrets/sops/provider-age-key.txt").expanduser() MINIMAX_SOPS_KEY = Path("~/dev/claude/secrets/sops/provider-age-key.txt").expanduser()
MINIMAX_DOTENV_KEY = "ANTHROPIC_AUTH_TOKEN" MINIMAX_DOTENV_KEY = "ANTHROPIC_AUTH_TOKEN"
# Reset field names. Anthropic emits Unix epoch seconds; MiniMax emits milliseconds. # Reset field names. MiniMax emits milliseconds (Anthropic emits ISO 8601 strings).
ANTHROPIC_RESET_HEADERS = {
"five_hour": "anthropic-ratelimit-unified-5h-reset",
"seven_day": "anthropic-ratelimit-unified-7d-reset",
}
MINIMAX_RESET_FIELDS = { MINIMAX_RESET_FIELDS = {
"five_hour": "end_time", # ms since epoch "five_hour": "end_time", # ms since epoch
"seven_day": "weekly_end_time", # ms since epoch "seven_day": "weekly_end_time", # ms since epoch
@@ -78,7 +80,11 @@ def _parse_dotenv_key(content: str, key: str) -> str:
def _read_anthropic_token() -> str: def _read_anthropic_token() -> str:
return ANTHROPIC_TOKEN_PATH.read_text().strip() """Access token from the Claude Code profile; raises if missing or expired."""
oauth = json.loads(ANTHROPIC_TOKEN_PATH.read_text())["claudeAiOauth"]
if oauth.get("expiresAt", 0) / 1000 < datetime.now(timezone.utc).timestamp():
raise RuntimeError(f"access token expired; run claude once with CLAUDE_CONFIG_DIR={ANTHROPIC_PROFILE_DIR}")
return oauth["accessToken"]
def _read_minimax_key() -> str: def _read_minimax_key() -> str:
@@ -90,75 +96,61 @@ def _read_minimax_key() -> str:
return _parse_dotenv_key(result.stdout, MINIMAX_DOTENV_KEY) return _parse_dotenv_key(result.stdout, MINIMAX_DOTENV_KEY)
def _parse_anthropic_headers(headers) -> tuple[dict[str, float], dict[str, int]]: def _anthropic_get(path: str, token: str) -> dict:
"""Returns (utilization_pct_by_window, reset_epoch_seconds_by_window). req = urllib.request.Request(
f"{ANTHROPIC_API}/{path}",
headers={
"Authorization": f"Bearer {token}",
"anthropic-beta": "oauth-2025-04-20",
"User-Agent": "claude-code/2.1",
},
)
with urllib.request.urlopen(req, timeout=15) as resp:
return json.load(resp)
Reset values are parsed as integer Unix epoch seconds; absent or malformed
headers yield no entry for that window. def _next_monthly(anchor: datetime, now: datetime) -> datetime:
"""Next monthly anniversary of anchor strictly after now (day clamped to 28)."""
day = min(anchor.day, 28)
y, m = now.year, now.month
cand = anchor.replace(year=y, month=m, day=day)
if cand <= now:
y, m = (y + 1, 1) if m == 12 else (y, m + 1)
cand = anchor.replace(year=y, month=m, day=day)
return cand
def probe_anthropic(token: str) -> tuple[dict[str, float], dict[str, int], dict]:
"""Returns (utilization_pct, reset_epoch_seconds, subscription_info).
Raises on HTTP/network failure. The profile call is best-effort: a failure
there yields an empty subscription dict.
""" """
usage = _anthropic_get("usage", token)
util: dict[str, float] = {} util: dict[str, float] = {}
reset: dict[str, int] = {} reset: dict[str, int] = {}
fh_u = headers.get("anthropic-ratelimit-unified-5h-utilization") for key in ("five_hour", "seven_day"):
sd_u = headers.get("anthropic-ratelimit-unified-7d-utilization") w = usage.get(key) or {}
for key, raw in [("five_hour", fh_u), ("seven_day", sd_u)]: if w.get("utilization") is not None:
if raw is not None: util[key] = float(w["utilization"])
try: if w.get("resets_at"):
util[key] = float(raw) * 100.0 reset[key] = int(datetime.fromisoformat(w["resets_at"]).timestamp())
except ValueError:
pass
for key, hdr in ANTHROPIC_RESET_HEADERS.items():
raw = headers.get(hdr)
if raw is None:
continue
try:
reset[key] = int(raw)
except ValueError:
pass
return util, reset
def probe_anthropic(token: str) -> tuple[dict[str, float], dict[str, int]] | None:
"""Returns (utilization_pct, reset_epoch_seconds) or None on total failure.
Either dict may be empty; both are populated independently.
"""
auth = {"Authorization": f"Bearer {token}", "anthropic-version": "2023-06-01"}
sub: dict = {}
try: try:
req = urllib.request.Request("https://api.anthropic.com/v1/models", headers=auth) org = _anthropic_get("profile", token).get("organization", {})
with urllib.request.urlopen(req, timeout=10) as resp: sub = {
util, reset = _parse_anthropic_headers(resp.headers) "plan": org.get("rate_limit_tier"),
if util or reset: "status": org.get("subscription_status"),
return util, reset "started": org.get("subscription_created_at"),
}
if sub["started"]:
started = datetime.fromisoformat(sub["started"].replace("Z", "+00:00"))
# The API exposes no renewal date; assume a monthly cycle anchored on the start day.
sub["next_renewal_inferred"] = _next_monthly(started, datetime.now(timezone.utc)).isoformat()
except Exception: except Exception:
pass pass
return util, reset, sub
try:
body = json.dumps({
"model": "claude-haiku-5-5",
"max_tokens": 1,
"messages": [{"role": "user", "content": "hi"}],
}).encode()
req = urllib.request.Request(
"https://api.anthropic.com/v1/messages",
data=body,
headers={**auth, "content-type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=15) as resp:
util, reset = _parse_anthropic_headers(resp.headers)
if util or reset:
return util, reset
return None
except urllib.error.HTTPError as e:
util, reset = _parse_anthropic_headers(e.headers)
if util or reset:
return util, reset
return None
except Exception:
pass
return None
def probe_minimax(api_key: str) -> tuple[dict[str, float], dict[str, int]] | tuple[None, str]: def probe_minimax(api_key: str) -> tuple[dict[str, float], dict[str, int]] | tuple[None, str]:
@@ -347,7 +339,7 @@ def print_dryrun(json_mode: bool = False) -> None:
print(f"{_BOLD}[dryrun] Would probe:{_RESET}") print(f"{_BOLD}[dryrun] Would probe:{_RESET}")
print(f" {_BOLD}Anthropic{_RESET} — {ANTHROPIC_TOKEN_PATH} (Bearer OAuth token)") print(f" {_BOLD}Anthropic{_RESET} — {ANTHROPIC_TOKEN_PATH} (Bearer OAuth token)")
print(f" GET https://api.anthropic.com/v1/models") print(f" GET {ANTHROPIC_API}/usage, {ANTHROPIC_API}/profile")
print(f" {_BOLD}MiniMax{_RESET} — {MINIMAX_SOPS_PATH} (SOPS, key: {MINIMAX_DOTENV_KEY})") print(f" {_BOLD}MiniMax{_RESET} — {MINIMAX_SOPS_PATH} (SOPS, key: {MINIMAX_DOTENV_KEY})")
print(f" GET https://www.minimax.io/v1/token_plan/remains") print(f" GET https://www.minimax.io/v1/token_plan/remains")
@@ -394,6 +386,7 @@ def _provider_record(
reset: dict[str, int] | None, reset: dict[str, int] | None,
error: str | None, error: str | None,
now_epoch: int, now_epoch: int,
subscription: dict | None = None,
) -> dict: ) -> dict:
if error is not None: if error is not None:
return {"provider": name, "available": False, "error": error} return {"provider": name, "available": False, "error": error}
@@ -402,7 +395,10 @@ def _provider_record(
windows: dict[str, dict] = {} windows: dict[str, dict] = {}
for key in ("five_hour", "seven_day"): for key in ("five_hour", "seven_day"):
windows[key] = _window_record(util or {}, reset or {}, key, now_epoch) windows[key] = _window_record(util or {}, reset or {}, key, now_epoch)
return {"provider": name, "available": True, "windows": windows} rec = {"provider": name, "available": True, "windows": windows}
if subscription:
rec["subscription"] = subscription
return rec
def main() -> None: def main() -> None:
@@ -456,16 +452,14 @@ def main() -> None:
# Probe state: (utilization_pct dict, reset_epoch_seconds dict, error string) # Probe state: (utilization_pct dict, reset_epoch_seconds dict, error string)
anth: tuple[dict | None, dict | None, str | None] = (None, None, None) anth: tuple[dict | None, dict | None, str | None] = (None, None, None)
anth_sub: dict = {}
mini: tuple[dict | None, dict | None, str | None] = (None, None, None) mini: tuple[dict | None, dict | None, str | None] = (None, None, None)
# Anthropic # Anthropic
try: try:
token = _read_anthropic_token() token = _read_anthropic_token()
result = probe_anthropic(token) a_util, a_reset, anth_sub = probe_anthropic(token)
if result is None: anth = (a_util, a_reset, None)
anth = (None, None, "probe returned no data")
else:
anth = (result[0], result[1], None)
except Exception as e: except Exception as e:
anth = (None, None, str(e)) anth = (None, None, str(e))
print(f"{_DIM}Warning: Anthropic probe failed: {e}{_RESET}", file=sys.stderr) print(f"{_DIM}Warning: Anthropic probe failed: {e}{_RESET}", file=sys.stderr)
@@ -489,7 +483,7 @@ def main() -> None:
if json_mode: if json_mode:
report = build_json_report([ report = build_json_report([
_provider_record("Anthropic", anth[0], anth[1], anth[2], now_epoch), _provider_record("Anthropic", anth[0], anth[1], anth[2], now_epoch, anth_sub),
_provider_record("MiniMax", mini[0], mini[1], mini[2], now_epoch), _provider_record("MiniMax", mini[0], mini[1], mini[2], now_epoch),
]) ])
print(json.dumps(report, indent=2)) print(json.dumps(report, indent=2))
@@ -519,6 +513,13 @@ def main() -> None:
add_rows("MiniMax", mini[0], mini[1], mini[2]) add_rows("MiniMax", mini[0], mini[1], mini[2])
print_table(rows) print_table(rows)
if anth_sub:
renew = anth_sub.get("next_renewal_inferred")
when = ""
if renew:
secs = int(datetime.fromisoformat(renew).timestamp()) - now_epoch
when = f", renews ~{renew[:10]} (in {_format_relative(secs)}, inferred)"
print(f"Anthropic plan: {anth_sub.get('plan')} ({anth_sub.get('status')}){when}\n")
if __name__ == "__main__": if __name__ == "__main__":

25
specs/agent-subscriptions.spec.md
View File

@@ -19,13 +19,10 @@ The default mode (no flag) prints a coloured table suitable for terminals.
## Behaviour ## Behaviour
1. Read the Anthropic OAuth token from `~/dev/claude/secrets/anthropic/api_key` (whole-file Bearer token). 1. Read the Anthropic OAuth access token from `claudeAiOauth.accessToken` in `$AGENT_SUBSCRIPTIONS_CLAUDE_DIR/.credentials.json` (default `~/.claude-oreillyit-anthropic`). This is the interactive-login credential, which carries the `user:profile` scope; the setup-token key in `secrets/anthropic/api_key` does not and gets 403 `oauth_scope_insufficient`. The file is read-only here: the script never refreshes it. If `expiresAt` has passed, Anthropic is reported unavailable with a "run claude once" hint.
2. Probe Anthropic subscription usage and reset times: 2. Probe Anthropic with `anthropic-beta: oauth-2025-04-20`:
- First attempt: `GET https://api.anthropic.com/v1/models` with `Authorization: Bearer <token>` — zero token cost. - `GET https://api.anthropic.com/api/oauth/usage` returns `five_hour` and `seven_day` objects with `utilization` (percent, 0-100) and `resets_at` (ISO 8601 string). The old `anthropic-ratelimit-unified-*` response headers are no longer returned and are not used.
- Read `anthropic-ratelimit-unified-5h-utilization` and `anthropic-ratelimit-unified-7d-utilization` (fractions 0.0–1.0; multiply by 100 for percentage). - `GET https://api.anthropic.com/api/oauth/profile` (best-effort) returns `organization.rate_limit_tier`, `subscription_status`, `subscription_created_at`. No renewal date is exposed, so `next_renewal_inferred` assumes a monthly cycle on the day-of-month of `subscription_created_at` (day clamped to 28).
- Read `anthropic-ratelimit-unified-5h-reset` and `anthropic-ratelimit-unified-7d-reset`. Values are **Unix epoch seconds** (integer string).
- Source: https://platform.claude.com/docs/en/api/rate-limits and https://github.com/anthropics/claude-code/issues/12829 (example values `1764554400`, `1764615600`).
- If headers absent, fall back: `POST https://api.anthropic.com/v1/messages` with model `claude-haiku-5-5`, `max_tokens=1`, message `"hi"` — same headers on response (or on the HTTPError if 429).
3. Read the MiniMax API key from `~/dev/claude/projects/agent-runtime-secrets/providers/minimax/v1/provider.sops.env` via `sops --decrypt --output-type dotenv` (key: `ANTHROPIC_AUTH_TOKEN`). Uses `SOPS_AGE_KEY_FILE=~/dev/claude/secrets/sops/provider-age-key.txt`. 3. Read the MiniMax API key from `~/dev/claude/projects/agent-runtime-secrets/providers/minimax/v1/provider.sops.env` via `sops --decrypt --output-type dotenv` (key: `ANTHROPIC_AUTH_TOKEN`). Uses `SOPS_AGE_KEY_FILE=~/dev/claude/secrets/sops/provider-age-key.txt`.
4. Probe MiniMax subscription usage and reset times: 4. Probe MiniMax subscription usage and reset times:
- `GET https://www.minimax.io/v1/token_plan/remains` with `Authorization: Bearer <key>` and `User-Agent: curl/7.88.1` (minimax.io blocks Python-urllib default UA). - `GET https://www.minimax.io/v1/token_plan/remains` with `Authorization: Bearer <key>` and `User-Agent: curl/7.88.1` (minimax.io blocks Python-urllib default UA).
@@ -47,12 +44,12 @@ The default mode (no flag) prints a coloured table suitable for terminals.
| Provider | Field | Format | | Provider | Field | Format |
|-----------|---------------------------------------------|---------------------------| |-----------|---------------------------------------------|---------------------------|
| Anthropic | `anthropic-ratelimit-unified-5h-reset` | Unix epoch seconds (int) | | Anthropic | `usage.five_hour.resets_at` | ISO 8601 string |
| Anthropic | `anthropic-ratelimit-unified-7d-reset` | Unix epoch seconds (int) | | Anthropic | `usage.seven_day.resets_at` | ISO 8601 string |
| MiniMax | `model_remains[general].end_time` | Unix epoch **milliseconds** | | MiniMax | `model_remains[general].end_time` | Unix epoch **milliseconds** |
| MiniMax | `model_remains[general].weekly_end_time` | Unix epoch **milliseconds** | | MiniMax | `model_remains[general].weekly_end_time` | Unix epoch **milliseconds** |
Note the unit difference: Anthropic emits seconds; MiniMax emits milliseconds. The implementation normalises both to seconds. Note the format difference: Anthropic emits ISO strings; MiniMax emits milliseconds. The implementation normalises both to seconds.
If a reset field is absent (header missing on the Anthropic side, or the v1 fallback path on the MiniMax side), the reset cell in the table shows `UNKNOWN` and the JSON field `reset_at` is set to `null` with `reset_in_seconds: null`. `reset_in_seconds` is computed as `reset_at - now` at probe time. If a reset field is absent (header missing on the Anthropic side, or the v1 fallback path on the MiniMax side), the reset cell in the table shows `UNKNOWN` and the JSON field `reset_at` is set to `null` with `reset_in_seconds: null`. `reset_in_seconds` is computed as `reset_at - now` at probe time.
@@ -64,8 +61,8 @@ Default (no `--output`):
``` ```
[dryrun] Would probe: [dryrun] Would probe:
Anthropic — ~/dev/claude/secrets/anthropic/api_key (Bearer OAuth token) Anthropic — ~/.claude-oreillyit-anthropic/.credentials.json (Bearer OAuth token)
GET https://api.anthropic.com/v1/models GET https://api.anthropic.com/api/oauth/usage, /api/oauth/profile
MiniMax — ~/dev/claude/projects/agent-runtime-secrets/providers/minimax/v1/provider.sops.env (SOPS) MiniMax — ~/dev/claude/projects/agent-runtime-secrets/providers/minimax/v1/provider.sops.env (SOPS)
GET https://www.minimax.io/v1/token_plan/remains GET https://www.minimax.io/v1/token_plan/remains
``` ```
@@ -122,6 +119,10 @@ With `--output json` (or `--dryrun --output json`):
Note: `probed_at` and the reset values are placeholders — the dryrun output does not depend on the actual wall clock, only on the structure. `window_seconds` is always the fixed constant for that window key (never a placeholder); `elapsed_pct` is derived from `reset_in_seconds` and `window_seconds` so the sample stays internally consistent (shown rounded to 4 decimal places above; the implementation does not round). Note: `probed_at` and the reset values are placeholders — the dryrun output does not depend on the actual wall clock, only on the structure. `window_seconds` is always the fixed constant for that window key (never a placeholder); `elapsed_pct` is derived from `reset_in_seconds` and `window_seconds` so the sample stays internally consistent (shown rounded to 4 decimal places above; the implementation does not round).
## Subscription info (Anthropic)
Default output prints a line below the table: `Anthropic plan: <tier> (<status>), renews ~<date> (in <rel>, inferred)`. JSON adds an optional `subscription` object to the Anthropic provider entry: `plan`, `status`, `started`, `next_renewal_inferred` (ISO 8601). The object is omitted when the profile call fails. The renewal date is inferred, not authoritative.
## Output format ## Output format
### Default table ### Default table