Worked example: a team status board

A backend you host reads the service snapshot once a minute and serves it to your signed-in team, with omitted services, coverage and staleness shown honestly.

Read as Markdown ↗Use with an LLM ↗

On this page

A status board for your team: which services reported activity in the last hour, how busy they were, their error rate, latency and health. One backend reads Anectico once a minute and serves every viewer from that reading. Viewers sign in to the board, never to Anectico, and never see the key.

The runnable example is scripts/testing/examples/apps/status_board.py in the Anectico repository, with the shared client anectico_rest.py beside it. It needs Python 3.11 or later and nothing else. Your agent adapts it in your own workspace; Anectico does not host, generate or run it.

Before you start

Decision	This example
Where it runs	Your server or internal-tool host, as one long-running process.
Project	Exactly one. The key is bound to it and every read names it in `project_id`.
Runtime permission	`services:read`, nothing else.
Builder authority	Your agent's own credential provisions the key (see Give an app its own key); the board never holds it.
Secret	`ANECTICO_API_KEY` from the host's secret store. Viewer passwords (`BOARD_VIEWERS`) are the board's own.
Expiry and stop	A fixed expiry, at most 365 days. Stop the board's access with `anectico apikey revoke <key-id>`.
Contract	The revision and app-document digest in `contract-pin.json`. The board refuses to start against any other.

What it calls

Operation	Qualification	What the restriction means for this board
`GET /api/v1/api-contract`		Read once at start-up, with no key, to compare the deployment's contract with the pin.
`GET /api/v1/services`	Restricted	Its limits are published but qualified only on a development workstation, and a filter (for example `environment`) is not yet applied before the list is cut. The board therefore reads one project, unfiltered, over a window of at most 24 hours, at most once a minute, and treats every answer as a measurement with stated gaps rather than a guarantee.

The service snapshot is a top list: it orders every service that reported entry spans in the window and then cuts the list. There is no next page, no cursor and no offset (both are refused), and the answer says how many services were eligible, matched, returned and omitted. A service health snapshot has no proof page in Anectico, so the board shows no "open in Anectico" link for a service.

Your agent builds against the declaration of exactly these operations, fetched from the deployment and checked against the pinned revision:

anectico api spec --partition apps --revision "$(jq -r .contract_revision contract-pin.json)" \
  --operation get_services --out status-board.contract.json

Configuration

@dataclass(frozen=True)
class Config:
    api_url: str
    api_key: str
    project_id: str
    viewers: dict[str, str]
    listen: tuple[str, int]
    window_minutes: int
    refresh_seconds: int
    limit: int


def load_config(env: Mapping[str, str]) -> Config:
    def need(name: str) -> str:
        value = env.get(name, "").strip()
        if not value:
            raise SystemExit(f"{name} is required")
        return value

    viewers = dict(pair.split(":", 1) for pair in need("BOARD_VIEWERS").split(",") if ":" in pair)
    if not viewers or any(len(password) < 12 for password in viewers.values()):
        raise SystemExit("BOARD_VIEWERS must list name:password pairs with passwords of 12 or more characters")
    host, _, port = env.get("BOARD_LISTEN", "127.0.0.1:8090").rpartition(":")
    window = int(env.get("BOARD_WINDOW_MINUTES", "60"))
    refresh = int(env.get("BOARD_REFRESH_SECONDS", "60"))
    limit = int(env.get("BOARD_LIMIT", "20"))
    if not 1 <= window <= 24 * 60:
        raise SystemExit("BOARD_WINDOW_MINUTES must be 1 to 1440: the published advice is a window of 24 hours or less")
    if refresh < 60:
        raise SystemExit("BOARD_REFRESH_SECONDS must be 60 or more: read the snapshot at most once a minute per project")
    if not 1 <= limit <= 200:
        raise SystemExit("BOARD_LIMIT must be 1 to 200")
    return Config(need("ANECTICO_API_URL"), need("ANECTICO_API_KEY"), need("ANECTICO_PROJECT_ID"),
                  viewers, (host or "127.0.0.1", int(port)), window, refresh, limit)

The board refuses to start with a cadence faster than once a minute, a window longer than a day or a top list longer than 200. HTTP Basic sign-in keeps the example small; put the board behind your own single sign-on in production.

Before serving, the board compares the deployment's contract with the one it was tested against:

def verify_contract(self, pin_file: Path = PIN_FILE) -> dict[str, Any]:
    """Read the deployment's contract manifest and compare it with the pin.

    The manifest needs no key. The pin holds the contract revision and the
    digest of the app document the program was tested against. Any
    difference refuses to start: before the first supported release routes
    and shapes may change without notice.
    """
    pin = json.loads(pin_file.read_text())
    manifest = self._send("GET", "/api/v1/api-contract", None, authenticated=False, deadline=None).body
    served = {a["partition"]: a["digest"] for a in manifest.get("artifacts", [])}
    if manifest.get("contract_revision") != pin["contract_revision"] or served.get("apps") != pin["apps_digest"]:
        raise ContractMismatch(
            f"this program was tested against contract {pin['contract_revision']} (apps {pin['apps_digest']}); "
            f"the deployment serves {manifest.get('contract_revision')} (apps {served.get('apps')}). "
            "Re-test the program against the served contract before moving the pin."
        )
    return manifest

One read for everyone

A background thread refreshes the reading every minute. Viewers read the last result and never cause a read:

