CLI

This is a reference implementation to vendor, not an installed package. Copy the file below into your repository or ops tooling (for example constellation-cli.py), review it, and adapt it. It uses only the Python standard library, so it runs anywhere Python 3.9+ exists with no pip install.

Configuration

The CLI reads two environment variables:

CONSTELLATION_API_KEY (required for authenticated commands): your tenant API key, sent as the x-api-key header.
CONSTELLATION_API_URL (optional): base URL, defaults to https://api.constellation.space. Set it to https://dev.api.constellation.space or https://sandbox.api.constellation.space to target another environment.

Exit codes

Designed for cron and CI:

Code	Meaning
0	Success
1	API or network failure, degraded health, or partial telemetry rejection
2	Usage or configuration error (missing key, unreadable file, invalid input)

The file

#!/usr/bin/env python3
"""constellation-cli: reference CLI for the ConstellationOS API.

Standard library only. Vendor this file and adapt it; it is not a package.

Environment:
  CONSTELLATION_API_KEY  tenant API key (required for authenticated commands)
  CONSTELLATION_API_URL  base URL (default https://api.constellation.space)
"""

import argparse
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request

EXIT_OK = 0
EXIT_API = 1
EXIT_USAGE = 2

MAX_BATCH = 1000
MAX_LINK_IDS = 100
SUBSYSTEMS = ("telemetry", "topology", "predictions")
MEASUREMENTS = ("link", "satellite", "ground_station", "weather", "telemetry")


def base_url() -> str:
    return os.environ.get("CONSTELLATION_API_URL", "https://api.constellation.space").rstrip("/")


def api_key() -> str:
    key = os.environ.get("CONSTELLATION_API_KEY", "")
    if not key:
        print("error: CONSTELLATION_API_KEY is not set", file=sys.stderr)
        sys.exit(EXIT_USAGE)
    return key


def call(method, path, query=None, body=None, auth=True):
    """Returns (status, parsed_body). Exits EXIT_API on network failure."""
    url = base_url() + path
    if query:
        url += "?" + urllib.parse.urlencode(query, doseq=True)
    data = json.dumps(body).encode() if body is not None else None
    headers = {}
    if auth:
        headers["x-api-key"] = api_key()
    if data is not None:
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return resp.status, _parse(resp.read())
    except urllib.error.HTTPError as err:
        return err.code, _parse(err.read())
    except urllib.error.URLError as err:
        print(f"error: cannot reach {url}: {err.reason}", file=sys.stderr)
        sys.exit(EXIT_API)


def _parse(raw: bytes):
    if not raw:
        return None
    try:
        return json.loads(raw)
    except ValueError:
        return raw.decode(errors="replace")


def fail(status, payload):
    code = payload.get("error", f"http_{status}") if isinstance(payload, dict) else f"http_{status}"
    retry = payload.get("retry_after_seconds") if isinstance(payload, dict) else None
    msg = f"error: {code} (HTTP {status})"
    if retry is not None:
        msg += f", retry after {retry}s"
    print(msg, file=sys.stderr)
    print(json.dumps(payload, indent=2), file=sys.stderr)
    sys.exit(EXIT_API)


def emit(payload):
    print(json.dumps(payload, indent=2))


def cmd_health(args):
    path = f"/health/{args.subsystem}" if args.subsystem else "/health"
    status, payload = call("GET", path, auth=False)
    emit(payload)
    sys.exit(EXIT_OK if status == 200 else EXIT_API)


def cmd_topology(args):
    query = {}
    if args.measurement:
        query["measurement"] = args.measurement
    if args.freshness is not None:
        query["freshness_seconds"] = args.freshness
    if args.as_of:
        query["as_of_utc"] = args.as_of
    status, payload = call("GET", "/topology", query=query)
    if status != 200:
        fail(status, payload)
    emit(payload)


def cmd_telemetry_ingest(args):
    try:
        with open(args.file, encoding="utf-8") as handle:
            records = json.load(handle)
    except (OSError, ValueError) as err:
        print(f"error: cannot read {args.file}: {err}", file=sys.stderr)
        sys.exit(EXIT_USAGE)
    if not isinstance(records, list):
        print("error: file must contain a JSON array of telemetry records", file=sys.stderr)
        sys.exit(EXIT_USAGE)

    accepted = rejected = 0
    rejected_indices = []
    for offset in range(0, len(records), MAX_BATCH):
        chunk = records[offset : offset + MAX_BATCH]
        status, payload = call("POST", "/telemetry", body=chunk)
        if status != 202:
            fail(status, payload)
        accepted += payload["accepted_count"]
        rejected += payload["rejected_count"]
        rejected_indices += [offset + i for i in (payload.get("rejected_indices") or [])]
    emit({"accepted_count": accepted, "rejected_count": rejected, "rejected_indices": rejected_indices})
    sys.exit(EXIT_OK if rejected == 0 else EXIT_API)


