From 7f62d70619218cb2ccaa30358327299c47502cc9 Mon Sep 17 00:00:00 2001 From: Lee Chapman Date: Thu, 5 Mar 2026 00:42:52 -0800 Subject: [PATCH 1/2] feat: outcome-based and analytics MCP tools - Add analytics module (mission, war, planet analytics, query_stats) - Add outcomes module (war summary, where to deploy, liberation priority, efficiency) - Add get_raw_api, outcome tools (4), analytics tools (4); keep 7 raw tools - Standard agent-friendly envelope (status, outcome, summary, data, metrics) - Update server list_tools/call_tool for 16 tools; add tests Made-with: Cursor --- docs/MCP_TOOLS_REVIEW.md | 148 ++++++++++++++++++ highcommand/analytics.py | 267 +++++++++++++++++++++++++++++++ highcommand/outcomes.py | 192 +++++++++++++++++++++++ highcommand/server.py | 156 +++++++++++++++---- highcommand/tools.py | 302 ++++++++++++++++++++++++++++++++++++ tests/demo_all_endpoints.py | 2 +- tests/test_server.py | 45 +++++- 7 files changed, 1078 insertions(+), 34 deletions(-) create mode 100644 docs/MCP_TOOLS_REVIEW.md create mode 100644 highcommand/analytics.py create mode 100644 highcommand/outcomes.py diff --git a/docs/MCP_TOOLS_REVIEW.md b/docs/MCP_TOOLS_REVIEW.md new file mode 100644 index 0000000..4deb6ef --- /dev/null +++ b/docs/MCP_TOOLS_REVIEW.md @@ -0,0 +1,148 @@ +# MCP Tools Review: Outcome-Based, Agent-Friendly, and Analytics + +**Goal**: Tools that are outcome-based and agent-friendly, with analytics so users can ask questions about raw stats and API data and complete goals/missions more efficiently. + +--- + +## 1. Current State + +### Existing tools (endpoint-centric, raw data) + +| Tool | Purpose | Agent-friendly? | Outcome-based? | +|------|---------|-----------------|----------------| +| `get_war_status` | Raw war status JSON | Low – agent must interpret | No | +| `get_planets` | Raw planets list | Low | No | +| `get_statistics` | Raw global stats | Low | No | +| `get_campaign_info` | Raw active campaigns | Low | No | +| `get_planet_status` | Raw single-planet status | Low | No | +| `get_biomes` | Raw biomes | Low | No | +| `get_factions` | Raw factions | Low | No | + +**Gaps**: + +- **Outcome-based**: No tools that answer “what should I do?” or “what’s the outcome?” (e.g. where to deploy, war summary, priority missions). +- **Analytics**: No derived metrics for efficiency (success rates, best sectors, mission effectiveness, time-to-liberate). +- **Queryable raw data**: Raw JSON is returned but there’s no structured “ask a question about stats/messages” surface—agents get dumps and must parse everything themselves. + +--- + +## 2. Design Principles + +### Outcome-based + +- **Input**: Minimal or high-level (e.g. “current war”, “where to help”). +- **Output**: Clear **outcome** or **recommendation**, not raw payloads. +- **Examples**: “War summary”, “Planets that need reinforcements”, “Best missions for impact”, “Liberation priority list”. + +### Agent-friendly + +- **Stable, documented response shape**: Same top-level keys (`outcome`, `summary`, `recommendations`, `data`, `error`) so agents can branch on success/error and use fields reliably. +- **Structured text + optional structured data**: Short `summary` for LLM context; optional `data` for programmatic use. +- **Explicit semantics**: Tool names and descriptions state the outcome (e.g. “Get planets that need reinforcements” not “Get campaign info”). +- **Single-call outcomes**: One tool call returns a complete answer when possible, reducing round-trips and reasoning load. + +### Analytics for efficient goals/missions + +- **Derived metrics** from raw API data: + - Mission efficiency: success rate, time per mission, kills per hour. + - War progress: liberation trend, sector pressure, campaign completion. + - Impact: which planets/campaigns contribute most to war progress. +- **Answers to questions** like: + - “How efficiently are we completing missions?” (success rate, time, etc.) + - “Which sectors need the most help?” + - “What do the latest stats say about bug vs automaton kills?” + +--- + +## 3. Proposed Tool Layout + +### A. Keep: Raw data access (for “questions on raw stats and messages”) + +Keep existing tools as the **raw API layer**, but clarify their role: + +- **Purpose**: “Get raw stats and API messages so the user/agent can ask arbitrary questions.” +- **Naming**: Optional rename for clarity (e.g. keep `get_statistics` but describe as “Raw global statistics from the API for custom analysis”). + +Add one optional **query-style** tool so agents can request “one place” for raw data: + +| Tool | Description | Parameters | Returns | +|------|-------------|------------|--------| +| `get_raw_api` | Return raw API response for one known endpoint. Use when the user asks about raw stats or API messages. | `endpoint`: one of `war/status`, `planets`, `statistics`, `campaigns/active`, `biomes`, `factions`; optional `planet_index` for planet detail | `{ "status", "data", "error", "endpoint" }` | + +This gives a single, consistent way to “ask questions on raw stats and messages” without adding many new endpoints. + +### B. New: Outcome-based tools + +High-level answers; one call = one outcome. + +| Tool | Description | Parameters | Returns (conceptual) | +|------|-------------|------------|----------------------| +| `get_war_summary` | Human-readable war summary and current phase. Outcome: “What’s the state of the war?” | None | `outcome`, `summary`, `war_id`, `phase`, `ends_at`, optional `data` | +| `get_where_to_deploy` | Planets (or campaigns) that need reinforcements most. Outcome: “Where should I deploy?” | Optional `limit` | `outcome`, `summary`, `recommendations` (list of planet/campaign + reason), optional `data` | +| `get_liberation_priority` | Ordered list of planets by liberation priority (e.g. by health, campaign count, sector). Outcome: “What to liberate first?” | Optional `limit`, `sector` | `outcome`, `summary`, `priorities` (list), optional `data` | +| `get_mission_efficiency_snapshot` | Current mission efficiency (from global stats): success rate, time played, kills. Outcome: “How are we doing on missions?” | None | `outcome`, `summary`, `success_rate`, `missions_won/lost`, `time_played`, optional `data` | + +All return a **stable envelope**: `outcome` (e.g. "ok" / "no_data"), `summary` (text), then outcome-specific fields, plus optional `data` for raw-ish detail. + +### C. New: Analytics tools + +Derived metrics and answers for “efficiently complete goals and missions” and “questions on raw stats”. + +| Tool | Description | Parameters | Returns (conceptual) | +|------|-------------|------------|----------------------| +| `get_mission_analytics` | Derived mission analytics: success rate, missions won/lost, mission time, kills breakdown (bugs/automatons/illuminate). Use for “efficiency” and “raw stats” questions. | None | `outcome`, `summary`, `success_rate`, `missions_won`, `missions_lost`, `mission_time`, `kills`, optional `data` (raw stats slice) | +| `get_war_analytics` | War-level analytics: time left, progress indicators (if API supports), active campaigns count, planets under attack. | None | `outcome`, `summary`, `time_left`, `active_campaigns`, optional `data` | +| `get_planet_analytics` | Per-planet or aggregate planet analytics: e.g. count by sector, by owner, under attack. Enables “which sectors need help?” | Optional `sector`, `group_by` | `outcome`, `summary`, `by_sector` / `by_owner`, optional `data` | +| `query_stats` | Answer a simple stats question from global statistics. Accepts a question type or key (e.g. “mission_success_rate”, “bug_kills”, “accuracy”). Use for “ask questions on raw stats”. | `question` or `metric`: string (e.g. "mission_success_rate", "deaths", "time_played") | `outcome`, `answer` (text), `value`, `unit`, optional `data` | + +Implementations can map `question`/`metric` to known fields in the statistics response so agents can ask “what’s mission success rate?” and get a single number + short answer. + +--- + +## 4. Response Envelope (agent-friendly) + +Use one envelope for all outcome and analytics tools: + +```json +{ + "status": "success", + "outcome": "ok", + "summary": "One-line or short paragraph for the LLM.", + "data": { ... }, + "error": null, + "metrics": { "elapsed_ms": 12 } +} +``` + +For outcome tools, add outcome-specific fields at the top level (e.g. `recommendations`, `priorities`, `success_rate`) so agents don’t have to dig into `data` for common use cases. Keep `data` for raw or extended payloads so users can still “ask questions on raw stats and messages” when needed. + +--- + +## 5. Implementation Outline + +1. **Keep existing 7 tools** as the raw layer; document them as “raw API data for custom and stats questions.” +2. **Add `get_raw_api`** (optional): single entry point for raw API by `endpoint` (+ optional `planet_index`). +3. **Add analytics module** (e.g. `highcommand/analytics.py`): + - `mission_analytics(raw_stats) -> dict` + - `war_analytics(war_status, campaigns, planets?) -> dict` + - `planet_analytics(planets, campaigns?) -> dict` + - `query_stat_metric(raw_stats, metric_key) -> { answer, value, unit }` +4. **Add outcome module** (e.g. `highcommand/outcomes.py`): + - `war_summary(war_status) -> { outcome, summary, ... }` + - `where_to_deploy(campaigns, planets, planet_statuses?) -> { outcome, summary, recommendations }` + - `liberation_priority(planets, campaigns?, planet_statuses?) -> { outcome, summary, priorities }` + - `mission_efficiency_snapshot(statistics) -> { outcome, summary, success_rate, ... }` +5. **Wire in server**: Register new tools in `server.py` and implement handlers in `tools.py` (or a dedicated `outcome_tools.py` / `analytics_tools.py`) that call the API client, then analytics/outcomes, and return the standard envelope. +6. **Tool registry**: Register new tools in `ToolRegistry` with clear names and descriptions so Cursor/agents see outcome-based descriptions. + +--- + +## 6. Summary + +| Category | Role | Example tools | +|----------|------|----------------| +| **Raw** | “Ask questions on raw stats and messages” | Existing 7 tools + optional `get_raw_api` | +| **Outcome** | “What should I do?” / “What’s the outcome?” | `get_war_summary`, `get_where_to_deploy`, `get_liberation_priority`, `get_mission_efficiency_snapshot` | +| **Analytics** | “How to efficiently complete goals and missions” + stats questions | `get_mission_analytics`, `get_war_analytics`, `get_planet_analytics`, `query_stats` | + +This keeps raw access for power users and agents while adding outcome-based and analytics tools that are agent-friendly and support efficient completion of goals and missions. diff --git a/highcommand/analytics.py b/highcommand/analytics.py new file mode 100644 index 0000000..caa7846 --- /dev/null +++ b/highcommand/analytics.py @@ -0,0 +1,267 @@ +"""Analytics module: derived metrics for mission efficiency and war/planet stats.""" + +from typing import Any + +# Metric keys supported by query_stat_metric (map to statistics payload keys) +STAT_METRIC_KEYS = { + "mission_success_rate": ("missionSuccessRate", "%", "Mission success rate (percentage)"), + "missions_won": ("missionsWon", "", "Total missions won"), + "missions_lost": ("missionsLost", "", "Total missions lost"), + "mission_time": ("missionTime", "s", "Total mission time (seconds)"), + "time_played": ("timePlayed", "s", "Total time played (seconds)"), + "bug_kills": ("bugKills", "", "Terminid/bug kills"), + "automaton_kills": ("automatonKills", "", "Automaton kills"), + "illuminate_kills": ("illuminateKills", "", "Illuminate kills"), + "bullets_fired": ("bulletsFired", "", "Bullets fired"), + "bullets_hit": ("bulletsHit", "", "Bullets hit"), + "accuracy": ("accuracy", "%", "Accuracy (percentage)"), + "deaths": ("deaths", "", "Total deaths"), + "revives": ("revives", "", "Revives"), + "friendly_kills": ("friendlyKills", "", "Friendly kills"), +} + + +def _get_data(payload: dict[str, Any]) -> Any: + """Extract data from API response envelope.""" + if payload is None: + return None + return payload.get("data") if isinstance(payload, dict) else payload + + +def mission_analytics(raw_stats_response: dict[str, Any]) -> dict[str, Any]: + """Derive mission analytics from global statistics API response. + + Returns outcome, summary, success_rate, missions_won/lost, mission_time, kills. + """ + data = _get_data(raw_stats_response) + if not data: + return { + "outcome": "no_data", + "summary": "No statistics data available.", + "success_rate": None, + "missions_won": None, + "missions_lost": None, + "mission_time": None, + "kills": None, + "data": raw_stats_response, + } + + # API may return list of one stats object or single object + if isinstance(data, list) and len(data) > 0: + stats = data[0] + elif isinstance(data, dict): + stats = data + else: + return { + "outcome": "no_data", + "summary": "Statistics format not recognized.", + "data": raw_stats_response, + } + + missions_won = stats.get("missionsWon", 0) or 0 + missions_lost = stats.get("missionsLost", 0) or 0 + total = missions_won + missions_lost + success_rate = round(100 * missions_won / total, 1) if total else None + + return { + "outcome": "ok", + "summary": ( + f"Mission success rate: {success_rate}% ({missions_won:,} won, {missions_lost:,} lost). " + f"Total mission time: {stats.get('missionTime', 0):,}s. " + f"Kills: Bugs {stats.get('bugKills', 0):,}, Automatons {stats.get('automatonKills', 0):,}, Illuminate {stats.get('illuminateKills', 0):,}." + ), + "success_rate": success_rate, + "missions_won": missions_won, + "missions_lost": missions_lost, + "mission_time": stats.get("missionTime"), + "time_played": stats.get("timePlayed"), + "kills": { + "bugKills": stats.get("bugKills"), + "automatonKills": stats.get("automatonKills"), + "illuminateKills": stats.get("illuminateKills"), + }, + "data": stats, + } + + +def war_analytics( + war_status_response: dict[str, Any], + campaigns_response: dict[str, Any] | None = None, + planets_response: dict[str, Any] | None = None, +) -> dict[str, Any]: + """War-level analytics: time left, active campaigns count, high-level progress.""" + war_data = _get_data(war_status_response) + campaigns_data = _get_data(campaigns_response) if campaigns_response else None + planets_data = _get_data(planets_response) if planets_response else None + + if not war_data and not campaigns_data: + return { + "outcome": "no_data", + "summary": "No war or campaign data available.", + "time_left": None, + "active_campaigns": None, + "data": { + "war": war_data, + "campaigns": campaigns_data, + }, + } + + active_campaigns = 0 + if campaigns_data is not None: + active_campaigns = len(campaigns_data) if isinstance(campaigns_data, list) else (1 if campaigns_data else 0) + + time_left = None + war_id = None + if isinstance(war_data, dict): + war_id = war_data.get("id") or war_data.get("index") + end_date = war_data.get("endDate") + if end_date: + try: + from datetime import datetime, timezone + if isinstance(end_date, str): + end = datetime.fromisoformat(end_date.replace("Z", "+00:00")) + else: + end = end_date + now = datetime.now(timezone.utc) + if end.tzinfo is None: + end = end.replace(tzinfo=timezone.utc) + delta = end - now + time_left = max(0, int(delta.total_seconds())) + except Exception: + pass + + summary_parts = [] + if war_id is not None: + summary_parts.append(f"War {war_id} active.") + if time_left is not None: + days = time_left // 86400 + summary_parts.append(f"Time remaining: {days} days.") + summary_parts.append(f"Active campaigns: {active_campaigns}.") + summary = " ".join(summary_parts) + + return { + "outcome": "ok", + "summary": summary, + "time_left_seconds": time_left, + "active_campaigns": active_campaigns, + "war_id": war_id, + "data": { + "war": war_data, + "campaigns": campaigns_data, + "planets_count": len(planets_data) if isinstance(planets_data, list) else None, + }, + } + + +def planet_analytics( + planets_response: dict[str, Any], + campaigns_response: dict[str, Any] | None = None, + sector: str | None = None, + group_by: str | None = None, +) -> dict[str, Any]: + """Per-planet or aggregate planet analytics: by sector, by owner, etc.""" + planets_data = _get_data(planets_response) + campaigns_data = _get_data(campaigns_response) if campaigns_response else None + + if not planets_data or not isinstance(planets_data, list): + return { + "outcome": "no_data", + "summary": "No planet data available.", + "by_sector": {}, + "by_owner": {}, + "data": planets_data, + } + + campaign_planet_indices = set() + if isinstance(campaigns_data, list): + for c in campaigns_data: + if isinstance(c, dict) and "planet" in c: + campaign_planet_indices.add(c["planet"]) + + by_sector: dict[str, int] = {} + by_owner: dict[str, int] = {} + filtered = planets_data + if sector: + filtered = [p for p in planets_data if isinstance(p, dict) and p.get("sector") == sector] + + for p in filtered: + if not isinstance(p, dict): + continue + sec = p.get("sector") or "Unknown" + by_sector[sec] = by_sector.get(sec, 0) + 1 + owner = "Unknown" + if "status" in p and isinstance(p["status"], dict): + owner = p["status"].get("owner") or owner + by_owner[owner] = by_owner.get(owner, 0) + 1 + + under_attack = [p for p in filtered if isinstance(p, dict) and p.get("index") in campaign_planet_indices] + summary = ( + f"{len(filtered)} planets total. " + f"Sectors: {len(by_sector)}. " + f"{len(under_attack)} planets with active campaigns." + ) + + result = { + "outcome": "ok", + "summary": summary, + "by_sector": by_sector, + "by_owner": by_owner, + "planets_with_campaigns": len(under_attack), + "data": {"count": len(filtered), "by_sector": by_sector, "by_owner": by_owner}, + } + if group_by == "sector": + result["grouped"] = by_sector + elif group_by == "owner": + result["grouped"] = by_owner + return result + + +def query_stat_metric(raw_stats_response: dict[str, Any], metric_key: str) -> dict[str, Any]: + """Answer a single stats question. metric_key e.g. mission_success_rate, deaths, time_played.""" + data = _get_data(raw_stats_response) + if not data: + return { + "outcome": "no_data", + "answer": "No statistics data available.", + "value": None, + "unit": None, + "data": None, + } + + if isinstance(data, list) and len(data) > 0: + stats = data[0] + elif isinstance(data, dict): + stats = data + else: + return { + "outcome": "no_data", + "answer": "Statistics format not recognized.", + "value": None, + "unit": None, + "data": None, + } + + key_lower = metric_key.strip().lower().replace(" ", "_") + if key_lower not in STAT_METRIC_KEYS: + valid = ", ".join(STAT_METRIC_KEYS.keys()) + return { + "outcome": "unknown_metric", + "answer": f"Unknown metric '{metric_key}'. Valid metrics: {valid}.", + "value": None, + "unit": None, + "data": None, + } + + api_key, unit, description = STAT_METRIC_KEYS[key_lower] + value = stats.get(api_key) + if value is None: + value = stats.get(api_key) # try as-is + + return { + "outcome": "ok", + "answer": f"{description}: {value} {unit}".strip() if value is not None else f"{description}: no value", + "value": value, + "unit": unit, + "metric": key_lower, + "data": {api_key: value}, + } diff --git a/highcommand/outcomes.py b/highcommand/outcomes.py new file mode 100644 index 0000000..46956f3 --- /dev/null +++ b/highcommand/outcomes.py @@ -0,0 +1,192 @@ +"""Outcome module: high-level answers (war summary, where to deploy, liberation priority, efficiency).""" + +from typing import Any + + +def _get_data(payload: dict[str, Any] | None) -> Any: + if payload is None: + return None + return payload.get("data") if isinstance(payload, dict) else payload + + +def war_summary(war_status_response: dict[str, Any]) -> dict[str, Any]: + """Human-readable war summary and current phase. Outcome: What's the state of the war?""" + war_data = _get_data(war_status_response) + if not war_data or not isinstance(war_data, dict): + return { + "outcome": "no_data", + "summary": "No war status data available.", + "war_id": None, + "phase": None, + "ends_at": None, + "data": war_data, + } + + war_id = war_data.get("id") or war_data.get("index") + phase = "active" # API may not expose phase; default + ends_at = war_data.get("endDate") + summary = f"War {war_id} is {phase}. End date: {ends_at}." + return { + "outcome": "ok", + "summary": summary, + "war_id": war_id, + "phase": phase, + "ends_at": ends_at, + "data": war_data, + } + + +def where_to_deploy( + campaigns_response: dict[str, Any], + planets_response: dict[str, Any], + limit: int = 10, +) -> dict[str, Any]: + """Planets/campaigns that need reinforcements most. Outcome: Where should I deploy?""" + campaigns_data = _get_data(campaigns_response) + planets_data = _get_data(planets_response) + + if not campaigns_data or not isinstance(campaigns_data, list): + return { + "outcome": "no_data", + "summary": "No active campaign data available.", + "recommendations": [], + "data": None, + } + + planets_list = planets_data if isinstance(planets_data, list) else [] + planet_by_index = {p.get("index"): p for p in planets_list if isinstance(p, dict)} + + recommendations = [] + for c in campaigns_data[: limit * 2]: # allow extra to fill limit + if not isinstance(c, dict): + continue + planet_index = c.get("planet") + if planet_index is None: + continue + p = planet_by_index.get(planet_index) + name = p.get("name", f"Planet {planet_index}") if p else f"Planet {planet_index}" + sector = p.get("sector", "Unknown") if p else "Unknown" + recommendations.append({ + "planet_index": planet_index, + "name": name, + "sector": sector, + "reason": "Active campaign", + }) + if len(recommendations) >= limit: + break + + summary = f"{len(recommendations)} planets with active campaigns need reinforcements." + if recommendations: + names = ", ".join(r["name"] for r in recommendations[:5]) + if len(recommendations) > 5: + names += f" and {len(recommendations) - 5} more" + summary += f" Top: {names}." + + return { + "outcome": "ok", + "summary": summary, + "recommendations": recommendations, + "data": {"campaigns": campaigns_data, "count": len(recommendations)}, + } + + +def liberation_priority( + planets_response: dict[str, Any], + campaigns_response: dict[str, Any] | None = None, + limit: int = 10, + sector: str | None = None, +) -> dict[str, Any]: + """Ordered list of planets by liberation priority. Outcome: What to liberate first?""" + planets_data = _get_data(planets_response) + campaigns_data = _get_data(campaigns_response) if campaigns_response else None + + if not planets_data or not isinstance(planets_data, list): + return { + "outcome": "no_data", + "summary": "No planet data available.", + "priorities": [], + "data": None, + } + + campaign_planet_indices = set() + if isinstance(campaigns_data, list): + for c in campaigns_data: + if isinstance(c, dict) and "planet" in c: + campaign_planet_indices.add(c["planet"]) + + # Build priority: planets with active campaigns first, then by sector filter + with_campaigns = [] + without = [] + for p in planets_data: + if not isinstance(p, dict): + continue + if sector and p.get("sector") != sector: + continue + idx = p.get("index") + name = p.get("name", f"Planet {idx}") + rec = {"planet_index": idx, "name": name, "sector": p.get("sector"), "has_campaign": idx in campaign_planet_indices} + if rec["has_campaign"]: + with_campaigns.append(rec) + else: + without.append(rec) + + priorities = with_campaigns + without[: max(0, limit - len(with_campaigns))] + priorities = priorities[:limit] + + summary = f"Top {len(priorities)} planets by liberation priority. {len(with_campaigns)} have active campaigns." + return { + "outcome": "ok", + "summary": summary, + "priorities": priorities, + "data": {"count": len(priorities), "with_campaigns": len(with_campaigns)}, + } + + +def mission_efficiency_snapshot(statistics_response: dict[str, Any]) -> dict[str, Any]: + """Current mission efficiency from global stats. Outcome: How are we doing on missions?""" + data = _get_data(statistics_response) + if not data: + return { + "outcome": "no_data", + "summary": "No statistics data available.", + "success_rate": None, + "missions_won": None, + "missions_lost": None, + "time_played": None, + "data": None, + } + + if isinstance(data, list) and len(data) > 0: + stats = data[0] + elif isinstance(data, dict): + stats = data + else: + return { + "outcome": "no_data", + "summary": "Statistics format not recognized.", + "success_rate": None, + "missions_won": None, + "missions_lost": None, + "time_played": None, + "data": None, + } + + missions_won = stats.get("missionsWon", 0) or 0 + missions_lost = stats.get("missionsLost", 0) or 0 + total = missions_won + missions_lost + success_rate = round(100 * missions_won / total, 1) if total else stats.get("missionSuccessRate") + + summary = ( + f"Mission success rate: {success_rate}%. " + f"Missions won: {missions_won:,}, lost: {missions_lost:,}. " + f"Time played: {stats.get('timePlayed', 0):,}s." + ) + return { + "outcome": "ok", + "summary": summary, + "success_rate": success_rate, + "missions_won": missions_won, + "missions_lost": missions_lost, + "time_played": stats.get("timePlayed"), + "data": stats, + } diff --git a/highcommand/server.py b/highcommand/server.py index 98da6c3..e34763d 100644 --- a/highcommand/server.py +++ b/highcommand/server.py @@ -30,74 +30,130 @@ @server.list_tools() async def list_tools() -> list[Tool]: - """List available MCP tools.""" + """List available MCP tools (raw, outcome-based, and analytics).""" return [ + # ----- Raw API (for custom analysis and raw stats questions) ----- Tool( name="get_war_status", - description="Get current war status from High-Command API", - inputSchema={ - "type": "object", - "properties": {}, - "required": [], - }, + description="Raw war status from API. Use for custom analysis or when user asks for raw war data.", + inputSchema={"type": "object", "properties": {}, "required": []}, ), Tool( name="get_planets", - description="Get planet information from High-Command API", + description="Raw planet list from API. Use for custom analysis or when user asks for raw planet data.", + inputSchema={"type": "object", "properties": {}, "required": []}, + ), + Tool( + name="get_statistics", + description="Raw global statistics from API. Use for custom analysis or when user asks for raw stats.", + inputSchema={"type": "object", "properties": {}, "required": []}, + ), + Tool( + name="get_campaign_info", + description="Raw active campaigns from API. Use for custom analysis.", + inputSchema={"type": "object", "properties": {}, "required": []}, + ), + Tool( + name="get_planet_status", + description="Raw status for a specific planet by index.", inputSchema={ "type": "object", - "properties": {}, - "required": [], + "properties": {"planet_index": {"type": "integer", "description": "The index of the planet"}}, + "required": ["planet_index"], }, ), Tool( - name="get_statistics", - description="Get global game statistics from High-Command API", + name="get_biomes", + description="Raw biome data from API.", + inputSchema={"type": "object", "properties": {}, "required": []}, + ), + Tool( + name="get_factions", + description="Raw faction data from API.", + inputSchema={"type": "object", "properties": {}, "required": []}, + ), + Tool( + name="get_raw_api", + description="Return raw API response for one endpoint. Use when user asks about raw stats or API messages. Endpoint: war/status, planets, statistics, campaigns/active, biomes, factions.", inputSchema={ "type": "object", - "properties": {}, - "required": [], + "properties": { + "endpoint": { + "type": "string", + "description": "One of: war/status, planets, statistics, campaigns/active, biomes, factions", + }, + "planet_index": {"type": "integer", "description": "Optional; for planets endpoint, fetch this planet's detail"}, + }, + "required": ["endpoint"], }, ), + # ----- Outcome-based tools ----- Tool( - name="get_campaign_info", - description="Get campaign information from High-Command API", + name="get_war_summary", + description="Human-readable war summary and current phase. Use when user asks: What's the state of the war?", + inputSchema={"type": "object", "properties": {}, "required": []}, + ), + Tool( + name="get_where_to_deploy", + description="Planets that need reinforcements most. Use when user asks: Where should I deploy?", inputSchema={ "type": "object", - "properties": {}, + "properties": {"limit": {"type": "integer", "description": "Max number of recommendations (default 10)"}}, "required": [], }, ), Tool( - name="get_planet_status", - description="Get status for a specific planet", + name="get_liberation_priority", + description="Ordered list of planets by liberation priority. Use when user asks: What to liberate first?", inputSchema={ "type": "object", "properties": { - "planet_index": { - "type": "integer", - "description": "The index of the planet", - } + "limit": {"type": "integer", "description": "Max number of planets (default 10)"}, + "sector": {"type": "string", "description": "Filter by sector name"}, }, - "required": ["planet_index"], + "required": [], }, ), Tool( - name="get_biomes", - description="Get biome information from High-Command API", + name="get_mission_efficiency_snapshot", + description="Current mission efficiency (success rate, time, kills). Use when user asks: How are we doing on missions?", + inputSchema={"type": "object", "properties": {}, "required": []}, + ), + # ----- Analytics tools ----- + Tool( + name="get_mission_analytics", + description="Derived mission analytics: success rate, missions won/lost, mission time, kills breakdown. Use for efficiency or raw stats questions.", + inputSchema={"type": "object", "properties": {}, "required": []}, + ), + Tool( + name="get_war_analytics", + description="War-level analytics: time left, active campaigns count, progress. Use for war overview.", + inputSchema={"type": "object", "properties": {}, "required": []}, + ), + Tool( + name="get_planet_analytics", + description="Planet analytics by sector/owner and planets under attack. Use when user asks: Which sectors need help?", inputSchema={ "type": "object", - "properties": {}, + "properties": { + "sector": {"type": "string", "description": "Filter by sector"}, + "group_by": {"type": "string", "description": "sector or owner"}, + }, "required": [], }, ), Tool( - name="get_factions", - description="Get faction information from High-Command API", + name="query_stats", + description="Answer a single stats question. Use when user asks for one metric. Metric: mission_success_rate, missions_won, missions_lost, mission_time, time_played, bug_kills, automaton_kills, illuminate_kills, deaths, revives, accuracy, bullets_fired, bullets_hit, friendly_kills.", inputSchema={ "type": "object", - "properties": {}, - "required": [], + "properties": { + "metric": { + "type": "string", + "description": "Metric key, e.g. mission_success_rate, bug_kills, deaths", + } + }, + "required": ["metric"], }, ), ] @@ -109,6 +165,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: logger.info(f"Calling tool: {name}") try: + # Raw API tools if name == "get_war_status": result = await tools.get_war_status_tool() elif name == "get_planets": @@ -126,6 +183,43 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: result = await tools.get_biomes_tool() elif name == "get_factions": result = await tools.get_factions_tool() + elif name == "get_raw_api": + endpoint = arguments.get("endpoint") + if not endpoint: + raise ValueError("endpoint is required") + result = await tools.get_raw_api_tool( + endpoint=str(endpoint), + planet_index=arguments.get("planet_index"), + ) + # Outcome-based tools + elif name == "get_war_summary": + result = await tools.get_war_summary_tool() + elif name == "get_where_to_deploy": + result = await tools.get_where_to_deploy_tool( + limit=int(arguments.get("limit", 10)), + ) + elif name == "get_liberation_priority": + result = await tools.get_liberation_priority_tool( + limit=int(arguments.get("limit", 10)), + sector=arguments.get("sector"), + ) + elif name == "get_mission_efficiency_snapshot": + result = await tools.get_mission_efficiency_snapshot_tool() + # Analytics tools + elif name == "get_mission_analytics": + result = await tools.get_mission_analytics_tool() + elif name == "get_war_analytics": + result = await tools.get_war_analytics_tool() + elif name == "get_planet_analytics": + result = await tools.get_planet_analytics_tool( + sector=arguments.get("sector"), + group_by=arguments.get("group_by"), + ) + elif name == "query_stats": + metric = arguments.get("metric") + if not metric: + raise ValueError("metric is required") + result = await tools.query_stats_tool(metric=str(metric)) else: raise ValueError(f"Unknown tool: {name}") diff --git a/highcommand/tools.py b/highcommand/tools.py index 14b8702..5b3a36b 100644 --- a/highcommand/tools.py +++ b/highcommand/tools.py @@ -7,9 +7,41 @@ import structlog from highcommand.api_client import HighCommandAPIClient +from highcommand import analytics +from highcommand import outcomes logger = structlog.get_logger(__name__) +# Endpoints supported by get_raw_api +RAW_API_ENDPOINTS = frozenset({ + "war/status", "planets", "statistics", "campaigns/active", "biomes", "factions", +}) + + +def _envelope( + status: str, + outcome: str | None = None, + summary: str | None = None, + data: Any = None, + error: str | None = None, + elapsed_ms: float | None = None, + **extra: Any, +) -> dict[str, Any]: + """Build agent-friendly response envelope.""" + out = { + "status": status, + "outcome": outcome, + "summary": summary, + "data": data, + "error": error, + } + if elapsed_ms is not None: + out["metrics"] = {"elapsed_ms": round(elapsed_ms, 2)} + for k, v in extra.items(): + if k not in out and v is not None: + out[k] = v + return out + class HighCommandTools: """Tools for interacting with High-Command API.""" @@ -154,3 +186,273 @@ async def _fetch() -> Any: return await client.get_factions() return await self._run_tool(_fetch) + + # ---------- Raw API (single entry point for raw stats/messages) ---------- + + async def get_raw_api_tool(self, endpoint: str, planet_index: int | None = None) -> dict[str, Any]: + """Return raw API response for one endpoint. Use when the user asks about raw stats or API messages.""" + start = time.perf_counter() + try: + if endpoint not in RAW_API_ENDPOINTS: + return _envelope( + "error", + outcome="invalid_endpoint", + summary=f"Unknown endpoint. Use one of: {', '.join(sorted(RAW_API_ENDPOINTS))}.", + data=None, + error=f"Unknown endpoint: {endpoint}", + elapsed_ms=(time.perf_counter() - start) * 1000, + endpoint=endpoint, + ) + async with HighCommandAPIClient() as client: + if endpoint == "war/status": + data = await client.get_war_status() + elif endpoint == "planets": + data = await client.get_planet_status(planet_index) if planet_index is not None else await client.get_planets() + elif endpoint == "statistics": + data = await client.get_statistics() + elif endpoint == "campaigns/active": + data = await client.get_campaign_info() + elif endpoint == "biomes": + data = await client.get_biomes() + elif endpoint == "factions": + data = await client.get_factions() + else: + return _envelope( + "error", + outcome="invalid_endpoint", + summary=f"Unknown endpoint: {endpoint}", + data=None, + error=f"Unknown endpoint: {endpoint}", + elapsed_ms=(time.perf_counter() - start) * 1000, + endpoint=endpoint, + ) + elapsed_ms = (time.perf_counter() - start) * 1000 + return _envelope( + "success", + outcome="ok", + summary="Raw API response for custom analysis.", + data=data, + error=None, + elapsed_ms=elapsed_ms, + endpoint=endpoint, + ) + except Exception as e: + logger.error("get_raw_api failed", endpoint=endpoint, error=str(e)) + return _envelope( + "error", + outcome="error", + summary=str(e), + data=None, + error=str(e), + elapsed_ms=(time.perf_counter() - start) * 1000, + endpoint=endpoint, + ) + + # ---------- Outcome-based tools ---------- + + async def get_war_summary_tool(self) -> dict[str, Any]: + """Human-readable war summary and current phase. Outcome: What's the state of the war?""" + start = time.perf_counter() + try: + async with HighCommandAPIClient() as client: + war = await client.get_war_status() + result = outcomes.war_summary(war) + return _envelope( + "success", + error=None, + elapsed_ms=(time.perf_counter() - start) * 1000, + **result, + ) + except Exception as e: + logger.error("get_war_summary failed", error=str(e)) + return _envelope( + "error", + outcome="error", + summary=str(e), + data=None, + error=str(e), + elapsed_ms=(time.perf_counter() - start) * 1000, + ) + + async def get_where_to_deploy_tool(self, limit: int = 10) -> dict[str, Any]: + """Planets that need reinforcements most. Outcome: Where should I deploy?""" + start = time.perf_counter() + try: + async with HighCommandAPIClient() as client: + campaigns = await client.get_campaign_info() + planets = await client.get_planets() + result = outcomes.where_to_deploy(campaigns, planets, limit=limit) + return _envelope( + "success", + error=None, + elapsed_ms=(time.perf_counter() - start) * 1000, + **result, + ) + except Exception as e: + logger.error("get_where_to_deploy failed", error=str(e)) + return _envelope( + "error", + outcome="error", + summary=str(e), + data=None, + error=str(e), + elapsed_ms=(time.perf_counter() - start) * 1000, + ) + + async def get_liberation_priority_tool( + self, limit: int = 10, sector: str | None = None + ) -> dict[str, Any]: + """Ordered list of planets by liberation priority. Outcome: What to liberate first?""" + start = time.perf_counter() + try: + async with HighCommandAPIClient() as client: + planets = await client.get_planets() + campaigns = await client.get_campaign_info() + result = outcomes.liberation_priority( + planets, campaigns, limit=limit, sector=sector + ) + return _envelope( + "success", + error=None, + elapsed_ms=(time.perf_counter() - start) * 1000, + **result, + ) + except Exception as e: + logger.error("get_liberation_priority failed", error=str(e)) + return _envelope( + "error", + outcome="error", + summary=str(e), + data=None, + error=str(e), + elapsed_ms=(time.perf_counter() - start) * 1000, + ) + + async def get_mission_efficiency_snapshot_tool(self) -> dict[str, Any]: + """Current mission efficiency from global stats. Outcome: How are we doing on missions?""" + start = time.perf_counter() + try: + async with HighCommandAPIClient() as client: + stats = await client.get_statistics() + result = outcomes.mission_efficiency_snapshot(stats) + return _envelope( + "success", + error=None, + elapsed_ms=(time.perf_counter() - start) * 1000, + **result, + ) + except Exception as e: + logger.error("get_mission_efficiency_snapshot failed", error=str(e)) + return _envelope( + "error", + outcome="error", + summary=str(e), + data=None, + error=str(e), + elapsed_ms=(time.perf_counter() - start) * 1000, + ) + + # ---------- Analytics tools ---------- + + async def get_mission_analytics_tool(self) -> dict[str, Any]: + """Derived mission analytics: success rate, missions won/lost, mission time, kills breakdown.""" + start = time.perf_counter() + try: + async with HighCommandAPIClient() as client: + stats = await client.get_statistics() + result = analytics.mission_analytics(stats) + return _envelope( + "success", + error=None, + elapsed_ms=(time.perf_counter() - start) * 1000, + **result, + ) + except Exception as e: + logger.error("get_mission_analytics failed", error=str(e)) + return _envelope( + "error", + outcome="error", + summary=str(e), + data=None, + error=str(e), + elapsed_ms=(time.perf_counter() - start) * 1000, + ) + + async def get_war_analytics_tool(self) -> dict[str, Any]: + """War-level analytics: time left, active campaigns count, progress indicators.""" + start = time.perf_counter() + try: + async with HighCommandAPIClient() as client: + war = await client.get_war_status() + campaigns = await client.get_campaign_info() + planets = await client.get_planets() + result = analytics.war_analytics(war, campaigns, planets) + return _envelope( + "success", + error=None, + elapsed_ms=(time.perf_counter() - start) * 1000, + **result, + ) + except Exception as e: + logger.error("get_war_analytics failed", error=str(e)) + return _envelope( + "error", + outcome="error", + summary=str(e), + data=None, + error=str(e), + elapsed_ms=(time.perf_counter() - start) * 1000, + ) + + async def get_planet_analytics_tool( + self, sector: str | None = None, group_by: str | None = None + ) -> dict[str, Any]: + """Planet analytics: count by sector, by owner, planets under attack. Which sectors need help?""" + start = time.perf_counter() + try: + async with HighCommandAPIClient() as client: + planets = await client.get_planets() + campaigns = await client.get_campaign_info() + result = analytics.planet_analytics( + planets, campaigns, sector=sector, group_by=group_by + ) + return _envelope( + "success", + error=None, + elapsed_ms=(time.perf_counter() - start) * 1000, + **result, + ) + except Exception as e: + logger.error("get_planet_analytics failed", error=str(e)) + return _envelope( + "error", + outcome="error", + summary=str(e), + data=None, + error=str(e), + elapsed_ms=(time.perf_counter() - start) * 1000, + ) + + async def query_stats_tool(self, metric: str) -> dict[str, Any]: + """Answer a single stats question. metric e.g. mission_success_rate, deaths, time_played, bug_kills.""" + start = time.perf_counter() + try: + async with HighCommandAPIClient() as client: + stats = await client.get_statistics() + result = analytics.query_stat_metric(stats, metric) + return _envelope( + "success", + error=None, + elapsed_ms=(time.perf_counter() - start) * 1000, + **result, + ) + except Exception as e: + logger.error("query_stats failed", metric=metric, error=str(e)) + return _envelope( + "error", + outcome="error", + summary=str(e), + data=None, + error=str(e), + elapsed_ms=(time.perf_counter() - start) * 1000, + ) diff --git a/tests/demo_all_endpoints.py b/tests/demo_all_endpoints.py index 076948a..65e7840 100644 --- a/tests/demo_all_endpoints.py +++ b/tests/demo_all_endpoints.py @@ -7,7 +7,7 @@ async def test_all_endpoints(): - """Test all 7 available MCP tools.""" + """Test raw + outcome + analytics MCP tools.""" print("\n" + "=" * 70) print(" HIGH-COMMAND MCP SERVER - ALL ENDPOINTS DEMONSTRATION") print("=" * 70 + "\n") diff --git a/tests/test_server.py b/tests/test_server.py index 6ee54c0..827f140 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -10,10 +10,10 @@ @pytest.mark.asyncio async def test_list_tools(): - """Test that tools are properly listed.""" + """Test that tools are properly listed (raw + get_raw_api + outcome + analytics).""" tools = await list_tools() - assert len(tools) == 7 + assert len(tools) == 16 tool_names = {tool.name for tool in tools} expected_tools = { @@ -24,6 +24,15 @@ async def test_list_tools(): "get_planet_status", "get_biomes", "get_factions", + "get_raw_api", + "get_war_summary", + "get_where_to_deploy", + "get_liberation_priority", + "get_mission_efficiency_snapshot", + "get_mission_analytics", + "get_war_analytics", + "get_planet_analytics", + "query_stats", } assert tool_names == expected_tools @@ -231,3 +240,35 @@ async def test_call_tool_get_factions(): content = json.loads(result[0].text) assert content["status"] == "success" assert content["data"] == mock_data + + +@pytest.mark.asyncio +async def test_call_tool_get_war_summary(): + """Test outcome tool get_war_summary.""" + mock_war = {"data": {"id": 1, "index": 801, "endDate": "2028-02-08T20:04:55.000Z", "phase": "active"}} + + with patch("highcommand.tools.HighCommandAPIClient") as mock_client_class: + mock_client = AsyncMock() + mock_client_class.return_value = mock_client + mock_client.__aenter__.return_value = mock_client + mock_client.__aexit__.return_value = None + mock_client.get_war_status.return_value = mock_war + + result = await call_tool("get_war_summary", {}) + + assert len(result) == 1 + content = json.loads(result[0].text) + assert content["status"] == "success" + assert content.get("outcome") == "ok" + assert "summary" in content + + +@pytest.mark.asyncio +async def test_call_tool_query_stats_missing_metric(): + """Test query_stats with missing required parameter.""" + result = await call_tool("query_stats", {}) + + assert len(result) == 1 + content = json.loads(result[0].text) + assert content["status"] == "error" + assert "metric" in content["error"].lower() From 3b366f12698bcc64e5dba233d5acdd1b1e7edf07 Mon Sep 17 00:00:00 2001 From: Lee Chapman Date: Sun, 29 Mar 2026 14:33:30 -0700 Subject: [PATCH 2/2] style: fix ruff I001 import order in tools.py Made-with: Cursor --- highcommand/tools.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/highcommand/tools.py b/highcommand/tools.py index 5b3a36b..e033aee 100644 --- a/highcommand/tools.py +++ b/highcommand/tools.py @@ -6,9 +6,8 @@ import structlog +from highcommand import analytics, outcomes from highcommand.api_client import HighCommandAPIClient -from highcommand import analytics -from highcommand import outcomes logger = structlog.get_logger(__name__)