API correction & mcp validation

This commit is contained in:
KuMiShi
2026-01-31 15:53:32 +01:00
parent def1a4be3e
commit 6160823160
2 changed files with 211 additions and 156 deletions

340
server.py
View File

@@ -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

View File

@@ -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?
return action_performed, test_result