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 meta = _lookup_metadata(metadata["source"], metadata["source_path"])
141 if not meta.get("context_length"):
142 continue
143 models.append(_build_model(alias_id, meta))
144
145 return models
146
147
148def main() -> None:
149 parser = argparse.ArgumentParser(description=__doc__)
150 fmt = parser.add_mutually_exclusive_group()
151 fmt.add_argument("--tsv", action="store_const", dest="format", const="tsv")
152 fmt.add_argument("--json", action="store_const", dest="format", const="json")
153 parser.set_defaults(format="tsv")
154 args = parser.parse_args()
155
156 models = fetch_models()
157
158 if args.format == "json":
159 emit_json(models, sys.stdout)
160 else:
161 emit_tsv(models, sys.stdout)
162
163
164if __name__ == "__main__":
165 main()