255 lines
7.3 KiB
Python
255 lines
7.3 KiB
Python
# Game imports
|
|
from dice import Dice
|
|
from game import Game
|
|
from 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 = None
|
|
|
|
# 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)
|
|
}
|
|
|
|
@DeprecationWarning
|
|
def append_to_history(event: Dict[str, Any]):
|
|
"""Append a game event to the history file."""
|
|
history = []
|
|
if os.path.exists(HISTORY_FILE):
|
|
with open(HISTORY_FILE, "r", encoding="utf-8") as f:
|
|
try:
|
|
history = json.load(f)
|
|
except json.JSONDecodeError:
|
|
history = []
|
|
history.append(event)
|
|
with open(HISTORY_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(history, f, ensure_ascii=False, indent=2)
|
|
|
|
@DeprecationWarning
|
|
def read_history() -> list:
|
|
"""Read the game history from the file."""
|
|
if os.path.exists(HISTORY_FILE):
|
|
with open(HISTORY_FILE, "r", encoding="utf-8") as f:
|
|
try:
|
|
return json.load(f)
|
|
except json.JSONDecodeError:
|
|
return []
|
|
return []
|
|
|
|
# EVENTS TOOLS
|
|
@mcp.tool()
|
|
async def start_event(location:str, initial_description:str, entity_list:list[str]):
|
|
new_event = Event(location=location, initial_description=initial_description, entities=entity_list)
|
|
game.add_event(new_event)
|
|
|
|
@mcp.tool()
|
|
async def perform_action():
|
|
pass
|
|
|
|
# ITEM TOOLS
|
|
@mcp.tool()
|
|
async def create_item():
|
|
pass
|
|
|
|
@mcp.tool()
|
|
async def add_item_to_entity():
|
|
pass
|
|
|
|
# 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")
|
|
dice = Dice(n_faces)
|
|
|
|
if n_faces < 1:
|
|
raise ValueError("Number of faces must be at least 1")
|
|
elif n_faces == 1:
|
|
return 1
|
|
elif n_faces == 2:
|
|
return dice.head_or_tails()
|
|
else:
|
|
return dice.roll()
|
|
|
|
@mcp.tool()
|
|
async def get_entity_status():
|
|
pass
|
|
|
|
@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: str = "") -> Dict[str, Any]:
|
|
"""Create a new player. Need all the stats to function properly. Throw a d20 for every stats you don't have,
|
|
and a d6 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 with name={name}")
|
|
player = Player(name, strength, dexterity, intelligence, wisdom, charisma, hp, armor, speed)
|
|
logging.info(f"Created player: {player}")
|
|
game.players.append(player)
|
|
return player.serialize_dict()
|
|
|
|
@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: str = "") -> Dict[str, Any]:
|
|
"""Create a new NPC. Need all the stats to function properly. Throw a d20 for every stats you don't have,
|
|
and a d6 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 with name={name}")
|
|
npc = NPC(name, strength, dexterity, intelligence, wisdom, charisma, hp, armor, speed)
|
|
logging.info(f"Created NPC: {npc}")
|
|
game.npcs.append(npc)
|
|
return npc.serialize_dict()
|
|
|
|
@mcp.tool()
|
|
async def create_item(name: str, description: str, bonus: str) -> Dict[str, Any]:
|
|
"""Create a new item.
|
|
|
|
Args:
|
|
name: Name of the item
|
|
description: Description of the item
|
|
bonus: Bonus of the item ex: strength+1,hp+5
|
|
"""
|
|
logging.info(f"Creating item with name={name}")
|
|
item = Item(name, description, bonus)
|
|
logging.info(f"Created item: {item}")
|
|
game.items.append(item)
|
|
return item.serialize_dict()
|
|
|
|
@mcp.tool()
|
|
async def add_item_to_player(player_name: str, item_name: str) -> Dict[str, Any]:
|
|
"""Add an item to a player's inventory.
|
|
|
|
Args:
|
|
player_name: The name of the player to add the item to
|
|
item_name: The name of the item to add
|
|
"""
|
|
logging.info(f"Adding item {item_name} to player {player_name}")
|
|
return {"status": "Item added"}
|
|
|
|
@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"
|
|
|
|
|
|
# Example MCP tool to add an event to history
|
|
@mcp.tool()
|
|
async def add_event_to_history(event: Dict[str, Any]) -> str:
|
|
"""Add a game event to the history resource."""
|
|
append_to_history(event)
|
|
return "Event added to history."
|
|
|
|
# Example MCP tool to read history
|
|
@mcp.tool()
|
|
async def get_game_history() -> list:
|
|
"""Get the full game history."""
|
|
return read_history()
|
|
|
|
@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() |