ai-skills-api/mcp/gameservers.py

from mcp.server.fastmcp import FastMCP
import os
import json
from pathlib import Path
from typing import Optional
import subprocess

mcp = FastMCP("gameservers")

GAME_SERVERS_DIR = Path(os.getenv("GAME_SERVERS_DIR", "/opt/game-servers"))


@mcp.tool()
def list_servers() -> list[dict]:
    """List all game servers"""
    if not GAME_SERVERS_DIR.exists():
        return []
    
    servers = []
    for server_dir in GAME_SERVERS_DIR.iterdir():
        if server_dir.is_dir():
            config_file = server_dir / "config.json"
            servers.append({
                "name": server_dir.name,
                "path": str(server_dir),
                "has_config": config_file.exists(),
                "config": json.loads(config_file.read_text()) if config_file.exists() else None
            })
    return servers


@mcp.tool()
def get_server_config(server_name: str) -> dict:
    """Get config for a specific game server"""
    config_file = GAME_SERVERS_DIR / server_name / "config.json"
    if not config_file.exists():
        return {"error": f"No config found for {server_name}"}
    return json.loads(config_file.read_text())


@mcp.tool()
def update_server_config(server_name: str, config: dict) -> dict:
    """Update game server config"""
    server_dir = GAME_SERVERS_DIR / server_name
    if not server_dir.exists():
        return {"error": f"Server {server_name} not found"}
    
    config_file = server_dir / "config.json"
    config_file.write_text(json.dumps(config, indent=2))
    return {"success": True, "message": f"Updated config for {server_name}"}


@mcp.tool()
def server_status(server_name: str) -> dict:
    """Get status of a game server"""
    server_dir = GAME_SERVERS_DIR / server_name
    if not server_dir.exists():
        return {"error": f"Server {server_name} not found"}
    
    status_file = server_dir / "status.json"
    if not status_file.exists():
        return {"status": "unknown", "message": "No status file found"}
    
    return json.loads(status_file.read_text())


@mcp.tool()
def start_server(server_name: str) -> dict:
    """Start a game server"""
    server_dir = GAME_SERVERS_DIR / server_name
    if not server_dir.exists():
        return {"error": f"Server {server_name} not found"}
    
    start_script = server_dir / "start.sh"
    if not start_script.exists():
        return {"error": f"No start script found for {server_name}"}
    
    try:
        subprocess.run(
            ["bash", str(start_script)],
            cwd=server_dir,
            capture_output=True,
            text=True,
            timeout=30
        )
        return {"success": True, "message": f"Started {server_name}"}
    except subprocess.TimeoutExpired:
        return {"error": "Start command timed out"}
    except Exception as e:
        return {"error": str(e)}


@mcp.tool()
def stop_server(server_name: str) -> dict:
    """Stop a game server"""
    server_dir = GAME_SERVERS_DIR / server_name
    if not server_dir.exists():
        return {"error": f"Server {server_name} not found"}
    
    stop_script = server_dir / "stop.sh"
    if not stop_script.exists():
        return {"error": f"No stop script found for {server_name}"}
    
    try:
        subprocess.run(
            ["bash", str(stop_script)],
            cwd=server_dir,
            capture_output=True,
            text=True,
            timeout=30
        )
        return {"success": True, "message": f"Stopped {server_name}"}
    except subprocess.TimeoutExpired:
        return {"error": "Stop command timed out"}
    except Exception as e:
        return {"error": str(e)}


@mcp.tool()
def get_server_logs(server_name: str, lines: int = 100) -> str:
    """Get logs from a game server"""
    server_dir = GAME_SERVERS_DIR / server_name
    if not server_dir.exists():
        return f"Server {server_name} not found"
    
    log_file = server_dir / "logs" / "latest.log"
    if not log_file.exists():
        return f"No log file found for {server_name}"
    
    try:
        result = subprocess.run(
            ["tail", "-n", str(lines), str(log_file)],
            capture_output=True,
            text=True
        )
        return result.stdout
    except Exception as e:
        return f"Error reading logs: {e}"


@mcp.tool()
def create_server(
    name: str,
    game_type: str,
    port: int,
    max_players: int = 10,
    config: Optional[dict] = None
) -> dict:
    """Create a new game server directory structure"""
    server_dir = GAME_SERVERS_DIR / name
    if server_dir.exists():
        return {"error": f"Server {name} already exists"}
    
    server_dir.mkdir(parents=True, exist_ok=True)
    (server_dir / "logs").mkdir(exist_ok=True)
    
    default_config = {
        "name": name,
        "game_type": game_type,
        "port": port,
        "max_players": max_players,
        "auto_restart": True,
        "restart_cron": "0 4 * * *"
    }
    if config:
        default_config.update(config)
    
    config_file = server_dir / "config.json"
    config_file.write_text(json.dumps(default_config, indent=2))
    
    start_script = server_dir / "start.sh"
    start_script.write_text(f"#!/bin/bash\n# TODO: Add start command for {game_type}\necho 'Starting {name}'\n")
    start_script.chmod(0o755)
    
    stop_script = server_dir / "stop.sh"
    stop_script.write_text(f"#!/bin/bash\n# TODO: Add stop command for {game_type}\necho 'Stopping {name}'\n")
    stop_script.chmod(0o755)
    
    return {
        "success": True,
        "message": f"Created server {name}",
        "path": str(server_dir)
    }


@mcp.tool()
def delete_server(server_name: str, keep_logs: bool = False) -> dict:
    """Delete a game server"""
    server_dir = GAME_SERVERS_DIR / server_name
    if not server_dir.exists():
        return {"error": f"Server {server_name} not found"}
    
    try:
        if keep_logs:
            import shutil
            logs_dir = server_dir / "logs"
            if logs_dir.exists():
                shutil.rmtree(logs_dir)
        else:
            import shutil
            shutil.rmtree(server_dir)
        return {"success": True, "message": f"Deleted {server_name}"}
    except Exception as e:
        return {"error": str(e)}


@mcp.tool()
def get_templates() -> list[dict]:
    """Get available game server templates"""
    templates = {
        "valheim": {
            "game_type": "valheim",
            "default_port": 2456,
            "description": "Valheim dedicated server"
        },
        "minecraft": {
            "game_type": "minecraft",
            "default_port": 25565,
            "description": "Minecraft Java Edition server"
        },
        "terraria": {
            "game_type": "terraria",
            "default_port": 7777,
            "description": "Terraria server"
        },
        "factorio": {
            "game_type": "factorio",
            "default_port": 34197,
            "description": "Factorio dedicated server"
        }
    }
    return [{"name": k, **v} for k, v in templates.items()]


if __name__ == "__main__":
    mcp.run()