"""MCP stdio server exposing Overlord data to Claude Code. Configured via .mcp.json at the repo root, which Claude Code auto-loads when invoked with cwd=/home/erik/MosswartOverlord. Tool implementations live in tools.py — this file is just MCP protocol plumbing. Run directly with: python3 /home/erik/MosswartOverlord/agent/mcp_overlord.py """ from __future__ import annotations import asyncio import json import logging from typing import Any from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import TextContent, Tool from . import tools as T logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s mcp_overlord: %(message)s", ) logger = logging.getLogger("mcp_overlord") server: Server = Server("overlord") # ─── Tool registry ────────────────────────────────────────────────── # # Each entry: name → (description, JSON schema, callable async fn). # We register them with @server.list_tools / @server.call_tool below. TOOL_DEFS: dict[str, dict[str, Any]] = { "get_live_players": { "description": ( "Return active characters seen in the last ~30 seconds with their " "current position, kills, KPH, vitae, online time, and VTank state. " "Use this for any 'who is online right now / what is X doing' question." ), "schema": {"type": "object", "properties": {}}, "fn": lambda _args: T.get_live_players(), }, "get_recent_rares": { "description": ( "Return rare item finds from the last N hours, newest first. " "Use for questions about recent drops, who is finding rares, or " "rare-rate analysis. Defaults to 24 hours, max 30 days." ), "schema": { "type": "object", "properties": { "hours": { "type": "integer", "minimum": 1, "maximum": 720, "default": 24, }, "limit": { "type": "integer", "minimum": 1, "maximum": 200, "default": 100, }, }, }, "fn": lambda args: T.get_recent_rares( hours=int(args.get("hours", 24)), limit=int(args.get("limit", 100)), ), }, "query_telemetry_db": { "description": ( "Run a read-only SQL query against the telemetry database (TimescaleDB). " "Only SELECT / WITH statements are accepted; any DML or DDL is rejected. " "Useful for questions that aren't covered by the other tools — top-N " "lists, custom aggregations, time-window comparisons. " "Available tables include: telemetry_events (hypertable, 30d retention), " "rare_events, spawn_events (hypertable, 7d retention), portals, " "char_stats, rare_stats, rare_stats_sessions, character_stats, " "combat_stats, combat_stats_sessions, server_status. " "The query has a 10s timeout and returns at most 200 rows." ), "schema": { "type": "object", "required": ["sql"], "properties": { "sql": { "type": "string", "description": "A single PostgreSQL SELECT or WITH ... SELECT statement.", } }, }, "fn": lambda args: T.query_telemetry_db(str(args["sql"])), }, "get_player_state": { "description": ( "Combined snapshot for ONE character: live telemetry (if online) " "+ full character stats (attributes, skills, augmentations). " "Use this for questions like 'what is X doing right now' or 'show me X's stats'." ), "schema": { "type": "object", "required": ["character_name"], "properties": { "character_name": {"type": "string"}, }, }, "fn": lambda args: T.get_player_state(str(args["character_name"])), }, "get_inventory": { "description": ( "Full inventory listing for one character — every item with name, " "icon, container, equipped slot, spells, material, tinkers, etc. " "Large response — prefer get_inventory_search for narrow queries." ), "schema": { "type": "object", "required": ["character_name"], "properties": {"character_name": {"type": "string"}}, }, "fn": lambda args: T.get_inventory(str(args["character_name"])), }, "get_inventory_search": { "description": ( "Filtered inventory search for ONE character. Use search_items " "instead when the user wants to find something across ALL chars." ), "schema": { "type": "object", "required": ["character_name"], "properties": { "character_name": {"type": "string"}, "filters": { "type": "object", "description": "Query params dict, e.g. {\"name\": \"pearl\", \"armor_level_min\": 500}", }, }, }, "fn": lambda args: T.get_inventory_search( str(args["character_name"]), args.get("filters") or {} ), }, "search_items": { "description": ( "CROSS-CHARACTER item search — one query that scans every " "character's inventory. Use this whenever the user asks " "'find me an X on any of my chars'. **Do not** iterate " "get_inventory per character — this single tool call is far " "faster and avoids agent timeouts.\n\n" "Filter keys (pass as `filters` object, all optional):\n" " include_all_characters: true (default if no scope given)\n" " character: 'Name' (single char)\n" " characters: 'A,B,C' (specific list, comma-separated)\n" " text: substring of item name/description\n" " has_spell: 'Legendary Acid Ward' (exact spell name match)\n" " spell_contains: 'Legendary' (substring)\n" " legendary_cantrips: 'Foo,Bar'\n" " equipment_status: 'equipped' | 'unequipped'\n" " equipment_slot: int bitmask (4=chest, 2048=bracelet, 4096=ring)\n" " slot_names: 'Bracelet,Ring'\n" " armor_only / jewelry_only / weapon_only: bool\n" " min_armor / max_armor / min_damage / max_damage: int\n" ), "schema": { "type": "object", "required": ["filters"], "properties": { "filters": { "type": "object", "description": "Query params dict — see tool description for keys.", }, }, }, "fn": lambda args: T.search_items_global(args.get("filters") or {}), }, "get_combat_stats": { "description": ( "Lifetime + session combat stats for one character. Includes total " "damage given/received, per-element offense/defense breakdown, kill " "counts, and aetheria surge counts." ), "schema": { "type": "object", "required": ["character_name"], "properties": {"character_name": {"type": "string"}}, }, "fn": lambda args: T.get_combat_stats(str(args["character_name"])), }, "get_equipment_cantrips": { "description": ( "Currently-equipped items for a character along with their active " "cantrip/spell state. Useful for 'what is X wearing' or 'is X " "running their suit' questions." ), "schema": { "type": "object", "required": ["character_name"], "properties": {"character_name": {"type": "string"}}, }, "fn": lambda args: T.get_equipment_cantrips(str(args["character_name"])), }, "get_quest_status": { "description": ( "Active quest timers and progress across ALL characters. Returns " "for each character which quests are READY vs counting down." ), "schema": {"type": "object", "properties": {}}, "fn": lambda _args: T.get_quest_status(), }, "get_server_health": { "description": ( "Current Coldeve game-server status: up/down, latency in ms, " "current player count from TreeStats.net, total uptime. Updated " "every 30 seconds in the background." ), "schema": {"type": "object", "properties": {}}, "fn": lambda _args: T.get_server_health(), }, "suitbuilder_search": { "description": ( "Run a constraint-satisfaction armor optimization across all " "characters' inventories ('mules'). Drives the same suitbuilder " "the /suitbuilder.html page uses. Pass the same params dict the " "page sends — see /suitbuilder.html JS for the schema. The search " "is SSE-streaming on the backend; this tool collects until done " "and returns the final suit(s) plus the last few phase events. " "Can take up to 5 minutes for complex constraints — only call " "when the user explicitly asks for an optimization run." ), "schema": { "type": "object", "required": ["params"], "properties": { "params": { "type": "object", "description": "Suitbuilder request body (characters, locked slots, set constraints, etc.)", }, }, }, "fn": lambda args: T.suitbuilder_search(args.get("params") or {}), }, } # ─── MCP protocol wiring ──────────────────────────────────────────── @server.list_tools() async def list_tools() -> list[Tool]: return [ Tool(name=name, description=defn["description"], inputSchema=defn["schema"]) for name, defn in TOOL_DEFS.items() ] @server.call_tool() async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: if name not in TOOL_DEFS: return [TextContent(type="text", text=f"unknown tool: {name}")] fn = TOOL_DEFS[name]["fn"] try: result = await fn(arguments or {}) except T.SqlNotAllowed as e: return [TextContent(type="text", text=f"REJECTED: {e}")] except Exception as e: # noqa: BLE001 logger.exception("tool %s failed", name) return [TextContent(type="text", text=f"ERROR: {type(e).__name__}: {e}")] text = json.dumps(result, default=str, ensure_ascii=False, indent=2) return [TextContent(type="text", text=text)] async def _run() -> None: logger.info("starting MCP stdio server (overlord)") try: async with stdio_server() as (reader, writer): await server.run(reader, writer, server.create_initialization_options()) finally: await T.shutdown() def main() -> None: asyncio.run(_run()) if __name__ == "__main__": main()