tweaks for server using serialization
This commit is contained in:
3
.gitignore
vendored
3
.gitignore
vendored
@@ -1,3 +1,6 @@
|
|||||||
# UV properties
|
# UV properties
|
||||||
.venv/
|
.venv/
|
||||||
.python-version
|
.python-version
|
||||||
|
__pycache__/
|
||||||
|
uv.lock
|
||||||
|
*.json
|
||||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
12
dice.py
12
dice.py
@@ -1,15 +1,13 @@
|
|||||||
import random as rd
|
import random as rd
|
||||||
|
|
||||||
class Dice():
|
class Dice():
|
||||||
def __init__(self):
|
def __init__(self, num_faces:int=20):
|
||||||
raise TypeError("Un dé ne peut pas être instanciée!")
|
self.num_faces = num_faces
|
||||||
|
|
||||||
@staticmethod
|
def roll(self):
|
||||||
def roll(self, num_faces=20):
|
return rd.randrange(start=1, stop=self.num_faces+1, step=1)
|
||||||
return rd.randrange(start=1, stop=num_faces+1, step=1)
|
|
||||||
|
|
||||||
@staticmethod
|
def head_or_tails(self):
|
||||||
def head_or_tails():
|
|
||||||
result = rd.randint(0,1)
|
result = rd.randint(0,1)
|
||||||
if result: # true
|
if result: # true
|
||||||
return "head" # face
|
return "head" # face
|
||||||
|
|||||||
6
game.py
6
game.py
@@ -3,4 +3,8 @@ from dice import Dice
|
|||||||
|
|
||||||
class Game(Serializable):
|
class Game(Serializable):
|
||||||
def __init__(self, seed:int=42):
|
def __init__(self, seed:int=42):
|
||||||
pass
|
self.players = []
|
||||||
|
self.npcs = []
|
||||||
|
self.items = []
|
||||||
|
|
||||||
|
|
||||||
@@ -11,7 +11,7 @@ class Serializable:
|
|||||||
instance.deserialize_dict(data)
|
instance.deserialize_dict(data)
|
||||||
return instance
|
return instance
|
||||||
|
|
||||||
def serialize(self, file) -> str:
|
def serialize(self, file=None) -> str:
|
||||||
"""Serializes the object and all nested Serializable objects to JSON."""
|
"""Serializes the object and all nested Serializable objects to JSON."""
|
||||||
def serialize_value(value: Any) -> Any:
|
def serialize_value(value: Any) -> Any:
|
||||||
if isinstance(value, Serializable):
|
if isinstance(value, Serializable):
|
||||||
@@ -24,7 +24,9 @@ class Serializable:
|
|||||||
return value
|
return value
|
||||||
|
|
||||||
attrs = {k: serialize_value(v) for k, v in self.__dict__.items() if not k.startswith('_')}
|
attrs = {k: serialize_value(v) for k, v in self.__dict__.items() if not k.startswith('_')}
|
||||||
return json.dumps(attrs, file, ensure_ascii=False, indent=2)
|
if file:
|
||||||
|
json.dump(attrs, file, ensure_ascii=False, indent=2)
|
||||||
|
return json.dumps(attrs, ensure_ascii=False, indent=2)
|
||||||
|
|
||||||
def serialize_dict(self) -> Dict[str, Any]:
|
def serialize_dict(self) -> Dict[str, Any]:
|
||||||
"""Serializes the object to a dictionary (for nested serialization)."""
|
"""Serializes the object to a dictionary (for nested serialization)."""
|
||||||
|
|||||||
102
server.py
102
server.py
@@ -7,7 +7,9 @@ from player import Player
|
|||||||
from item import Item
|
from item import Item
|
||||||
from game import Game
|
from game import Game
|
||||||
from npc import NPC
|
from npc import NPC
|
||||||
|
from serializable import Serializable
|
||||||
|
import json
|
||||||
|
import os
|
||||||
mcp = FastMCP("wyvern-castle")
|
mcp = FastMCP("wyvern-castle")
|
||||||
game = Game()
|
game = Game()
|
||||||
|
|
||||||
@@ -19,6 +21,32 @@ logging.basicConfig(
|
|||||||
]
|
]
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# History file path
|
||||||
|
HISTORY_FILE = "game_history.json"
|
||||||
|
|
||||||
|
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)
|
||||||
|
|
||||||
|
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 []
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
async def throw_a_dice(n_faces: int) -> Any:
|
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. If n==2 its a coin toss.
|
||||||
@@ -27,7 +55,7 @@ async def throw_a_dice(n_faces: int) -> Any:
|
|||||||
n_faces: Number of faces of the dice
|
n_faces: Number of faces of the dice
|
||||||
"""
|
"""
|
||||||
logging.info(f"Throwing a dice with {n_faces} faces")
|
logging.info(f"Throwing a dice with {n_faces} faces")
|
||||||
dice = Dice()
|
dice = Dice(n_faces)
|
||||||
|
|
||||||
if n_faces < 1:
|
if n_faces < 1:
|
||||||
raise ValueError("Number of faces must be at least 1")
|
raise ValueError("Number of faces must be at least 1")
|
||||||
@@ -36,36 +64,12 @@ async def throw_a_dice(n_faces: int) -> Any:
|
|||||||
elif n_faces == 2:
|
elif n_faces == 2:
|
||||||
return dice.head_or_tails()
|
return dice.head_or_tails()
|
||||||
else:
|
else:
|
||||||
return dice.roll(n_faces)
|
return dice.roll()
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
async def mult(a: int, b: int) -> int:
|
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]:
|
||||||
"""Multiply two numbers.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
a: First number
|
|
||||||
b: Second number
|
|
||||||
"""
|
|
||||||
logging.info(f"Calling mult with a={a}, b={b}")
|
|
||||||
result = a * b
|
|
||||||
logging.info(f"mult result: {result}")
|
|
||||||
return result
|
|
||||||
|
|
||||||
@mcp.tool()
|
|
||||||
async def add(a: int, b: int) -> int:
|
|
||||||
"""Add two numbers.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
a: First number
|
|
||||||
b: Second number
|
|
||||||
"""
|
|
||||||
logging.info(f"Calling add with a={a}, b={b}")
|
|
||||||
result = a + b
|
|
||||||
logging.info(f"add result: {result}")
|
|
||||||
return result
|
|
||||||
|
|
||||||
@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 = None) -> Dict[str, Any]:
|
|
||||||
"""Create a new player. Need all the stats to function properly. Throw a d20 for every stats you don't have,
|
"""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.
|
and a d6 for hp, armor and speed.
|
||||||
|
|
||||||
@@ -84,10 +88,11 @@ async def create_player(name: str, strength: int, dexterity: int, intelligence:
|
|||||||
logging.info(f"Creating player with name={name}")
|
logging.info(f"Creating player with name={name}")
|
||||||
player = Player(name, strength, dexterity, intelligence, wisdom, charisma, hp, armor, speed)
|
player = Player(name, strength, dexterity, intelligence, wisdom, charisma, hp, armor, speed)
|
||||||
logging.info(f"Created player: {player}")
|
logging.info(f"Created player: {player}")
|
||||||
|
game.players.append(player)
|
||||||
return player.serialize_dict()
|
return player.serialize_dict()
|
||||||
|
|
||||||
@mcp.tool()
|
@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 = None) -> Dict[str, Any]:
|
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,
|
"""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.
|
and a d6 for hp, armor and speed.
|
||||||
|
|
||||||
@@ -106,6 +111,7 @@ async def create_npc(name: str, strength: int, dexterity: int, intelligence: int
|
|||||||
logging.info(f"Creating NPC with name={name}")
|
logging.info(f"Creating NPC with name={name}")
|
||||||
npc = NPC(name, strength, dexterity, intelligence, wisdom, charisma, hp, armor, speed)
|
npc = NPC(name, strength, dexterity, intelligence, wisdom, charisma, hp, armor, speed)
|
||||||
logging.info(f"Created NPC: {npc}")
|
logging.info(f"Created NPC: {npc}")
|
||||||
|
game.npcs.append(npc)
|
||||||
return npc.serialize_dict()
|
return npc.serialize_dict()
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
@@ -120,6 +126,7 @@ async def create_item(name: str, description: str, bonus: str) -> Dict[str, Any]
|
|||||||
logging.info(f"Creating item with name={name}")
|
logging.info(f"Creating item with name={name}")
|
||||||
item = Item(name, description, bonus)
|
item = Item(name, description, bonus)
|
||||||
logging.info(f"Created item: {item}")
|
logging.info(f"Created item: {item}")
|
||||||
|
game.items.append(item)
|
||||||
return item.serialize_dict()
|
return item.serialize_dict()
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
@@ -133,6 +140,39 @@ async def add_item_to_player(player_name: str, item_name: str) -> Dict[str, Any]
|
|||||||
logging.info(f"Adding item {item_name} to player {player_name}")
|
logging.info(f"Adding item {item_name} to player {player_name}")
|
||||||
return {"status": "Item added"}
|
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():
|
def main():
|
||||||
# Initialize and run the server
|
# Initialize and run the server
|
||||||
mcp.run(transport="stdio")
|
mcp.run(transport="stdio")
|
||||||
|
|||||||
Reference in New Issue
Block a user