timmy-home/angband/mcp_server.py

#!/usr/bin/env python3
"""
Angband MCP Server — Timmy's watchable ASCII game interface.

Body:   tmux session running terminal Angband
Eyes:   tmux capture-pane
Hands:  tmux send-keys
Brain:  Hermes TUI via MCP tools

This keeps gameplay visible, local, and telemetry-friendly.
"""

import json
import os
import shlex
import subprocess
import time
from pathlib import Path

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

ANGBAND_BIN = "/opt/homebrew/bin/angband"
ANGBAND_ROOT = Path.home() / ".timmy" / "angband"
RUNTIME_DIR = ANGBAND_ROOT / "runtime"
USER_DIR = RUNTIME_DIR / "user"
SAVE_DIR = RUNTIME_DIR / "save"
ARCHIVE_DIR = RUNTIME_DIR / "archive"
PANIC_DIR = RUNTIME_DIR / "panic"
SCORES_DIR = RUNTIME_DIR / "scores"
LOG_DIR = ANGBAND_ROOT / "logs"
SESSION_NAME = "Angband"
DEFAULT_USER = "timmy"
DEFAULT_WIDTH = 120
DEFAULT_HEIGHT = 40

app = Server("angband")


def ensure_dirs():
    for path in (ANGBAND_ROOT, RUNTIME_DIR, USER_DIR, SAVE_DIR, ARCHIVE_DIR, PANIC_DIR, SCORES_DIR, LOG_DIR):
        path.mkdir(parents=True, exist_ok=True)


def tmux(args, check=True):
    result = subprocess.run(["tmux", *args], capture_output=True, text=True)
    if check and result.returncode != 0:
        raise RuntimeError(result.stderr.strip() or result.stdout.strip() or f"tmux failed: {' '.join(args)}")
    return result


def session_exists(session_name=SESSION_NAME):
    return tmux(["has-session", "-t", session_name], check=False).returncode == 0


def pane_id(session_name=SESSION_NAME):
    if not session_exists(session_name):
        return None
    out = tmux(["list-panes", "-t", session_name, "-F", "#{pane_id}"]).stdout.strip().splitlines()
    return out[0].strip() if out else None


def capture_screen(lines=60, session_name=SESSION_NAME):
    pid = pane_id(session_name)
    if not pid:
        return "No Angband tmux pane found."
    # Angband runs in the terminal's alternate screen buffer. `-a` is required
    # or tmux returns an empty capture even while the game is visibly running.
    result = tmux(["capture-pane", "-a", "-p", "-t", pid, "-S", f"-{max(10, int(lines))}"])
    return result.stdout.rstrip()


def has_save(user=DEFAULT_USER):
    if not SAVE_DIR.exists():
        return False
    for path in SAVE_DIR.iterdir():
        if path.name.startswith(user):
            return True
    return False


SPECIAL_KEYS = {
    "enter": "Enter",
    "return": "Enter",
    "esc": "Escape",
    "escape": "Escape",
    "up": "Up",
    "down": "Down",
    "left": "Left",
    "right": "Right",
    "space": "Space",
    "tab": "Tab",
    "backspace": "BSpace",
    "delete": "DC",
    "home": "Home",
    "end": "End",
    "pageup": "PageUp",
    "pagedown": "PageDown",
    "pgup": "PageUp",
    "pgdn": "PageDown",
    "ctrl-c": "C-c",
    "ctrl-x": "C-x",
    "ctrl-z": "C-z",
}


def send_key(key, session_name=SESSION_NAME):
    pid = pane_id(session_name)
    if not pid:
        raise RuntimeError("No Angband tmux pane found.")
    normalized = str(key).strip()
    mapped = SPECIAL_KEYS.get(normalized.lower())
    if mapped:
        tmux(["send-keys", "-t", pid, mapped])
    elif len(normalized) == 1:
        tmux(["send-keys", "-t", pid, "-l", normalized])
    else:
        # Let tmux interpret names like F1 if passed through.
        tmux(["send-keys", "-t", pid, normalized])


def send_text(text, session_name=SESSION_NAME):
    pid = pane_id(session_name)
    if not pid:
        raise RuntimeError("No Angband tmux pane found.")
    tmux(["send-keys", "-t", pid, "-l", text])


def maybe_continue_splash(session_name=SESSION_NAME):
    screen = capture_screen(80, session_name)
    advanced = False
    if "Press any key to continue" in screen:
        send_key("enter", session_name)
        time.sleep(0.8)
        screen = capture_screen(80, session_name)
        advanced = True
    return advanced, screen


