1
Architecture
Latte edited this page 2026-06-29 13:43:07 +00:00

Architecture

familiar is a FastMCP server with a deliberately uniform shape: a single FastMCP instance, a set of tool modules that all look the same, and a few shared helpers underneath them. This page explains how those pieces fit together.

Server assembly

src/familiar/server.py creates the single mcp = FastMCP(...) instance, defines the ping health tool, and then calls register(mcp) on every tool module in turn.

Every tool module follows the same shape:

  • module-level async def functions — the actual tools (plain importable functions, so tests call them directly);
  • a TOOLS list naming the functions to expose;
  • a register(mcp) that loops over TOOLS calling mcp.tool(fn).

Registration is kept separate from definition, which is why tests can import and call a tool function without going through MCP. To add tools: create a new module in this shape, then import and register it in server.py.

The modules, in registration order: universe, market, sde, account, industry, trade, structures, writes, killmails, profile, corp, corp_economy, misc, comms, contacts, fittings, fleet, ui, planets.

Data sources

ESI — the EVE Swagger Interface

esi.py is a thin async httpx wrapper around one ESI base URL:

  • A client() singleton (created on first use), replaceable via set_client() so tests can inject an httpx.MockTransport.
  • get / post / put / delete, plus get_all_pages which follows ESI's X-Pages header to concatenate every page of a paginated endpoint.
  • A descriptive User-Agent (ESI asks for one) and a uniform EsiError carrying the HTTP status.
  • Response caching honoring each response's Expires header (GET only; disable with FAMILIAR_DISABLE_CACHE).
  • Error-limit back-off: it watches the rolling X-ESI-Error-Limit-Remain / -Reset headers and waits for the window to reset when it gets near the threshold, so familiar never trips ESI's error limit.

Authenticated calls pass a Bearer token=; the base URL and datasource are env-overridable.

zKillboard

zkill.py mirrors the ESI client's structure for zKillboard (public PvP stats), and also exposes set_client() for test injection.

SDE — the Static Data Export (optional, offline)

sde.py backs offline name/type lookups from a local SQLite copy of EVE's Static Data Export (the Fuzzwork conversion).

  • The store is entirely optional. connect() opens the SQLite file lazily and caches the connection; a missing file yields None rather than raising, and available() reports False.
  • When present, get_item / get_system / get_station serve from it without a network round-trip; familiar falls back to ESI whenever the SDE is absent or doesn't know the id.
  • Only three tables are read — invTypes, mapSolarSystems, staStations — and only a handful of columns each, so a full official conversion works as well as a hand-built subset.
  • download_sde() streams and bz2-decompresses the dump incrementally (so arbitrarily large files work) and resets the cached connection on success.

Authentication (multi-character)

auth.py implements EVE SSO with PKCE (no client secret):

  • interactive_login() builds the authorize URL, opens the browser, runs a tiny localhost HTTP server to capture the code on the callback, checks the state (CSRF guard), exchanges the code, and persists the result.
  • Credentials are a dict of characters keyed by id, stored in one Fernet-encrypted file. A legacy single-character file is migrated on load.
  • get_auth(character_id) returns a valid access token, transparently refreshing via the rotating refresh token when the current one is within 60s of expiry.
  • Helpers: load_store / list_characters / remove_character / resolve_character (name-or-id → id).

See Authentication & Scopes for the full flow and scope list.

The per-character / per-corp pattern

charutil.py is the backbone of every authenticated tool. Authenticated reads take an optional character argument; the helper resolves what that means:

  • run_per_character(character, fetch) — runs fetch for one named character or all logged-in ones (when character is None), returning a uniform {"characters": [...]}.
  • run_per_corp(character, fetch) — same, but additionally resolves each character's corporation_id first (corp endpoints 403 without the right corp role; that surfaces per-character).
  • one_token(character) — picks a single (access_token, character_id) for thing-scoped lookups (e.g. one structure).
  • character_targets(character) — (id, name) pairs without needing a token (e.g. public zKill lookups that still default to the logged-in characters).
  • summarize(items, limit) — uniform {count, sample} payloads.

Write safety — the choke point

writes.py exposes guard(action, params, confirm, execute), the single gate every mutating tool calls. It is documented in detail on the Write Safety page. The rule for contributors: a write tool takes a required character and a confirm flag, and routes through writes.guard — never calling a mutating ESI endpoint directly.

Request flow (an authenticated read)

Claude → mcp.tool (e.g. get_wallet)
       → charutil.run_per_character(character, fetch)
       → auth.get_auth(character_id)        # refresh if near expiry
       → esi.client().get(..., token=...)   # cached by Expires; error-limit aware
       → {"characters": [ {character_id, character_name, ...}, ... ]}

Request flow (a write)

Claude → mcp.tool (e.g. set_waypoint, confirm=?)
       → writes.guard(action, params, confirm, execute)
            confirm=False              → dry-run preview, nothing happens
            confirm=True, writes off   → WriteDisabled
            confirm=True, writes on     → execute(), append to audit.log, return result