Use MOBS from an AI agent

Connect to MOBS over WebSocket. HTTP is backfill only.

- Read the API key from the environment variable MOBS_TOKEN.
  Never log the token, and never log a URL that includes it.
- Open wss://mobs.ws/v1/ws/metar?airports=ICAO,ICAO
  (comma list, case-insensitive). Default mode is raw.
- Authenticate with the header
  Authorization: Bearer $MOBS_TOKEN.
  Browsers cannot set socket headers; they may pass
  ?token=$MOBS_TOKEN on the handshake URL instead.
  If no Bearer header is present, the server reads
  ?token=, then the website cookie. Do not use the
  website cookie. The server does not accept a
  WebSocket subprotocol as the key.
- The first server message is subscription_updated.
  No reports are sent on connect.
- Then GET https://mobs.ws/v1/metar/latest?airports=…
  with the same Bearer header to backfill the latest
  report per airport. After every reconnect, backfill
  again. There is no history API; missed reports other
  than the current latest cannot be recovered.
- Send {"action":"ping"} every 20 seconds. The reply is
  {"type":"pong"}. After 60 seconds with no JSON ping
  and no valid subscribe or unsubscribe, the server
  closes with 4000. RFC 6455 protocol pings do not
  reset that idle timer.
- Parse JSON text frames. Types:
  subscription_updated (icaos list),
  pong,
  error (socket stays open; codes invalid_airports,
  bad_request, unknown_action, too_many_airports),
  new_metar_arrived (data is an array of records).
- Each record always has icao, raw_metar, obs_time,
  report_type (METAR or SPECI), arrived_at. In mode=
  parsed a parsed object may be added; absent keys are
  omitted. Do not expect correction, replaces_report_id
  or revision on the wire.
- A SPECI is a new_metar_arrived record with
  report_type "SPECI" (raw_metar starts with SPECI).
- A correction is a later record with the same icao and
  obs_time and a different raw_metar. Keep the later
  record. In parsed mode, parsed.corrected may be true.
- Dedupe on icao + obs_time, replacing the stored
  record when raw_metar changes.
- Reconnect with exponential backoff and jitter on
  unexpected close, 4000 (fix the heartbeat first),
  4008 (24 h lifetime; reconnect), and 1013 (temporary:
  cap overshoot or a slow reader). Do not retry 4403
  (API key revoked or expired). Backoff is capped at
  about 30 s (with jitter, about 15-45 s).
- Reset the backoff after the first message, or after
  about 10 s of an open socket (the 10 s timer starts
  when the socket opens, not when it is created). A
  socket that is accepted and closed at once (for
  example 1013) must not reset the delay.
- After 24 hours the server closes with 4008; reconnect
  and backfill again.

wss://mobs.ws/v1/ws/metar?airports=KNYC,KLGA

{"action":"subscribe","icaos":["KORD"]}
{"action":"unsubscribe","icaos":["KLGA"]}

import asyncio, json, os, random
from websockets.asyncio.client import connect
from websockets.exceptions import (
    ConnectionClosedError,
    InvalidStatus,
)
import httpx

TOKEN = os.environ["MOBS_TOKEN"]
AIRPORTS = "KNYC,KLGA"
WS_URL = (
    "wss://mobs.ws/v1/ws/metar"
    f"?airports={AIRPORTS}"
)
HEADERS = {"Authorization": f"Bearer {TOKEN}"}
SEEN = {}  # (icao, obs_time) -> raw_metar


def merge(rec):
    key = (rec["icao"], rec["obs_time"])
    prev = SEEN.get(key)
    if prev == rec["raw_metar"]:
        return
    SEEN[key] = rec["raw_metar"]
    print(rec["icao"], rec["report_type"], rec["raw_metar"])


async def backfill():
    async with httpx.AsyncClient(timeout=30) as client:
        r = await client.get(
            "https://mobs.ws/v1/metar/latest",
            params={"airports": AIRPORTS},
            headers=HEADERS,
        )
        r.raise_for_status()
        for rec in r.json().get("data", []):
            merge(rec)