def launch_game(user=DEFAULT_USER, new_game=False, continue_splash=True, width=DEFAULT_WIDTH, height=DEFAULT_HEIGHT):
    ensure_dirs()

    if not Path(ANGBAND_BIN).exists():
        return {
            "error": f"Angband binary not found: {ANGBAND_BIN}"
        }

    if session_exists():
        advanced = False
        screen = capture_screen(80)
        if continue_splash:
            advanced, screen = maybe_continue_splash()
        return {
            "launched": False,
            "already_running": True,
            "session": SESSION_NAME,
            "attach": f"tmux attach -t {SESSION_NAME}",
            "continued_splash": advanced,
            "screen": screen,
        }

    use_new_game = bool(new_game or not has_save(user))
    cmd = [
        ANGBAND_BIN,
        f"-u{user}",
        "-mgcu",
        f"-duser={USER_DIR}",
        f"-dsave={SAVE_DIR}",
        f"-darchive={ARCHIVE_DIR}",
        f"-dpanic={PANIC_DIR}",
        f"-dscores={SCORES_DIR}",
    ]
    if use_new_game:
        cmd.insert(1, "-n")

    shell_cmd = "export TERM=xterm-256color; exec " + " ".join(shlex.quote(part) for part in cmd)
    tmux([
        "new-session", "-d",
        "-s", SESSION_NAME,
        "-x", str(int(width)),
        "-y", str(int(height)),
        shell_cmd,
    ])

    time.sleep(2.5)
    advanced = False
    screen = capture_screen(80)
    if continue_splash:
        advanced, screen = maybe_continue_splash()

    return {
        "launched": True,
        "already_running": False,
        "new_game": use_new_game,
        "session": SESSION_NAME,
        "attach": f"tmux attach -t {SESSION_NAME}",
        "continued_splash": advanced,
        "screen": screen,
    }


def stop_game():
    if not session_exists():
        return {"stopped": False, "message": "Angband session is not running."}
    tmux(["kill-session", "-t", SESSION_NAME])
    return {"stopped": True, "session": SESSION_NAME}


def status():
    running = session_exists()
    savefiles = []
    if SAVE_DIR.exists():
        savefiles = sorted(path.name for path in SAVE_DIR.iterdir())
    result = {
        "running": running,
        "session": SESSION_NAME if running else None,
        "attach": f"tmux attach -t {SESSION_NAME}" if running else None,
        "savefiles": savefiles,
    }
    if running:
        result["screen"] = capture_screen(40)
    return result


def observe(lines=60):
    return {
        "running": session_exists(),
        "session": SESSION_NAME if session_exists() else None,
        "screen": capture_screen(lines),
    }


def keypress(key, wait_ms=500):
    send_key(key)
    time.sleep(max(0, int(wait_ms)) / 1000.0)
    return {
        "sent": key,
        "screen": capture_screen(60),
    }


def type_and_observe(text, wait_ms=500):
    send_text(text)
    time.sleep(max(0, int(wait_ms)) / 1000.0)
    return {
        "sent": text,
        "screen": capture_screen(60),
    }


@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="status",
            description="Check whether the watchable Angband tmux session is running, list savefiles, and return the current visible screen when available.",
            inputSchema={"type": "object", "properties": {}, "required": []},
        ),
        Tool(
            name="launch",
            description="Launch terminal Angband inside a watchable tmux session named Angband. Loads an existing save for the given user when present; otherwise starts a new game. Can auto-advance the initial splash screen.",
            inputSchema={
                "type": "object",
                "properties": {
                    "user": {"type": "string", "description": "Savefile/user slot name (default: timmy)"},
                    "new_game": {"type": "boolean", "description": "Force a new game even if a save exists.", "default": False},
                    "continue_splash": {"type": "boolean", "description": "Press Enter automatically if the splash page says 'Press any key to continue'.", "default": True},
                    "width": {"type": "integer", "description": "tmux width for the visible game session", "default": 120},
                    "height": {"type": "integer", "description": "tmux height for the visible game session", "default": 40},
                },
                "required": [],
            },
        ),
        Tool(
            name="observe",
            description="Read the current Angband screen as plain text from the tmux pane. Use this before acting.",
            inputSchema={
                "type": "object",
                "properties": {
                    "lines": {"type": "integer", "description": "How many recent screen lines to capture", "default": 60},
                },
                "required": [],
            },
        ),
        Tool(
            name="keypress",
            description="Send one key to Angband and then return the updated screen. Common keys: Enter, Escape, Up, Down, Left, Right, Space, Tab, Backspace, ctrl-x, ?, *, @, letters, numbers.",
            inputSchema={
                "type": "object",
                "properties": {
                    "key": {"type": "string", "description": "Key to send"},
                    "wait_ms": {"type": "integer", "description": "Milliseconds to wait before recapturing the screen", "default": 500},
                },
                "required": ["key"],
            },
        ),
        Tool(
            name="type_text",
            description="Type literal text into Angband and then return the updated screen. Useful when a menu expects a name or command string.",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "Literal text to type"},
                    "wait_ms": {"type": "integer", "description": "Milliseconds to wait before recapturing the screen", "default": 500},
                },
                "required": ["text"],
            },
        ),
        Tool(
            name="stop",
            description="Kill the watchable Angband tmux session.",
            inputSchema={"type": "object", "properties": {}, "required": []},
        ),
    ]


@app.call_tool()
async def call_tool(name: str, arguments: dict):
    arguments = arguments or {}

    if name == "status":
        result = status()
    elif name == "launch":
        result = launch_game(
            user=arguments.get("user", DEFAULT_USER),
            new_game=arguments.get("new_game", False),
            continue_splash=arguments.get("continue_splash", True),
            width=arguments.get("width", DEFAULT_WIDTH),
            height=arguments.get("height", DEFAULT_HEIGHT),
        )
    elif name == "observe":
        result = observe(lines=arguments.get("lines", 60))
    elif name == "keypress":
        result = keypress(arguments.get("key", ""), wait_ms=arguments.get("wait_ms", 500))
    elif name == "type_text":
        result = type_and_observe(arguments.get("text", ""), wait_ms=arguments.get("wait_ms", 500))
    elif name == "stop":
        result = stop_game()
    else:
        result = {"error": f"Unknown tool: {name}"}

    return [TextContent(type="text", text=json.dumps(result, indent=2))]


async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())