from typing import Any, Dict import logging import httpx from mcp.server.fastmcp import FastMCP from utils.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()