# 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 import FastMCP import json import os # Constants SAVE_PATH = "game_" # 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(), ] ) # GLOBAL GAME TOOLS & RESOURCES @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: logging.info(f"OSError: " + str(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: logging.info(f"OSError: " + str(e)) return { "success": False, "error": str(e) } @mcp.tool(name="game_state", description="Retrieves the current game state") async def get_game_state() -> str: """Get the current game state as a serialized string.""" logging.info("Fetching current game state") return game.serialize() # EVENTS TOOLS & RESOURCES @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) return { "success": True, "new_event": new_event.serialize() } @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 """ try: logging.info(f"Entity {src_entity_id} is performing an attack on {target_entity_id} using {attack_type}") damage_amount = 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}", "damage_amount": damage_amount } except IndexError as e: logging.info(f"IndexError: " + str(e)) return { "success": False, "error": str(e) } except ReferenceError as e: logging.info(f"ReferenceError: " + str(e)) return { "success": False, "error": str(e) } @mcp.tool() async def perform_simple_action(entity_id:str, stat:Stat, difficulty:int, roll:int, description:str): """Perform a simple 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 like Stat.INTELLIGENCE or Stat.DEXTERITY) difficulty: The difficulty of the test according to a d20 roll: The value of the d20 launched for the test description: The description of the action being performed by the entity (like opening a door) """ try: logging.info(f"Entity {entity_id} is performing a test action on stat {stat} with difficulty {difficulty}") action_performed, test_result = game.simple_action(src=entity_id, stat=stat, difficulty=difficulty, roll=roll, description=description) return { "success": True, "action_performed": action_performed, "test_result": test_result, "initial_difficulty": difficulty } except IndexError as e: logging.info(f"IndexError: " + str(e)) return { "success": False, "error": str(e) } except ReferenceError as e: logging.info(f"ReferenceError: " + str(e)) return { "success": False, "error": str(e) } @mcp.tool() async def perform_stat_modification(entity_id:str, stat:Stat, value:int=0): """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) """ try: 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}" } except IndexError as e: logging.info(f"IndexError: " + str(e)) return { "success": False, "error": str(e) } except ReferenceError as e: logging.info(f"ReferenceError: " + str(e)) return { "success": False, "error": str(e) } @mcp.tool(name="current_event") async def get_current_event() -> Dict[str, Any]: """Get the current event in the game.""" try: logging.info("Getting current event") current_event = game.get_current_event() return { "success": True, "current_event": current_event.serialize() } except IndexError as e: logging.info(f"IndexError: " + str(e)) return { "success": False, "error": str(e) } # ITEM TOOLS @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} """ try: 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 } except ReferenceError as e: logging.info(f"ReferenceError: " + str(e)) return { "success": False, "error": str(e) } @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 """ try: 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 { "success": True, "msg": f"Item #{item_id} equipped to entity #{entity_id}" } except ReferenceError as e: logging.info(f"ReferenceError: " + str(e)) return { "success": False, "error": str(e) } @mcp.tool(name="item_properties") async def get_item_properties(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 """ try: logging.info(f"Getting info for item {item_id}") item = game.get_item(item_id) return { "success": True, "item_properties": item.serialize() } except ReferenceError as e: logging.info(f"ReferenceError: " + str(e)) return { "success": False, "error": str(e) } # OTHER UTILS (Dice & Entities) @mcp.tool() async def throw_a_dice(n_faces: int) -> Any: """Throw a dice with n faces. The number of faces should be greater than one! 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" } else: return { "success": True, "roll_result": Dice.roll(n_faces) } @mcp.tool() async def toss_coin(): """Throw a coin when you need head or tails for decision making.""" return { "success": True, "toss_result": Dice.head_or_tails() } @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, add_to_event:bool=True): """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 the armor and a d100 for 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 add_to_event: Boolean deciding whether or not to add the entity to the current event """ logging.info(f"Creating player named {name}") try: item = game.get_item(item_id) # Check if item exists player_id = game.create_player(name=name, strength=strength, dexterity=dexterity, intelligence=intelligence, wisdom=wisdom, charisma=charisma, hp=20 + hp, armor=50 + armor, speed=speed, equipped_item=item) 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") if add_to_event: game.add_entity_to_event(player_id) logging.info(f"Player #{player_id} added to current event!") 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, add_to_event:bool=True): """Create a new NPC. Need all the stats to function properly. Throw a d20 for every stats you don't have, and a d50 for the armor and a d100 for 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 add_to_event: Boolean deciding whether or not to add the entity to the current event """ 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=20 + hp, armor=50 + armor, speed=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") if add_to_event: game.add_entity_to_event(npc_id) logging.info(f"Player #{npc_id} added to current event!") 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(name="all_entities_status") async def get_all_entities_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(name="entity_status") async def get_entity_status(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 """ try: logging.info(f"Getting info for entity {entity_id}") entity = game.get_entity(entity_id) return { "success": True, "entity_status": entity.serialize() } except ReferenceError as e: logging.info(f"ReferenceError: " + str(e)) return { "success": False, "error": str(e) } def main(): # Initialize and run the server mcp.run(transport="stdio") if __name__ == "__main__": main()