async def heartbeat(ws):
    while True:
        await asyncio.sleep(20)
        await ws.send(json.dumps({"action": "ping"}))


async def arm_reset():
    global delay
    await asyncio.sleep(10)
    delay = 1


async def session():
    global delay
    async with connect(
        WS_URL, additional_headers=HEADERS
    ) as ws:
        beat = asyncio.create_task(heartbeat(ws))
        hold = asyncio.create_task(arm_reset())
        try:
            await backfill()
            async for text in ws:
                delay = 1
                hold.cancel()
                msg = json.loads(text)
                kind = msg.get("type")
                if kind == "new_metar_arrived":
                    for rec in msg["data"]:
                        merge(rec)
                elif kind == "error":
                    print("error", msg)
        finally:
            hold.cancel()
            beat.cancel()


delay = 1


async def main():
    global delay
    while True:
        try:
            await session()
        except ConnectionClosedError as exc:
            if exc.rcvd and exc.rcvd.code == 4403:
                raise SystemExit(
                    "API key revoked"
                ) from exc
            print("session ended", exc)
        except InvalidStatus as exc:
            if exc.response.status_code in (
                401, 403
            ):
                raise SystemExit(
                    "API key rejected"
                ) from exc
            print("session ended", exc)
        except SystemExit:
            raise
        except Exception as exc:
            print("session ended", exc)
        wait = delay * (0.5 + random.random())
        print("reconnect in", round(wait, 1), "s")
        await asyncio.sleep(wait)
        delay = min(delay * 2, 30)  # about 30 s


asyncio.run(main())

const TOKEN = globalThis.process?.env?.MOBS_TOKEN
  ?? "YOUR_API_KEY";
const AIRPORTS = "KNYC,KLGA";
const SEEN = new Map();
let attempt = 0;

function merge(rec) {
  const key = rec.icao + " " + rec.obs_time;
  if (SEEN.get(key) === rec.raw_metar) return;
  SEEN.set(key, rec.raw_metar);
  console.log(rec.icao, rec.report_type, rec.raw_metar);
}

async function backfill() {
  const url = (
    "https://mobs.ws/v1/metar/latest"
    + "?airports=" + encodeURIComponent(AIRPORTS)
  );
  const res = await fetch(url, {
    headers: {Authorization: "Bearer " + TOKEN},
  });
  if (!res.ok) throw new Error("backfill " + res.status);
  const body = await res.json();
  for (const rec of body.data || []) merge(rec);
}

function openSocket() {
  const qs = new URLSearchParams({
    airports: AIRPORTS,
    token: TOKEN,
  });
  const ws = new WebSocket(
    "wss://mobs.ws/v1/ws/metar?" + qs
  );
  const beat = setInterval(() => {
    if (ws.readyState === WebSocket.OPEN) {
      ws.send('{"action":"ping"}');
    }
  }, 20000);
  let hold = null;
  function markOk() {
    attempt = 0;
    if (hold) clearTimeout(hold);
  }
  ws.onmessage = (e) => {
    markOk();
    const msg = JSON.parse(e.data);
    if (msg.type === "new_metar_arrived") {
      for (const rec of msg.data) merge(rec);
    }
  };
  ws.onopen = () => {
    hold = setTimeout(() => {
      attempt = 0;
    }, 10000);
    backfill().catch(console.error);
  };
  ws.onclose = (e) => {
    clearInterval(beat);
    if (hold) clearTimeout(hold);
    if (e.code === 4403) {
      throw new Error("API key revoked");
    }
    const wait = Math.min(
      30000,  // about 30 s
      1000 * 2 ** attempt
    ) * (0.5 + Math.random());
    attempt = Math.min(attempt + 1, 5);
    setTimeout(openSocket, wait);
  };
}

openSocket();