Phase 4: planetary interaction reads (#29)
Add read-only PI tools: get_planets (colonies), get_planet_layout (pins/links/routes), get_corp_customs_offices. ESI exposes no PI write endpoint despite the manage_planets scope name, so these are reads. Closes #29. 132 tests, coverage 90%, 84 tools. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
3 files changed
+119
No files matched your search
@@ -0,0 +1,62 @@
|
|||||||
|
"""Read-only planetary interaction tools.
|
||||||
|
|
||||||
|
Despite the ``manage_planets`` scope name, ESI exposes only *reads* for PI, so
|
||||||
|
these tools list colonies and their layouts; there is no PI write endpoint.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from fastmcp import FastMCP
|
||||||
|
|
||||||
|
from . import auth
|
||||||
|
from .charutil import run_per_character, run_per_corp, summarize
|
||||||
|
from .esi import client
|
||||||
|
|
||||||
|
|
||||||
|
async def get_planets(character: str | int | None = None, limit: int = 20) -> dict[str, Any]:
|
||||||
|
"""Planetary colonies for one character or all (planet, type, upgrade level)."""
|
||||||
|
|
||||||
|
async def fetch(token: str, cid: int) -> dict[str, Any]:
|
||||||
|
colonies = await client().get(f"/characters/{cid}/planets/", token=token) or []
|
||||||
|
return summarize(colonies, limit)
|
||||||
|
|
||||||
|
return await run_per_character(character, fetch)
|
||||||
|
|
||||||
|
|
||||||
|
async def get_planet_layout(character: str | int, planet_id: int) -> dict[str, Any]:
|
||||||
|
"""Full layout (pins, links, routes) of one colony for a character."""
|
||||||
|
cid = auth.resolve_character(character)
|
||||||
|
token, _ = await auth.get_auth(cid)
|
||||||
|
data = await client().get(f"/characters/{cid}/planets/{planet_id}/", token=token) or {}
|
||||||
|
return {
|
||||||
|
"character_id": cid,
|
||||||
|
"planet_id": planet_id,
|
||||||
|
"pins": len(data.get("pins") or []),
|
||||||
|
"links": len(data.get("links") or []),
|
||||||
|
"routes": len(data.get("routes") or []),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
async def get_corp_customs_offices(
|
||||||
|
character: str | int | None = None, limit: int = 20
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
"""Corp customs offices (POCOs) — needs a corp role — for one corp or all."""
|
||||||
|
|
||||||
|
async def fetch(token: str, corp_id: int, cid: int) -> dict[str, Any]:
|
||||||
|
offices = await client().get_all_pages(
|
||||||
|
f"/corporations/{corp_id}/customs_offices/", token=token
|
||||||
|
)
|
||||||
|
return summarize(offices, limit)
|
||||||
|
|
||||||
|
return await run_per_corp(character, fetch)
|
||||||
|
|
||||||
|
|
||||||
|
TOOLS = [get_planets, get_planet_layout, get_corp_customs_offices]
|
||||||
|
|
||||||
|
|
||||||
|
def register(mcp: FastMCP) -> None:
|
||||||
|
"""Register the planetary interaction tools onto the given server."""
|
||||||
|
for fn in TOOLS:
|
||||||
|
mcp.tool(fn)
|
||||||
@@ -24,6 +24,7 @@ from . import (
|
|||||||
killmails,
|
killmails,
|
||||||
market,
|
market,
|
||||||
misc,
|
misc,
|
||||||
|
planets,
|
||||||
profile,
|
profile,
|
||||||
structures,
|
structures,
|
||||||
trade,
|
trade,
|
||||||
@@ -99,3 +100,6 @@ fittings.register(mcp)
|
|||||||
# Phase 4: fleet (read + guarded writes) and UI commands (guarded writes).
|
# Phase 4: fleet (read + guarded writes) and UI commands (guarded writes).
|
||||||
fleet.register(mcp)
|
fleet.register(mcp)
|
||||||
ui.register(mcp)
|
ui.register(mcp)
|
||||||
|
|
||||||
|
# Phase 4: planetary interaction (read-only).
|
||||||
|
planets.register(mcp)
|
||||||
@@ -0,0 +1,53 @@
|
|||||||
|
"""Planetary interaction read tools."""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from conftest import route
|
||||||
|
from familiar import auth, planets
|
||||||
|
from familiar.server import mcp
|
||||||
|
|
||||||
|
CHARS = [{"character_id": 99, "character_name": "Alpha", "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)
|
||||||
|
monkeypatch.setattr(auth, "resolve_character", lambda i: 99)
|
||||||
|
return CHARS
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_planets(mock_esi, fake_auth):
|
||||||
|
cols = [{"planet_id": 40000001, "planet_type": "barren", "upgrade_level": 3}]
|
||||||
|
mock_esi(lambda r: route(r, {"/planets/": cols}))
|
||||||
|
res = asyncio.run(planets.get_planets())
|
||||||
|
assert res["characters"][0]["count"] == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_planet_layout(mock_esi, fake_auth):
|
||||||
|
layout = {"pins": [{}, {}], "links": [{}], "routes": [{}, {}, {}]}
|
||||||
|
mock_esi(lambda r: route(r, {"/planets/40000001/": layout}))
|
||||||
|
res = asyncio.run(planets.get_planet_layout(99, 40000001))
|
||||||
|
assert res["pins"] == 2
|
||||||
|
assert res["routes"] == 3
|
||||||
|
|
||||||
|
|
||||||
|
def test_corp_customs_offices(mock_esi, fake_auth):
|
||||||
|
fixtures = {
|
||||||
|
"/customs_offices/": [{"office_id": 1}],
|
||||||
|
"/characters/": {"corporation_id": 2000},
|
||||||
|
}
|
||||||
|
mock_esi(lambda r: route(r, fixtures))
|
||||||
|
res = asyncio.run(planets.get_corp_customs_offices())
|
||||||
|
assert res["characters"][0]["count"] == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_planets_tools_registered():
|
||||||
|
tools = {t.name for t in asyncio.run(mcp.list_tools())}
|
||||||
|
assert {"get_planets", "get_planet_layout", "get_corp_customs_offices"} <= tools
|
||||||
Reference in new issue
Block a user