API 接入文档

curl https://tryaiapi.com/v1/models \ -H "Authorization: Bearer $TRYAIAPI_KEY"

curl https://tryaiapi.com/v1/user/balance \ -H "Authorization: Bearer <你的 API Key>"

{ "code": "success", "data": { "user_id": 1001, "scope": "account", "balance_usd": 61.66, "balance_cny": 431.62, "quota_remaining": 30829650, "currency_default": "USD", "conversion_ratio": 7, "status": "normal", "timestamp": 1791108566, "cached": false, "age_seconds": 0 } }

import os import requests ALERT_BELOW_USD = 15.0 # 换成你自己的预警线（美元） def check_balance(): r = requests.get( "https://tryaiapi.com/v1/user/balance", headers={"Authorization": f"Bearer {os.environ['INFWAVE_API_KEY']}"}, timeout=10, ) if r.status_code == 429: # 限流，不是余额不足：稍后再查 return None r.raise_for_status() return r.json()["data"] data = check_balance() if data is not None and (data["balance_usd"] < ALERT_BELOW_USD or data["status"] != "normal"): # 在这里接入你自己的通知渠道（企业微信 / 邮件 / 短信），并暂停提交新任务 print(f"余额不足预警：当前 ${data['balance_usd']:.2f}（约 ¥{data['balance_cny']:.2f}）")

#!/usr/bin/env python3 """TryAIAPI 余额提醒：每小时查一次余额，低于你设的线就推送到飞书 / 钉钉 / 企业微信 / Telegram。只用 Python 标准库。 环境变量： TRYAIAPI_KEY 调用模型用的 API Key NOTIFY_CHANNEL feishu / dingtalk / wecom / telegram，四选一 飞书：FEISHU_WEBHOOK（机器人开了签名校验再加 FEISHU_SECRET） 企业微信：WECOM_WEBHOOK 钉钉：DINGTALK_WEBHOOK（机器人开了加签再加 DINGTALK_SECRET） Telegram：TG_BOT_TOKEN、TG_CHAT_ID LOW_USD / CRIT_USD 偏低线 / 告急线（美元），默认 40 / 10 定时运行（crontab，每小时一次）： 0 * * * * /usr/bin/python3 /path/to/balance_alert.py >> balance_alert.log 2>&1 行为：偏低、告急各提醒一次；查询失败不当作余额不足，连续 3 次失败提醒「监控断了」。""" import base64 import hashlib import hmac import json import os import time import urllib.parse import urllib.request from pathlib import Path API = "https://tryaiapi.com/v1/user/balance" KEY = os.environ.get("TRYAIAPI_KEY", "") CHANNEL = os.environ.get("NOTIFY_CHANNEL", "feishu") # feishu / wecom / dingtalk / telegram LOW = float(os.environ.get("LOW_USD", "40")) CRIT = float(os.environ.get("CRIT_USD", "10")) STATE = Path(os.environ.get("STATE_FILE", str(Path.home() / ".tryaiapi_balance_alert.json"))) TIERS = ["正常", "偏低", "告急"] UA = "balance-alert/1.0" def http_json(url, payload=None, headers=None): data = json.dumps(payload).encode() if payload is not None else None h = {"User-Agent": UA, **(headers or {})} if data: h["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=h) with urllib.request.urlopen(req, timeout=15) as r: return json.loads(r.read().decode()) def query_balance(): data = http_json(API, headers={"Authorization": f"Bearer {KEY}"})["data"] usd = float(data["balance_usd"]) return usd, f"${usd:.2f}（约 ¥{float(data['balance_cny']):.2f}）" def send_once(text): if CHANNEL == "feishu": # 飞书自定义机器人 webhook payload = {"msg_type": "text", "content": {"text": text}} secret = os.environ.get("FEISHU_SECRET") if secret: # 机器人开了「签名校验」才需要 ts = str(int(time.time())) key = f"{ts}\n{secret}".encode() payload["timestamp"] = ts payload["sign"] = base64.b64encode(hmac.new(key, b"", hashlib.sha256).digest()).decode() return http_json(os.environ["FEISHU_WEBHOOK"], payload).get("code") == 0 if CHANNEL == "wecom": # 企业微信群机器人 webhook res = http_json(os.environ["WECOM_WEBHOOK"], {"msgtype": "text", "text": {"content": text}}) return res.get("errcode") == 0 if CHANNEL == "dingtalk": # 钉钉群自定义机器人 webhook url = os.environ["DINGTALK_WEBHOOK"] secret = os.environ.get("DINGTALK_SECRET") if secret: # 机器人开了「加签」才需要 ts = str(int(time.time() * 1000)) digest = hmac.new(secret.encode(), f"{ts}\n{secret}".encode(), hashlib.sha256).digest() url += f"&timestamp={ts}&sign={urllib.parse.quote_plus(base64.b64encode(digest))}" res = http_json(url, {"msgtype": "text", "text": {"content": text}}) return res.get("errcode") == 0 if CHANNEL == "telegram": # Telegram bot（国内服务器通常连不上） url = f"https://api.telegram.org/bot{os.environ['TG_BOT_TOKEN']}/sendMessage" return http_json(url, {"chat_id": os.environ["TG_CHAT_ID"], "text": text}).get("ok") is True raise SystemExit("NOTIFY_CHANNEL 只能是 feishu / wecom / dingtalk / telegram") def notify(text): for i in range(3): try: if send_once(text): return True print("消息被拒收，稍后重试") except Exception as e: print(f"消息发送异常：{type(e).__name__}") # 只打印异常类型，不打印网址，避免 token 进日志 time.sleep(2 ** i) return False def load_state(): try: return json.loads(STATE.read_text()) except Exception: return {"tier": 0, "fails": 0, "down_alerted": False} def main(): if not KEY: raise SystemExit("请先设置 TRYAIAPI_KEY") st = load_state() try: usd, shown = query_balance() except Exception as e: st["fails"] += 1 print(f"查询失败（第 {st['fails']} 次）：{type(e).__name__}") if st["fails"] >= 3 and not st["down_alerted"]: if notify("【TryAIAPI 余额监控】已连续 3 次查不到余额，监控可能中断，请检查网络或 API Key。"): st["down_alerted"] = True STATE.write_text(json.dumps(st)) return st["fails"], st["down_alerted"] = 0, False tier = 2 if usd < CRIT else 1 if usd < LOW else 0 if tier > st["tier"]: # 跌进更差的一档才提醒，同一档不重复提醒 line = CRIT if tier == 2 else LOW msg = (f"【TryAIAPI 余额{TIERS[tier]}】当前可用余额 {shown}，" f"已低于{TIERS[tier]}线 ${line:.0f}，请及时充值，以免调用中断。") if notify(msg): st["tier"] = tier # 发送成功才记账，失败下轮重试 elif tier < st["tier"] and (tier > 0 or usd >= LOW * 1.2): # 回到偏低线的 1.2 倍以上才算恢复，避免在线附近反复提醒 st["tier"] = tier if tier == 0: notify(f"【TryAIAPI 余额已恢复】当前可用余额 {shown}。") print(f"余额 ${usd:.2f}，档位 {TIERS[st['tier']]}") STATE.write_text(json.dumps(st)) if __name__ == "__main__": main()