100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Forward Claude Code status-line usage windows to Pipeline.
|
||
|
|
|
||
|
|
Claude Code passes a JSON snapshot on stdin to status-line commands. This helper
|
||
|
|
extracts only the documented rate_limits fields and posts them to Pipeline so the
|
||
|
|
dashboard can show provider-native Current session and All models reset windows.
|
||
|
|
|
||
|
|
Required environment:
|
||
|
|
PIPELINE_API_URL e.g. http://localhost:8001
|
||
|
|
PIPELINE_AUTH_TOKEN local bearer token or user token
|
||
|
|
PIPELINE_GATEWAY_ID gateway UUID to attach this local Claude Code session to
|
||
|
|
|
||
|
|
Optional environment:
|
||
|
|
PIPELINE_STATUSLINE_TIMEOUT_SECONDS default: 2
|
||
|
|
PIPELINE_STATUSLINE_PRINT set to 0 to suppress compact status output
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import urllib.error
|
||
|
|
import urllib.request
|
||
|
|
|
||
|
|
|
||
|
|
def _compact_status(payload: dict[str, object]) -> str:
|
||
|
|
model = payload.get("model")
|
||
|
|
model_name = "Claude"
|
||
|
|
if isinstance(model, dict):
|
||
|
|
display = model.get("display_name")
|
||
|
|
if isinstance(display, str) and display.strip():
|
||
|
|
model_name = display.strip()
|
||
|
|
|
||
|
|
rate_limits = payload.get("rate_limits")
|
||
|
|
if not isinstance(rate_limits, dict):
|
||
|
|
return f"[{model_name}]"
|
||
|
|
|
||
|
|
parts: list[str] = []
|
||
|
|
five = rate_limits.get("five_hour")
|
||
|
|
if isinstance(five, dict) and isinstance(five.get("used_percentage"), int | float):
|
||
|
|
parts.append(f"5h {five['used_percentage']:.0f}%")
|
||
|
|
week = rate_limits.get("seven_day")
|
||
|
|
if isinstance(week, dict) and isinstance(week.get("used_percentage"), int | float):
|
||
|
|
parts.append(f"7d {week['used_percentage']:.0f}%")
|
||
|
|
return f"[{model_name}] {' · '.join(parts)}" if parts else f"[{model_name}]"
|
||
|
|
|
||
|
|
|
||
|
|
def _sanitize(raw: dict[str, object]) -> dict[str, object]:
|
||
|
|
payload: dict[str, object] = {}
|
||
|
|
for key in ("session_id", "model", "workspace", "rate_limits"):
|
||
|
|
value = raw.get(key)
|
||
|
|
if value is not None:
|
||
|
|
payload[key] = value
|
||
|
|
return payload
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> int:
|
||
|
|
try:
|
||
|
|
raw = json.load(sys.stdin)
|
||
|
|
except json.JSONDecodeError:
|
||
|
|
return 0
|
||
|
|
if not isinstance(raw, dict):
|
||
|
|
return 0
|
||
|
|
|
||
|
|
payload = _sanitize(raw)
|
||
|
|
if os.environ.get("PIPELINE_STATUSLINE_PRINT") != "0":
|
||
|
|
print(_compact_status(payload))
|
||
|
|
|
||
|
|
api_url = os.environ.get("PIPELINE_API_URL", "").strip().rstrip("/")
|
||
|
|
token = os.environ.get("PIPELINE_AUTH_TOKEN", "").strip()
|
||
|
|
gateway_id = os.environ.get("PIPELINE_GATEWAY_ID", "").strip()
|
||
|
|
if not api_url or not token or not gateway_id:
|
||
|
|
return 0
|
||
|
|
|
||
|
|
url = f"{api_url}/api/v1/gateways/{gateway_id}/provider-usage/claude/statusline"
|
||
|
|
body = json.dumps(payload).encode("utf-8")
|
||
|
|
timeout = float(os.environ.get("PIPELINE_STATUSLINE_TIMEOUT_SECONDS", "2"))
|
||
|
|
request = urllib.request.Request(
|
||
|
|
url,
|
||
|
|
data=body,
|
||
|
|
method="POST",
|
||
|
|
headers={
|
||
|
|
"Authorization": f"Bearer {token}",
|
||
|
|
"Content-Type": "application/json",
|
||
|
|
"User-Agent": "PipelineClaudeStatusline/1.0",
|
||
|
|
},
|
||
|
|
)
|
||
|
|
try:
|
||
|
|
with urllib.request.urlopen(request, timeout=timeout):
|
||
|
|
pass
|
||
|
|
except (OSError, urllib.error.URLError, urllib.error.HTTPError):
|
||
|
|
# Status-line hooks must never disrupt Claude Code if Pipeline is down.
|
||
|
|
return 0
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|