423 lines
14 KiB
Python
423 lines
14 KiB
Python
# 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() |