Something went wrong. Try again.
replay a stream.place chat from a jetstream archive streamplace-mcp.fastmcp.app/stats
streaming mcp atproto
Something went wrong. Try again.
2.5 kB · 66 lines
Python
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667"""usage of the hosted server from Horizon's records: calls, errors, latency, clients.
HORIZON_API_KEY=… uv run --no-project python scripts/usage.py [--days 7]"""
from __future__ import annotations
import argparseimport jsonimport osimport sysimport urllib.parseimport urllib.requestfrom collections import Counterfrom datetime import UTC, datetime, timedelta
API = "https://horizon.prefect.io/api/v0"ORG = "ead2665d-917e-495b-94cf-29e4918382c3"PROJECT = "1ffb780e-002f-45e4-8e37-ea3a28cf78e5"
def api(path: str, **params) -> dict: url = f"{API}{path}?{urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})}" req = urllib.request.Request(url, headers={"authorization": f"Bearer {os.environ['HORIZON_API_KEY']}"}) with urllib.request.urlopen(req, timeout=60) as r: return json.loads(r.read())
def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--days", type=int, default=7) args = ap.parse_args() if "HORIZON_API_KEY" not in os.environ: sys.exit("HORIZON_API_KEY is not set") end = datetime.now(UTC) start = end - timedelta(days=args.days) s, e = start.isoformat(timespec="seconds"), end.isoformat(timespec="seconds") base = f"/organizations/{ORG}/projects/{PROJECT}/targets" target = api(base)["items"][0]["id"]
summary = api(f"{base}/{target}/usage/summary", start=s, end=e) print(f"last {args.days}d: {summary['calls']} calls, error rate {summary['errorRate']:.1%}, p95 {summary['p95Ms'] / 1000:.1f}s") for t in api(f"{base}/{target}/usage/tools", start=s, end=e)["items"]: print(f" {t['name']:<16} {t['calls']:>5} calls errors {t['errorRate']:.0%} p95 {t['p95Ms'] / 1000:.1f}s")
clients: Counter = Counter() cursor = None while True: page = api(f"{base}/{target}/logs", startDate=s, endDate=e, limit=200, cursor=cursor) for item in page.get("items", []): if (item.get("mcp") or {}).get("method") != "tools/call": continue client = (item.get("mcp") or {}).get("client") or {} clients[client.get("name") or (item.get("http") or {}).get("request", {}).get("userAgent") or "unknown"] += 1 cursor = page.get("cursor") or page.get("nextCursor") if not cursor: break if clients: print("tool calls by client:") for name, n in clients.most_common(15): print(f" {name:<40} {n}") return 0
if __name__ == "__main__": sys.exit(main())