Wyvern-Castle/server.py

from typing import Any, Dict
import logging
import httpx
from mcp.server.fastmcp import FastMCP
from dice import Dice
from entities.player import Player
from items.item import Item
from game import Game
from entities.npc import NPC
from serializable import Serializable
import json
import os
mcp = FastMCP("wyvern-castle")
game: Game = None

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(),
    ]
)

# Constants
HISTORY_FILE = "game_history.json"
SAVE_PATH = "save_"

@mcp.tool()
async def load_game(slot:int):
    """Loads an already existing game.
    
    Args:
        slot: Integer id of the save slot.
    """
    global game
    path = SAVE_PATH + str(slot) + ".json"
    try:
        with open(path, "r", encoding="utf-8") as f:
            game.deserialize(f.read())
        return {
            "success": True,
            "msg": f"{path} as been successfully loaded!"
        }
    except OSError as e:
        return {
            "success": False,
            "error": str(e)
        }

@mcp.tool()
async def save_game(slot:int):
    """Saves the current game to the given slot.
    
    Args:
        slot: Integer id of the save slot.
    """
    global game
    path = SAVE_PATH + str(slot) + ".json"
    try:
        with open(path, "w", encoding="utf-8") as f:
            f.write(game.serialize())
        return {
            "success": True,
            "msg": f"{path} as been successfully saved!"
        }
    except OSError as e:
        return {
            "success": False,
            "error": str(e)
        }

def append_to_history(event: Dict[str, Any]):
    """Append a game event to the history file."""
    history = []
    if os.path.exists(HISTORY_FILE):
        with open(HISTORY_FILE, "r", encoding="utf-8") as f:
            try:
                history = json.load(f)
            except json.JSONDecodeError:
                history = []
    history.append(event)
    with open(HISTORY_FILE, "w", encoding="utf-8") as f:
        json.dump(history, f, ensure_ascii=False, indent=2)

def read_history() -> list:
    """Read the game history from the file."""
    if os.path.exists(HISTORY_FILE):
        with open(HISTORY_FILE, "r", encoding="utf-8") as f:
            try:
                return json.load(f)
            except json.JSONDecodeError:
                return []
    return []

@mcp.tool()
async def throw_a_dice(n_faces: int) -> Any:
    """Throw a dice with n faces. If n==2 its a coin toss.

    Args:
        n_faces: Number of faces of the dice
    """
    logging.info(f"Throwing a dice with {n_faces} faces")

    if n_faces < 1:
        raise ValueError("Number of faces must be at least 1")
    elif n_faces == 1:
        return 1
    elif n_faces == 2:
        return Dice.head_or_tails()
    else:
        return Dice.roll(n_faces)



@mcp.tool()
async def create_player(name: str, strength: int, dexterity: int, intelligence: int, wisdom: int, charisma: int, hp: int, armor: int, speed: int, item: str = "") -> Dict[str, Any]:
    """Create a new player. Need all the stats to function properly. Throw a d20 for every stats you don't have,
    and a d6 for hp, armor and speed.

    Args:
        name: Name of the player
        strength: Strength of the player
        dexterity: Dexterity of the player
        intelligence: Intelligence of the player
        wisdom: Wisdom of the player
        charisma: Charisma of the player
        hp: Hit points of the player
        armor: Armor class of the player
        speed: Speed of the player
        item: Item carried by the player
    """
    logging.info(f"Creating player with name={name}")
    player = Player(name, strength, dexterity, intelligence, wisdom, charisma, hp, armor, speed)
    logging.info(f"Created player: {player}")
    game.active_players.append(player)
    return player.serialize_dict()

@mcp.tool()
async def create_npc(name: str, strength: int, dexterity: int, intelligence: int, wisdom: int, charisma: int, hp: int, armor: int, speed: int, item: str = "") -> Dict[str, Any]:
    """Create a new NPC. Need all the stats to function properly. Throw a d20 for every stats you don't have,
    and a d6 for hp, armor and speed.

    Args:
        name: Name of the NPC
        strength: Strength of the NPC
        dexterity: Dexterity of the NPC
        intelligence: Intelligence of the NPC
        wisdom: Wisdom of the NPC
        charisma: Charisma of the NPC
        hp: Hit points of the NPC
        armor: Armor class of the NPC
        speed: Speed of the NPC
        item: Item carried by the NPC
    """
    logging.info(f"Creating NPC with name={name}")
    npc = NPC(name, strength, dexterity, intelligence, wisdom, charisma, hp, armor, speed)
    logging.info(f"Created NPC: {npc}")
    game.active_npcs.append(npc)
    return npc.serialize_dict()

@mcp.tool()
async def create_item(name: str, description: str, bonus: str) -> Dict[str, Any]:
    """Create a new item.

    Args:
        name: Name of the item
        description: Description of the item
        bonus: Bonus of the item ex: strength+1,hp+5
    """
    logging.info(f"Creating item with name={name}")
    item = Item(name, description, bonus)
    logging.info(f"Created item: {item}")
    game.active_items.append(item)
    return item.serialize_dict()

@mcp.tool()
async def add_item_to_player(player_name: str, item_name: str) -> Dict[str, Any]:
    """Add an item to a player's inventory.

    Args:
        player_name: The name of the player to add the item to
        item_name: The name of the item to add
    """
    logging.info(f"Adding item {item_name} to player {player_name}")
    return {"status": "Item added"}

@mcp.tool()
async def save_game_state() -> str:
    """Save the current game state to a persistent storage each time
    the game state is modified."""
    logging.info("Saving game state")
    with open("game_state.json", "w", encoding="utf-8") as f:
        f.write(game.serialize())
    return "Game state saved to game_state.json"


# Example MCP tool to add an event to history
@mcp.tool()
async def add_event_to_history(event: Dict[str, Any]) -> str:
    """Add a game event to the history resource."""
    append_to_history(event)
    return "Event added to history."

# Example MCP tool to read history
@mcp.tool()
async def get_game_history() -> list:
    """Get the full game history."""
    return read_history()

@mcp.tool()
async def purge_game_history_and_state() -> str:
    """Purge the game history and state files when the player is starting a new game."""
    if os.path.exists(HISTORY_FILE):
        os.remove(HISTORY_FILE)
    if os.path.exists("game_state.json"):
        os.remove("game_state.json")
    return "Game history and state purged."
    

def main():
    # Initialize and run the server
    mcp.run(transport="stdio")

if __name__ == "__main__":
    main()