Phase 4: killmails + zKillboard (read)

Add read-only killmail tools: recent character and corp killmails (refs
enriched with system/victim/attacker detail) plus zKillboard PvP stats via
a tiny injectable zkill client. New charutil.character_targets helper for
token-less per-character iteration.

85 tests, coverage 93%. Closes #30.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/familiar/charutil.py

@@ -37,6 +37,16 @@ def _names() -> dict[int, str | None]:
     return {c["character_id"]: c["character_name"] for c in auth.list_characters()}
 
 
+def character_targets(character: str | int | None) -> list[tuple[int, str | None]]:
+    """``(character_id, name)`` pairs for one character or all — no token needed.
+
+    For public lookups keyed by character id (e.g. zKillboard) that still want
+    to default to the logged-in characters.
+    """
+    names = _names()
+    return [(cid, names.get(cid)) for cid in _targets(character)]
+
+
 async def run_per_character(character: str | int | None, fetch: CharFetcher) -> dict[str, Any]:
     """Run ``fetch`` for one character or all, returning a uniform list."""
     names = _names()

src/familiar/killmails.py

@@ -0,0 +1,81 @@
+"""Read-only killmail tools: recent kills (character + corp) and zKill stats."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastmcp import FastMCP
+
+from . import zkill
+from .charutil import character_targets, run_per_character, run_per_corp
+from .esi import client
+
+
+async def _enrich(refs: list[dict[str, Any]], limit: int) -> list[dict[str, Any]]:
+    """Turn killmail (id, hash) refs into trimmed detail for the most recent few."""
+    out = []
+    for ref in refs[: max(1, limit)]:
+        km = await client().get(f"/killmails/{ref['killmail_id']}/{ref['killmail_hash']}/")
+        victim = km.get("victim", {})
+        out.append(
+            {
+                "killmail_id": km.get("killmail_id"),
+                "time": km.get("killmail_time"),
+                "solar_system_id": km.get("solar_system_id"),
+                "victim_ship_type_id": victim.get("ship_type_id"),
+                "victim_character_id": victim.get("character_id"),
+                "attacker_count": len(km.get("attackers") or []),
+            }
+        )
+    return out
+
+
+async def get_killmails(character: str | int | None = None, limit: int = 10) -> dict[str, Any]:
+    """Recent killmails for one character or all (most recent enriched with detail)."""
+
+    async def fetch(token: str, cid: int) -> dict[str, Any]:
+        refs = await client().get_all_pages(f"/characters/{cid}/killmails/recent/", token=token)
+        return {"count": len(refs), "recent": await _enrich(refs, limit)}
+
+    return await run_per_character(character, fetch)
+
+
+async def get_corp_killmails(character: str | int | None = None, limit: int = 10) -> dict[str, Any]:
+    """Recent corp killmails (needs a corp role) for one character's corp or all."""
+
+    async def fetch(token: str, corp_id: int, cid: int) -> dict[str, Any]:
+        refs = await client().get_all_pages(
+            f"/corporations/{corp_id}/killmails/recent/", token=token
+        )
+        return {"count": len(refs), "recent": await _enrich(refs, limit)}
+
+    return await run_per_corp(character, fetch)
+
+
+async def get_zkill_stats(character: str | int | None = None) -> dict[str, Any]:
+    """PvP stats from zKillboard for one character or all (public, no auth)."""
+    results = []
+    for cid, name in character_targets(character):
+        stats = await zkill.client().get(f"/stats/characterID/{cid}/") or {}
+        results.append(
+            {
+                "character_id": cid,
+                "character_name": name,
+                "ships_destroyed": stats.get("shipsDestroyed"),
+                "ships_lost": stats.get("shipsLost"),
+                "isk_destroyed": stats.get("iskDestroyed"),
+                "isk_lost": stats.get("iskLost"),
+                "danger_ratio": stats.get("dangerRatio"),
+                "gang_ratio": stats.get("gangRatio"),
+            }
+        )
+    return {"characters": results}
+
+
+TOOLS = [get_killmails, get_corp_killmails, get_zkill_stats]
+
+
+def register(mcp: FastMCP) -> None:
+    """Register the killmail tools onto the given server."""
+    for fn in TOOLS:
+        mcp.tool(fn)

src/familiar/server.py

@@ -11,7 +11,17 @@ import datetime as _dt
 
 from fastmcp import FastMCP
 