def refresh_once(self) -> None:
    end = self._clock().replace(second=0, microsecond=0)
    start = end - timedelta(minutes=self._config.window_minutes)
    query = {
        "project_id": self._config.project_id,  # exact project, even though the key is bound to it
        "start_time": _iso(start),               # explicit, ordered window
        "end_time": _iso(end),
        "sort_by": "request_count",
        "sort_order": "desc",
        "limit": str(self._config.limit),        # no cursor, no offset: this is a top list
    }
    attempt_at = _iso(self._clock())
    try:
        answer = self._client.get("/api/v1/services", query, deadline=20).body
        snapshot = self._snapshot(answer, start, end)
    except (AccessEnded, Forbidden) as ended:
        self._fail("access_ended", ended, attempt_at)
        self.stopped = True  # a revoked or narrowed key stays refused; stop asking
        return
    except Paused as paused:
        self._fail("paused", paused, attempt_at)
        return
    except Busy as busy:
        self._fail(None, busy, attempt_at)
        return
    except AnecticoError as refused:
        self._fail("refused", refused, attempt_at)
        self.stopped = True  # the same request would be refused again
        return
    except (ValueError, KeyError, TypeError) as invalid:
        self._fail(None, AnecticoError(f"answer did not match the contract: {invalid}"), attempt_at)
        return
    with self._lock:
        self._view = {"state": "current", "snapshot": snapshot, "last_error": None,
                      "last_attempt_at": attempt_at, "last_success_at": attempt_at}

The shared client already retried a busy read (READ_ADMISSION_BUSY), a request-rate refusal and a 503 after Retry-After, within a 20-second deadline. What reaches the board is the outcome:

Outcome	Board state	What viewers see
An answer	`current`	The reading, its window and the time of the last successful refresh.
Still busy or unavailable after the retries	`stale`, or `unavailable` with no earlier reading	The previous reading, marked as such, and when it was taken. Never an empty board.
An operation budget, or a request rate an owner set to zero	`paused`	"Reading is paused by a limit an owner set." No retry storm.
`401` or `403`	`access_ended`	"This board's access to Anectico has ended." The board stops reading; it never switches keys.
Another refusal	`refused`	The board needs fixing; it stops reading.

What a reading may claim

def measurement(answer: dict[str, Any]) -> tuple[str, list[dict[str, Any]]]:
    """What the snapshot can honestly claim about its window."""
    coverage = (answer.get("read_coverage") or {}).get("sources") or []
    outcomes = {source.get("outcome") for source in coverage}
    if not coverage:
        state = "coverage_unknown"
    elif "READ_COVERAGE_OUTCOME_OUTSIDE_RETENTION" in outcomes:
        state = "outside_retention"
    elif "READ_COVERAGE_OUTCOME_PARTIALLY_COVERED" in outcomes:
        state = "partially_covered"
    elif outcomes != {"READ_COVERAGE_OUTCOME_COMPLETE"}:
        state = "coverage_unknown"
    elif int(_count(answer["matched_service_count"])) == 0:
        state = "no_observations"
    else:
        state = "observed"
    described = [{
        "source": source.get("source"),
        "outcome": source.get("outcome"),
        "requested": source.get("requested_interval"),
        "effective": source.get("effective_interval"),
        "retention_floor": source.get("retention_floor"),
    } for source in coverage]
    return state, described

No observations is not "all up". A window in which nothing reported shows "No service reported activity in this window. That is not the same as everything being up."
Coverage is shown. A window older than retained data, or only partly inside it, says so, with the effective window the answer reports.
Reporters are never claimed complete. upstream_completeness is UPSTREAM_COMPLETENESS_UNKNOWN: a service that sent nothing is absent, so the board says "A service that sent nothing in the window does not appear here" and never prints "all systems operational".
Omitted services are counted. "Showing 3 of 5 services that reported in the window, ordered by request_count; 2 not shown." The health tally covers only the services shown.
Counts stay exact. request_count is a 64-bit decimal string; the board validates it and shows the string, so 18446744073709551615 is never rounded. A rate of "NaN" shows as "not measurable".
Captured names are text. Service names come from your telemetry. The page is rendered on the server with every value HTML-escaped, carries no script at all, and is served with Content-Security-Policy: default-src 'none'; frame-ancestors 'none'.

The board's own API

The board serves / (HTML that refreshes itself every minute) and /board.json (the same reading as JSON) to signed-in viewers. Neither takes a parameter: a viewer cannot choose the project, the window, the order or an Anectico path, and any query string is refused with 400. Anything else is 404; the board is not a proxy.

Run it

export ANECTICO_API_URL=https://app.anectico.com ANECTICO_PROJECT_ID=<project-uuid>
export ANECTICO_API_KEY="$(your-secret-store read anectico/status-board)"
export BOARD_VIEWERS='ops:a-long-viewer-password' BOARD_LISTEN=127.0.0.1:8090 BOARD_LIMIT=20
python3 status_board.py

Your agent can compare the board with the same top list through the CLI, using its own credential:

anectico services list --since 1h --sort-by request_count --limit 20 -o json

Stop it and clean up

Revoke the board's key (anectico apikey revoke <key-id>); within one refresh the board shows that its access has ended and stops reading. Revocation does not stop the process, and readings it already served stay on screens until they reload. The board keeps only its last reading in memory; if you add a cache or logs, they are yours to retain and delete. See Revoke an agent's access.