# Game imports from events.turn import TurnAction from utils.dice import Dice from utils.game import Game, Stat 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() # 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) } # EVENTS TOOLS @mcp.tool() async def start_event(location:str, initial_description:str, entity_list:list[str]): """ Start a new event in the game. Args: location: Location of the event initial_description: Initial description of the event entity_list: List of entity IDs involved in the event """ new_event = Event(location=location, initial_description=initial_description, entities=entity_list) game.add_event(new_event) @mcp.tool() async def perform_attack(src_entity_id:str, target_entity_id:str, attack_type:Stat): """Perform an attack during an event. This add an attack action to the current event. Args: src_entity_id: The ID of the entity performing the attack target_entity_id: The ID of the entity being attacked attack_type: The type of attack being performed using the class Stat, can be Stat.STRENGTH, Stat.INTELLIGENCE, Stat.DEXTERITY which are 0, 1 and 2 respectively """ logging.info(f"Entity {src_entity_id} is performing an attack on {target_entity_id} using {attack_type}") game.deal_damage(src=src_entity_id, target=target_entity_id, roll=Dice.roll(20), stat=attack_type, description=f"Entity {src_entity_id} attacks {target_entity_id} with a {attack_type} based attack.") return { "success": True, "msg": f"Attack performed by {src_entity_id} on {target_entity_id} using {attack_type}" } @mcp.tool() async def perform_test_action(entity_id:str, stat:Stat, difficulty:int, roll:int): """Perform a test action during an event. This add a test action to the current event. This can be opening a door (DEXTERITY), solving a puzzle (INTELLIGENCE), resisting a charm (WISDOM) etc. Args: entity_id: The ID of the entity performing the test stat: The stat being tested (using the Stat enum) difficulty: The difficulty of the test """ logging.info(f"Entity {entity_id} is performing a test action on stat {stat} with difficulty {difficulty}") entity = game.get_entity(entity_id) stat_value = 0 match(stat): case Stat.STRENGTH: stat_value = entity.get_strength() case Stat.INTELLIGENCE: stat_value = entity.get_intelligence() case Stat.DEXTERITY: stat_value = entity.get_dexterity() case Stat.WISDOM: stat_value = entity.get_wisdom() case Stat.CHARISMA: stat_value = entity.get_charisma() total = roll + stat_value success = total >= difficulty description = f"Entity {entity_id} performs a {stat.name} test (roll: {roll} + stat: {stat_value} = total: {total}) against difficulty {difficulty}. " if success: description += "The test is successful." else: description += "The test fails." game.get_current_event().perform_action(TurnAction.BASIC, entity_id, description=description) return { "success": success, "total": total, "msg": description } @mcp.tool() async def perform_stat_modification(entity_id:str, stat:Stat, value:int): """Modify a stat of an entity during an event. This add a stat modification action to the current event. This can be due to a spell, a potion, a curse etc. Args: entity_id: The ID of the entity whose stat is being modified stat: The stat being modified (using the Stat enum) value: The value to modify the stat by (can be positive or negative) """ logging.info(f"Entity {entity_id} is having its stat {stat} modified by {value}") game.modifying_stat(src=entity_id, value=value, stat=stat, description=f"Entity {entity_id} has its {stat.name} modified by {value}.") return { "success": True, "msg": f"Stat {stat.name} of entity {entity_id} modified by {value}" } # ITEM TOOLS # 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(): """ Get the status of all entities in the game. """ logging.info("Getting status of all entities") players = [player.serialize() for player in game.active_players] npcs = [npc.serialize() for npc in game.active_npcs] return { "players": players, "npcs": npcs } @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 d50 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 named {name}") try: player_id = game.create_player(name=name, strength=strength, dexterity=dexterity, intelligence=intelligence, wisdom=wisdom, charisma=charisma, hp=50 + hp, armor=50 + armor, speed= 50 + speed, equipped_item=game.get_item(item_id)) # Check if item exists 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 player successful") return { "success": True, "player_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 d100 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: item = game.get_item(item_id) # Check if item exists npc_id = game.create_npc(name=name, strength=strength, dexterity=dexterity, intelligence=intelligence, wisdom=wisdom, charisma=charisma, hp=50 + hp, armor=50 + armor, speed=50 + speed, equipped_item=item) 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 or malus 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 equip_item_to_entity(entity_id: str, item_id: str) -> Dict[str, Any]: """Equip an item to an entity (player or NPC). Args: entity_id: The id of the entity to equip the item to item_id: The id of the item to equip """ logging.info(f"Equipping item {item_id} to entity {entity_id}") game.add_item_to_entity(item_id=item_id, entity_id=entity_id) return {"status": "Item equipped"} @mcp.tool() async def get_entity_info(entity_id: str) -> Dict[str, Any]: """Get the full information of an entity (player or NPC). Args: entity_id: The id of the entity to get information from """ logging.info(f"Getting info for entity {entity_id}") entity = game.get_entity(entity_id) return { "entity_properties": entity.serialize() } @mcp.tool() async def get_item_info(item_id: str) -> Dict[str, Any]: """Get the full information of an item. Args: item_id: The id of the item to get information from """ logging.info(f"Getting info for item {item_id}") item = game.get_item(item_id) return { "item_properties": item.serialize() } @mcp.tool() async def get_game_state() -> str: """Get the current game state as a serialized string.""" logging.info("Getting current game state") return game.serialize() @mcp.tool() async def get_current_event() -> Dict[str, Any]: """Get the current event in the game.""" logging.info("Getting current event") event = game.get_current_event() if event: return { "success": True, "event_properties": event.serialize() } else: return { "success": False, "error": "No current event" } @mcp.tool() async def add_event(location:str, initial_description:str, entity_list:list[str]) -> str: """Add a new event to the game. Args: location: Location of the event initial_description: Initial description of the event entity_list: List of entity IDs involved in the event """ logging.info("Adding new event to the game") new_event = Event(location=location, initial_description=initial_description, entities=entity_list) game.add_event(new_event) return "Event added successfully." @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" @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()