92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
BASE_URL = "https://modelhub.org.cn/agent_platform"
|
|
|
|
|
|
def get_token(args) -> str:
|
|
token = args.token or os.getenv("AGENT_PLATFORM_TOKEN") or os.getenv("AUTH_TOKEN")
|
|
if not token:
|
|
raise SystemExit("Missing token. Use --token, AGENT_PLATFORM_TOKEN, or AUTH_TOKEN.")
|
|
return token
|
|
|
|
|
|
def request(method: str, path: str, token: str, **kwargs) -> Any:
|
|
url = BASE_URL.rstrip("/") + path
|
|
headers = kwargs.pop("headers", {})
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
if "json" in kwargs:
|
|
headers["Content-Type"] = "application/json"
|
|
resp = requests.request(method, url, headers=headers, timeout=60, **kwargs)
|
|
if resp.status_code == 204:
|
|
return {"status": 204, "message": "no content"}
|
|
try:
|
|
body = resp.json()
|
|
except Exception:
|
|
body = {"status_code": resp.status_code, "text": resp.text}
|
|
if resp.status_code >= 400:
|
|
raise SystemExit(json.dumps(body, ensure_ascii=False, indent=2))
|
|
return body
|
|
|
|
|
|
def print_json(data: Any) -> None:
|
|
print(json.dumps(data, ensure_ascii=False, indent=2))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Agent platform API helper")
|
|
parser.add_argument("--token", default="", help="Bearer token; defaults to AGENT_PLATFORM_TOKEN or AUTH_TOKEN")
|
|
sub = parser.add_subparsers(dest="cmd", required=True)
|
|
|
|
sub.add_parser("me")
|
|
sub.add_parser("list")
|
|
|
|
create = sub.add_parser("create")
|
|
create.add_argument("--name", required=True)
|
|
create.add_argument("--repo-url", required=True)
|
|
create.add_argument("--tag", required=True)
|
|
|
|
for name in ["get", "sync", "deploy", "logs", "stop", "delete"]:
|
|
p = sub.add_parser(name)
|
|
p.add_argument("--strategy-id", required=True)
|
|
if name == "logs":
|
|
p.add_argument("--limit", type=int, default=100)
|
|
|
|
args = parser.parse_args()
|
|
token = get_token(args)
|
|
|
|
if args.cmd == "me":
|
|
print_json(request("GET", "/auth/me", token))
|
|
elif args.cmd == "list":
|
|
print_json(request("GET", "/strategies", token))
|
|
elif args.cmd == "create":
|
|
body = {"name": args.name, "repo_url": args.repo_url, "tag": args.tag}
|
|
data = request("POST", "/strategies", token, json=body)
|
|
print_json(data)
|
|
strategy_id = data.get("id")
|
|
if strategy_id:
|
|
print(f"\nSTRATEGY_ID={strategy_id}", file=sys.stderr)
|
|
elif args.cmd == "get":
|
|
print_json(request("GET", f"/strategies/{args.strategy_id}", token))
|
|
elif args.cmd == "sync":
|
|
print_json(request("POST", f"/strategies/{args.strategy_id}/sync", token))
|
|
elif args.cmd == "deploy":
|
|
print_json(request("POST", f"/strategies/{args.strategy_id}/deploy", token))
|
|
elif args.cmd == "logs":
|
|
print_json(request("GET", f"/strategies/{args.strategy_id}/logs", token, params={"limit": args.limit}))
|
|
elif args.cmd == "stop":
|
|
print_json(request("POST", f"/strategies/{args.strategy_id}/stop", token))
|
|
elif args.cmd == "delete":
|
|
print_json(request("DELETE", f"/strategies/{args.strategy_id}", token))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|