# Game imports from utils.dice import Dice from utils.game import Game from utils.serializable import Serializable from events.event import Event # Native imports from typing import Any, Dict import logging import httpx from mcp.server.fastmcp import FastMCP import json import os # Constants HISTORY_FILE = "game_history.json" SAVE_PATH = "save_" # Global Parameters mcp = FastMCP("wyvern-castle") game: Game = None # Logging config logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(), ] ) # SAVING & LOADING GAME STATE @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) } @DeprecationWarning 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) @DeprecationWarning 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 [] # EVENTS TOOLS @mcp.tool() async def start_event(location:str, initial_description:str, entity_list:list[str]): new_event = Event(location=location, initial_description=initial_description, entities=entity_list) game.add_event(new_event) @mcp.tool() async def perform_action(): pass # ITEM TOOLS @mcp.tool() async def create_item(): pass @mcp.tool() async def add_item_to_entity(): pass # OTHER UTILS @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: return { "success": False, "error": "Number of faces must be at least 1" } elif n_faces == 2: return { "success": True, "toss_result": Dice.head_or_tails() } else: return { "success": True, "roll_result": Dice.roll(n_faces) } @mcp.tool() async def get_entity_status(): pass @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_id:str): """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 NPC named {name}") try: player_id = game.create_player(name=name, strength=strength, dexterity=dexterity, intelligence=intelligence, wisdom=wisdom, charisma=charisma, hp=hp, armor=armor, speed=speed) game.add_item_to_entity(item_id=item_id, entity_id=player_id) player_dict = game.get_player(player_id=player_id).serialize() logging.info(f"Creation of NPC successful") return { "success": True, "npc_properties": player_dict } except ReferenceError as e: logging.info(f"ReferenceError: " + str(e)) return { "success": False, "error": str(e) } @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_id:str): """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 named {name}") try: npc_id = game.create_npc(name=name, strength=strength, dexterity=dexterity, intelligence=intelligence, wisdom=wisdom, charisma=charisma, hp=hp, armor=armor, speed=speed) game.add_item_to_entity(item_id=item_id, entity_id=npc_id) npc_dict = game.get_npc(npc_id=npc_id).serialize() logging.info(f"Creation of NPC successful") return { "success": True, "npc_properties": npc_dict } except ReferenceError as e: logging.info(f"ReferenceError: " + str(e)) return { "success": False, "error": str(e) } @mcp.tool() async def create_item(name: str, description: str, stat_modifier: dict[str,int]): """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, name={name} ; description={description}") item_id = game.create_item(name, description, stat_modifier) item_dict = game.get_item(item_id).serialize() return { "success": True, "item_properties": item_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()