From 61608231604bd6002af6a2680e3eb001ae63e12c Mon Sep 17 00:00:00 2001 From: KuMiShi Date: Sat, 31 Jan 2026 15:53:32 +0100 Subject: [PATCH] API correction & mcp validation --- server.py | 340 ++++++++++++++++++++++++++++---------------------- utils/game.py | 27 +++- 2 files changed, 211 insertions(+), 156 deletions(-) diff --git a/server.py b/server.py index d8202a1..33a6412 100644 --- a/server.py +++ b/server.py @@ -14,8 +14,7 @@ import json import os # Constants -HISTORY_FILE = "game_history.json" -SAVE_PATH = "save_" +SAVE_PATH = "game_" # Global Parameters mcp = FastMCP("wyvern-castle") @@ -30,7 +29,7 @@ logging.basicConfig( ] ) -# SAVING & LOADING GAME STATE +# GLOBAL GAME TOOLS & RESOURCES @mcp.tool() async def load_game(slot:int): """Loads an already existing game. @@ -48,6 +47,7 @@ async def load_game(slot:int): "msg": f"{path} as been successfully loaded!" } except OSError as e: + logging.info(f"OSError: " + str(e)) return { "success": False, "error": str(e) @@ -70,13 +70,19 @@ async def save_game(slot:int): "msg": f"{path} as been successfully saved!" } except OSError as e: + logging.info(f"OSError: " + str(e)) return { "success": False, "error": str(e) } +@mcp.resource(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 +# 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. @@ -87,6 +93,10 @@ async def start_event(location:str, initial_description:str, entity_list:list[st """ 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): @@ -96,53 +106,62 @@ async def perform_attack(src_entity_id:str, target_entity_id:str, attack_type:St 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}") - dmg = 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": dmg - } + 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_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. +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) - difficulty: The difficulty of 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) """ - 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 - } + 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): +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: @@ -150,22 +169,115 @@ async def perform_stat_modification(entity_id:str, stat:Stat, value:int): 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}" - } - + 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.resource(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) + } -# OTHER UTILS +@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.resource(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. If n==2 its a coin toss. + """Throw a dice with n faces. The number of faces should be greater than one! Args: n_faces: Number of faces of the dice @@ -177,11 +289,6 @@ async def throw_a_dice(n_faces: int) -> Any: "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, @@ -189,17 +296,12 @@ async def throw_a_dice(n_faces: int) -> Any: } @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] +async def toss_coin(): + """Throw a coin when you need head or tails for decision making.""" return { - "players": players, - "npcs": npcs - } + "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): @@ -279,105 +381,39 @@ async def create_npc(name: str, strength: int, dexterity: int, intelligence: int "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} +@mcp.resource(name="all_entities_status") +async def get_all_entities_status(): """ - 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() + 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 { - "success": True, - "item_properties": item_dict - } + "players": players, + "npcs": npcs + } -@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]: +@mcp.resource(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 """ - 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: + try: + logging.info(f"Getting info for entity {entity_id}") + entity = game.get_entity(entity_id) return { "success": True, - "event_properties": event.serialize() + "entity_status": entity.serialize() } - else: + except ReferenceError as e: + logging.info(f"ReferenceError: " + str(e)) return { "success": False, - "error": "No current event" + "error": str(e) } - -@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." - - - - - - - def main(): # Initialize and run the server diff --git a/utils/game.py b/utils/game.py index 218fbbf..6fba99e 100644 --- a/utils/game.py +++ b/utils/game.py @@ -100,7 +100,7 @@ class Game(Serializable): def add_event(self, new_event:Event): self.events.append(new_event) self.update_turn_order() - self.turn_idx = 0 + self.turn_idx = 0 def check_turn_ended(self): if self.turn_idx == len(self.turn_order)-1: @@ -205,12 +205,31 @@ class Game(Serializable): if turn_finished: self.update_turn_order() - def simple_action(self, src:str, description:str): + def simple_action(self, src:str, stat:Stat, difficulty:int, roll:int, description:str): if not self.is_turn_coherent(src): raise ReferenceError(f"Entity #{src} tried performing an action while it was #{self.turn_order[self.turn_idx]}'s turn!") - turn_finished = self.get_current_event().perform_action(TurnAction.BASIC, src, description=description) + src_entity = self.get_entity(src) + stat_boost = 0 # Between 0 and 5 + match(stat): + case Stat.STRENGTH: + stat_boost = src_entity.get_strength() / 4 + case Stat.INTELLIGENCE: + stat_boost = src_entity.get_intelligence() / 4 + case Stat.DEXTERITY: + stat_boost = src_entity.get_dexterity() / 4 + case Stat.WISDOM: + stat_boost = src_entity.get_wisdom() / 4 + case Stat.CHARISMA: + stat_boost = src_entity.get_charisma() / 4 + test_result = roll + stat_boost + action_performed = difficulty <= test_result + + additional_info = f", (Test difficulty: {difficulty}, Player roll: {test_result})" + + turn_finished = self.get_current_event().perform_action(TurnAction.BASIC, src, description=description+additional_info) self.turn_idx += 1 if turn_finished: self.update_turn_order() - #TODO: Add State Summary as Resource? \ No newline at end of file + + return action_performed, test_result \ No newline at end of file