-from . import __version__, account, industry, market, structures, trade, universe, writes
+from . import (
+    __version__,
+    account,
+    industry,
+    killmails,
+    market,
+    structures,
+    trade,
+    universe,
+    writes,
+)
 
 mcp: FastMCP = FastMCP(
     name="familiar",
@@ -54,3 +64,6 @@ structures.register(mcp)
 
 # Phase 4: write-action safety (the gate every write tool goes through).
 writes.register(mcp)
+
+# Phase 4: killmails + zKillboard (read-only).
+killmails.register(mcp)

src/familiar/zkill.py

@@ -0,0 +1,46 @@
+"""Tiny client for the public zKillboard API (kill/PvP stats).
+
+Same shape as the ESI client so tests can inject an ``httpx.MockTransport``.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import httpx
+
+from . import __version__
+
+ZKILL_BASE = os.environ.get("FAMILIAR_ZKILL_BASE", "https://zkillboard.com/api")
+USER_AGENT = f"familiar/{__version__} (+https://git.hiddenden.cafe/Hiddenden/familiar)"
+
+
+class ZkillClient:
+    def __init__(self, base_url: str = ZKILL_BASE, *, transport=None) -> None:
+        self._client = httpx.AsyncClient(
+            base_url=base_url,
+            transport=transport,
+            timeout=30.0,
+            headers={"User-Agent": USER_AGENT, "Accept": "application/json"},
+        )
+
+    async def get(self, path: str) -> Any:
+        resp = await self._client.get(path)
+        resp.raise_for_status()
+        return resp.json() if resp.content else None
+
+
+_default: ZkillClient | None = None
+
+
+def client() -> ZkillClient:
+    global _default
+    if _default is None:
+        _default = ZkillClient()
+    return _default
+
+
+def set_client(new: ZkillClient | None) -> None:
+    global _default
+    _default = new

tests/test_killmails.py

@@ -0,0 +1,96 @@
+"""Killmail tools (ESI char + corp) and zKillboard stats."""
+
+import asyncio
+
+import httpx
+import pytest
+
+from conftest import route
+from familiar import auth, killmails, zkill
+from familiar.server import mcp
+
+CHARS = [
+    {"character_id": 99, "character_name": "Alpha", "scopes": []},
+    {"character_id": 100, "character_name": "Beta", "scopes": []},
+]
+
+
+@pytest.fixture
+def fake_auth(monkeypatch):
+    monkeypatch.setattr(auth, "list_characters", lambda: list(CHARS))
+
+    async def get_auth(cid, **kw):
+        return f"token-{cid}", cid
+
+    monkeypatch.setattr(auth, "get_auth", get_auth)
+
+    def resolve(identifier):
+        for c in CHARS:
+            if (
+                str(c["character_id"]) == str(identifier)
+                or (c["character_name"] or "").lower() == str(identifier).lower()
+            ):
+                return c["character_id"]
+        raise auth.NotAuthenticated("no such character")
+
+    monkeypatch.setattr(auth, "resolve_character", resolve)
+    return CHARS
+
+
+def test_get_killmails_enriches(mock_esi, fake_auth):
+    detail = {
+        "killmail_id": 1,
+        "killmail_time": "2026-06-01T00:00:00Z",
+        "solar_system_id": 30000142,
+        "victim": {"ship_type_id": 670, "character_id": 99},
+        "attackers": [{}, {}, {}],
+    }
+    fixtures = {
+        "/killmails/recent/": [{"killmail_id": 1, "killmail_hash": "abc"}],
+        "/killmails/": detail,
+    }
+    mock_esi(lambda r: route(r, fixtures))
+    res = asyncio.run(killmails.get_killmails(character="Alpha", limit=5))
+    entry = res["characters"][0]
+    assert entry["count"] == 1
+    assert entry["recent"][0]["attacker_count"] == 3
+    assert entry["recent"][0]["solar_system_id"] == 30000142
+
+
+def test_get_corp_killmails(mock_esi, fake_auth):
+    detail = {"killmail_id": 9, "victim": {"ship_type_id": 1}, "attackers": [{}]}
+    fixtures = {
+        "/killmails/recent/": [{"killmail_id": 9, "killmail_hash": "h"}],
+        "/killmails/": detail,
+        "/characters/": {"corporation_id": 2000},
+    }
+    mock_esi(lambda r: route(r, fixtures))
+    res = asyncio.run(killmails.get_corp_killmails(character=99))
+    assert res["characters"][0]["corporation_id"] == 2000
+    assert res["characters"][0]["count"] == 1
+
+
+@pytest.fixture
+def mock_zkill():
+    def _make(payload):
+        zkill.set_client(
+            zkill.ZkillClient(
+                transport=httpx.MockTransport(lambda r: httpx.Response(200, json=payload))
+            )
+        )
+
+    yield _make
+    zkill.set_client(None)
+
+
+def test_zkill_stats(fake_auth, mock_zkill):
+    mock_zkill({"shipsDestroyed": 100, "shipsLost": 5, "dangerRatio": 95})
+    res = asyncio.run(killmails.get_zkill_stats())
+    assert len(res["characters"]) == 2
+    assert res["characters"][0]["ships_destroyed"] == 100
+    assert res["characters"][0]["danger_ratio"] == 95
+
+
+def test_killmail_tools_registered():
+    tools = {t.name for t in asyncio.run(mcp.list_tools())}
+    assert {"get_killmails", "get_corp_killmails", "get_zkill_stats"} <= tools