1#!/usr/bin/env python3
2"""Fetch normalized model metadata from Plexus.
3
4Plexus replaced its single-endpoint `/v1/models` (which used to include full
5metadata) with a flattened alias list. Enriched fields now require a
6two-step lookup:
7
81. List aliases via the management API (`/v0/management/aliases`) using the
9 admin key. Each alias may carry a `metadata` block pointing at an
10 external catalog entry (source + source_path).
112. For each alias with metadata, call the public `/v1/metadata/lookup`
12 endpoint to retrieve the normalized record (context_length, capabilities,
13 etc.).
14
15Aliases without a `metadata` block (diff-apply, fix-json, embedding models,
16etc.) are implicitly excluded.
17
18Usage:
19 plexus.py [--json | --tsv]
20"""
21
22from __future__ import annotations
23
24import argparse
25import json
26import os
27import subprocess
28import sys
29import urllib.error
30import urllib.parse
31import urllib.request
32from typing import Any, NotRequired, TypedDict
33
34from models import Model, clean_name, emit_json, emit_tsv
35
36PLEXUS_HOST = "harp-willow.snowy-egret.ts.net"
37PLEXUS_PORT = 4000
38BASE_URL = f"http://{PLEXUS_HOST}:{PLEXUS_PORT}"
39
40OP_ITEM = "op://Shared/Plexus/password"
41
42# Models that must never appear in any client config, even if Plexus reports
43# them. This is a safety net — they usually lack metadata and are filtered
44# at the alias step anyway.
45EXCLUDED_IDS: frozenset[str] = frozenset({
46 "nomic-embed-text-v1.5",
47 "diff-apply",
48 "fix-json",
49})
50
51
52class AliasMetadata(TypedDict):
53 source: str
54 source_path: str
55 overrides: NotRequired[dict[str, Any]]
56
57
58class AliasConfig(TypedDict, total=False):
59 target_groups: list[dict[str, Any]]
60 metadata: AliasMetadata | None
61
62
63def _get_admin_token() -> str:
64 """Retrieve the Plexus admin key from 1Password.
65
66 The `op` CLI must run without `OP_SERVICE_ACCOUNT_TOKEN` set, otherwise
67 it tries to use the service account which lacks access to the Shared
68 vault.
69 """
70 env = {k: v for k, v in os.environ.items() if k != "OP_SERVICE_ACCOUNT_TOKEN"}
71 result = subprocess.run(
72 ["op", "read", OP_ITEM],
73 check=True,
74 capture_output=True,
75 text=True,
76 env=env,
77 )
78 token = result.stdout.strip()
79 if not token:
80 sys.exit("op read returned an empty token")
81 return token
82
83
84def _http_get_json(url: str, headers: dict[str, str] | None = None) -> Any:
85 """Perform a GET request and return the parsed JSON body."""
86 req = urllib.request.Request(url, headers=headers or {})
87 try:
88 with urllib.request.urlopen(req, timeout=30) as resp:
89 body = resp.read().decode("utf-8")
90 except urllib.error.HTTPError as e:
91 sys.exit(f"HTTP {e.code} fetching {url}: {e.read().decode('utf-8', 'replace')}")
92 except urllib.error.URLError as e:
93 sys.exit(f"Network error fetching {url}: {e.reason}")
94 return json.loads(body)
95
96
97def _list_aliases(token: str) -> dict[str, AliasConfig]:
98 """Fetch the alias map from the management API."""
99 url = f"{BASE_URL}/v0/management/aliases"
100 data = _http_get_json(url, {"x-admin-key": token})
101 if not isinstance(data, dict):
102 sys.exit(f"Unexpected alias response type: {type(data).__name__}")
103 return data
104
105
106def _lookup_metadata(source: str, source_path: str) -> dict[str, Any]:
107 """Fetch normalized metadata for a single alias."""
108 params = urllib.parse.urlencode({"source": source, "source_path": source_path})
109 url = f"{BASE_URL}/v1/metadata/lookup?{params}"
110 payload = _http_get_json(url)
111 return payload.get("data") or {}
112
113
114def _build_model(alias_id: str, meta: dict[str, Any]) -> Model:
115 ctx = meta.get("context_length") or 0
116 supported = meta.get("supported_parameters") or []
117 input_modalities = (meta.get("architecture") or {}).get("input_modalities") or []
118
119 return Model(
120 id=alias_id,
121 name=clean_name(meta.get("name") or alias_id),
122 context_length=ctx,
123 max_completion_tokens=(ctx * 20) // 100,
124 supports_reasoning="reasoning" in supported,
125 supports_image="image" in input_modalities,
126 )
127
128
129def fetch_models() -> list[Model]:
130 token = _get_admin_token()
131 aliases = _list_aliases(token)
132
133 models: list[Model] = []
134 for alias_id, config in sorted(aliases.items()):
135 if alias_id in EXCLUDED_IDS:
136 continue
137 metadata = config.get("metadata")
138 if metadata is None:
139 continue
140 if metadata.get("source") == "custom" and "overrides" in metadata:
141 meta = metadata["overrides"]
142 else:
143 meta = _lookup_metadata(metadata["source"], metadata["source_path"])
144 if not meta.get("context_length"):
145 continue
146 models.append(_build_model(alias_id, meta))
147
148 return models
149
150
151def main() -> None:
152 parser = argparse.ArgumentParser(description=__doc__)
153 fmt = parser.add_mutually_exclusive_group()
154 fmt.add_argument("--tsv", action="store_const", dest="format", const="tsv")
155 fmt.add_argument("--json", action="store_const", dest="format", const="json")
156 parser.set_defaults(format="tsv")
157 args = parser.parse_args()
158
159 models = fetch_models()
160
161 if args.format == "json":
162 emit_json(models, sys.stdout)
163 else:
164 emit_tsv(models, sys.stdout)
165
166
167if __name__ == "__main__":
168 main()