def cmd_predictions(args):
    links = [s for s in (args.links.split(",") if args.links else []) if s]
    if len(links) > MAX_LINK_IDS:
        print(f"error: --links cap is {MAX_LINK_IDS}, received {len(links)}", file=sys.stderr)
        sys.exit(EXIT_USAGE)
    if args.live and not links:
        print("error: --live requires --links", file=sys.stderr)
        sys.exit(EXIT_USAGE)
    query = [("model_family", args.family)]
    query += [("link_ids", link) for link in links]
    if args.horizon is not None:
        query.append(("horizon_minutes", str(args.horizon)))
    if args.as_of:
        query.append(("as_of_utc", args.as_of))
    if args.live:
        query.append(("live", "true"))
    status, payload = call("GET", "/predictions", query=query)
    if status != 200:
        fail(status, payload)
    emit(payload)


def cmd_smoke(args):
    failures = []

    def stage(name, status, payload, expect=200):
        ok = status == expect
        print(("ok   " if ok else "FAIL ") + name + (f" (HTTP {status})" if not ok else ""))
        if not ok:
            failures.append((name, status, payload))

    status, payload = call("GET", "/health", auth=False)
    stage("GET /health", status, payload)
    for sub in SUBSYSTEMS:
        status, payload = call("GET", f"/health/{sub}", auth=False)
        stage(f"GET /health/{sub}", status, payload)
    status, payload = call("GET", "/topology", query={"freshness_seconds": 3600})
    stage("GET /topology (authenticated)", status, payload)

    if failures:
        for name, status, payload in failures:
            print(f"--- {name}: HTTP {status}", file=sys.stderr)
            print(json.dumps(payload, indent=2), file=sys.stderr)
        sys.exit(EXIT_API)
    print("smoke: all stages passed")


def main():
    parser = argparse.ArgumentParser(prog="constellation-cli", description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)

    p_health = sub.add_parser("health", help="GET /health or /health/<subsystem>")
    p_health.add_argument("subsystem", nargs="?", choices=SUBSYSTEMS)
    p_health.set_defaults(func=cmd_health)

    p_topo = sub.add_parser("topology", help="GET /topology")
    p_topo.add_argument("--measurement", choices=MEASUREMENTS)
    p_topo.add_argument("--freshness", type=float, help="freshness_seconds (1 to 604800)")
    p_topo.add_argument("--as-of", dest="as_of", help="as_of_utc ISO 8601 timestamp")
    p_topo.set_defaults(func=cmd_topology)

    p_tel = sub.add_parser("telemetry", help="telemetry operations")
    tel_sub = p_tel.add_subparsers(dest="telemetry_command", required=True)
    p_ingest = tel_sub.add_parser("ingest", help="POST a JSON array file (auto-batched at 1000)")
    p_ingest.add_argument("file", help="path to a JSON file containing an array of records")
    p_ingest.set_defaults(func=cmd_telemetry_ingest)

    p_pred = sub.add_parser("predictions", help="GET /predictions")
    p_pred.add_argument("--links", help="comma-separated link ids (max 100)")
    p_pred.add_argument("--family", default="snr", help="model_family (default snr)")
    p_pred.add_argument("--horizon", type=int, help="horizon_minutes")
    p_pred.add_argument("--as-of", dest="as_of", help="as_of_utc ISO 8601 timestamp")
    p_pred.add_argument("--live", action="store_true", help="live inference with provenance")
    p_pred.set_defaults(func=cmd_predictions)

    p_smoke = sub.add_parser("smoke", help="staged check: open health, then authenticated topology")
    p_smoke.set_defaults(func=cmd_smoke)

    args = parser.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()

Examples

Check overall and per-subsystem health (no key needed):

python3 constellation-cli.py health
python3 constellation-cli.py health predictions   # exits 1 if degraded (HTTP 503)

Read topology, scoped to fresh link samples:

export CONSTELLATION_API_KEY=...
python3 constellation-cli.py topology --measurement link --freshness 3600
python3 constellation-cli.py topology --as-of 2026-07-09T00:00:00Z

Ingest a file of records (the file must be a JSON array; batching at 1000 is automatic):

python3 constellation-cli.py telemetry ingest ./records.json
# exits 1 if any records were rejected, printing their input indices

Predictions, cached and live:

python3 constellation-cli.py predictions --links gs-madrid--sat-041 --horizon 15
python3 constellation-cli.py predictions --links gs-madrid--sat-041 --live

End-to-end smoke test, ideal for CI or a cron canary:

export CONSTELLATION_API_URL=https://sandbox.api.constellation.space
python3 constellation-cli.py smoke

The smoke command runs stages in order: open /health, each /health/{subsystem} probe, then an authenticated GET /topology that proves your key works end to end. It prints one line per stage and exits 1 if any stage fails.

Notes

The CLI does not retry automatically. On 429 or 503 it prints the error envelope including retry_after_seconds to stderr and exits 1, which is the right behavior under cron: the next scheduled run is the retry.
Output is always JSON on stdout; diagnostics go to stderr, so pipelines like ... | jq .entities stay clean.

Configuration​

Exit codes​

The file​

Examples​

Notes​

Configuration

Exit codes

The file

Examples

Notes