mcp_server.py

"""
Example: MCP Server for Text Adventures

A complete MCP server that exposes text adventure games via tools.
This demonstrates a full-featured server with memory, mapping, and inventory.
"""

import sys
import os

# Add parent directory to path to import games module
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from fastmcp import FastMCP
from games.zork_env import TextAdventureEnv, list_available_games

import asyncio

# Get game from environment variable (default: zork1)
INITIAL_GAME = os.environ.get("GAME", "zork1")

# Create the MCP server
mcp = FastMCP("Text Adventure Server")


class GameState:
    """Manages the text adventure game state and exploration data."""
    
    def __init__(self, game: str = "zork1"):
        self.game_name = game
        self.env = TextAdventureEnv(game)
        self.state = self.env.reset()
        self.history: list[tuple[str, str]] = []
        self.explored_locations: dict[str, set[str]] = {}
        self.current_location: str = self._extract_location(self.state.observation)
    
    def _extract_location(self, observation: str) -> str:
        """Extract location name from observation (usually first line)."""
        lines = observation.strip().split('\n')
        return lines[0] if lines else "Unknown"

    def clean_text(self, text: str) -> str:
        """Transforme 'obj91: pants parent87...' en 'pants'."""
        text = str(text).lower()
        if ":" in text:
            # On prend la partie après le matricule obj
            name_part = text.split(":", 1)[1].strip()
            # On coupe avant les métadonnées techniques
            for stop_word in [" parent", " sibling", " child", " attributes"]:
                if stop_word in name_part:
                    name_part = name_part.split(stop_word)[0].strip()
            return name_part
        return text
    
    def take_action(self, action: str) -> str:
        """Execute a game action and return the result."""
        self.state = self.env.step(action)
        result = self.state.observation
        
        # Track history
        self.history.append((action, result))
        if len(self.history) > 50:
            self.history = self.history[-50:]
        
        # Update map
        new_location = self._extract_location(result)
        if action in ["north", "south", "east", "west", "up", "down", 
                      "enter", "exit", "n", "s", "e", "w", "u", "d"]:
            if self.current_location not in self.explored_locations:
                self.explored_locations[self.current_location] = set()
            if new_location != self.current_location:
                self.explored_locations[self.current_location].add(f"{action} -> {new_location}")
        self.current_location = new_location
        
        return result
    
    # def get_heuristic_actions(self) -> str:
    #         """
    #         Génère des actions ultra-précises en fusionnant l'observation textuelle 
    #         et le scan de la mémoire vive du jeu (God Mode).
    #         """

    #         self._debug_log("HEURISTIC - STRUCTURED")

    #         # visible_objs = structured_data.get("takeable_objects", [])
    #         # puzzles = structured_data.get("puzzle_clues", [])
    #         # exits = structured_data.get("visible_exits", [])

    #         categorized_actions: Dict[str, Set[str]] = {
    #         "navigation": {"north", "south", "east", "west","northeast", "northwest", "southeast", "southwest","up", "down", "in", "out"},
    #         "system": {"look", "inventory", "wait"},
    #         "inventory_interactions": set(),
    #         "environment_interactions": set(),
    #         "contextual": set()
    #         }

    #         BASE_ACTIONS = {
    #         "look",
    #         "inventory",
    #         "wait"
    #         }

    #         DIRECTIONS = {
    #         "north", "south", "east", "west","northwest","northeast","southeast","southwest"
    #         "up", "down", "in", "out",
    #         "n", "s", "e", "w", "u", "d","nw","ne","se","sw"
    #         }

    #         INVENTORY_VERBS = {
    #             "examine",
    #             "use",
    #             "drop",
    #         }

    #         ROOM_VERBS = {
    #         "examine",
    #         "open",
    #         "close",
    #         "push",
    #         "pull",
    #         "take",
    #         }

    #         inventory_items = [i['name'] for i in self.state.inventory_raw] # Liste de dicts {'name':...}
    #         room_items = structured_data.get("raw_ram_objects", [])

    #         available_verbs = ["examine", "take", "drop", "look", "inventory", "wait"]

    #         actions: Set[str] = set()

    #         obs = self.state.observation.lower()

    #         clean_inventory = [self.clean_text(i) for i in self.state.inventory]

    #         for item in clean_inventory:
    #             for verb in INVENTORY_VERBS:
    #                 categorized_actions["inventory_interactions"].add(f"{verb} {item}")

    #         if "door" in obs:
    #             categorized_actions["environment_interactions"] |= {"open door", "close door", "examine door"}
            
    #         if "window" in obs:
    #             categorized_actions["environment_interactions"] |= {"open window", "examine window"}
            
    #         if "stairs" in obs or "staircase" in obs:
    #             actions |= {"up", "down"}
    #             categorized_actions["contextual"] |= {"up stairs", "down stairs"}

    #         if "dark" in obs or "can't see" in obs:
    #             for item in clean_inventory:
    #                 if any(k in item for k in ["lamp", "torch", "light"]):
    #                     categorized_actions["inventory_interactions"].add(f"use {item}")
    #                     categorized_actions["inventory_interactions"].add(f"turn on {item}")

    #         if any(k in obs for k in ["man", "woman", "person", "creature"]):
    #             categorized_actions["contextual"] |= {"talk", "listen", "attack"}

    #         if any(k in obs for k in ["noise", "sound", "voice", "squeal"]):
    #             categorized_actions["contextual"].add("listen")
    #             categorized_actions["contextual"].add("examine noise")
            
    #         # --- 7. Anti-boucle simple ---
    #         if self.history:
    #             last_action = self.history[-1]
    #             if last_action in actions:
    #                 actions.remove(last_action)


    #         final_output = []
    #         for category, acts in categorized_actions.items():
    #             if acts:
    #                 # On trie et on formate chaque catégorie
    #                 sorted_acts = sorted(list(acts))
    #                 category_text = f"\n{category.replace('_', ' ').upper()}:\n"
    #                 category_text += "\n".join([f"  - {a}" for a in sorted_acts])
    #                 final_output.append(category_text)

    #         return "\n".join(final_output)

    def get_heuristic_actions(self, structured_data: dict) -> str:
        """
        Fournit au LLM les briques nécessaires pour construire ses propres actions
        en se basant sur la vérité de la RAM.
        """
        self._debug_log("HEURISTIC - COMPONENT BASED")

        # 1. Récupération des objets réels (RAM + Inventaire)
        # On utilise les noms exacts venant de la RAM pour éviter les erreurs
        inventory_items = [i['name'] for i in self.state.inventory_raw] # Liste de dicts {'name':...}
        room_items = structured_data.get("raw_ram_objects", [])
        
        # 2. Définition des Verbes (Le "Lexique")
        # On ne donne que les verbes pertinents à la situation
        available_verbs = ["examine", "take", "drop", "look", "inventory", "wait"]
        
        obs = self.state.observation.lower()
        
        if any(k in obs for k in ["door", "window", "container", "box"]):
            available_verbs += ["open", "close", "unlock"]
        
        if any(k in obs for k in ["noise", "sound", "squeal", "quiet"]):
            available_verbs.append("listen")
            
        if "dark" in obs or "torch" in str(inventory_items):
            available_verbs += ["turn on", "light"]

        # 3. Construction du message d'aide
        final_output = [
            "### ACTION CONSTRUCTION KIT",
            f"VERBS: {', '.join(sorted(set(available_verbs)))}",
            f"INVENTORY: {', '.join(inventory_items) if inventory_items else 'Empty'}",
            f"ROOM OBJECTS: {', '.join(room_items) if room_items else 'None visible'}",
            "\nNAVIGATION: n, s, e, w, ne, nw, se, sw, u, d, in, out",
            "\nEXAMPLES: 'examine tracks', 'turn on torch', 'open mailbox', 'listen'"
        ]

        return "\n".join(final_output)


    def _debug_log(self, message: str):
        """Envoie un log de debug vers stderr pour ne pas polluer le flux MCP."""
        print(f"DEBUG_HEURISTIC: {message}", file=sys.stderr, flush=True)

    def get_memory(self) -> str:
        """Get a summary of current game state."""
        recent = self.history[-5:] if self.history else []
        recent_str = "\n".join([f"  > {a} -> {r[:60]}..." for a, r in recent]) if recent else "  (none yet)"
        
        return f"""Current State:
- Location: {self.current_location}
- Score: {self.state.score} points
- Moves: {self.state.moves}
- Game: {self.game_name}

Recent Actions:
{recent_str}

Current Observation:
{self.state.observation}"""
    
    def get_map(self) -> str:
        """Get a map of explored locations."""
        if not self.explored_locations:
            return "Map: No locations explored yet. Try moving around!"
        
        lines = ["Explored Locations and Exits:"]
        for loc, exits in sorted(self.explored_locations.items()):
            lines.append(f"\n* {loc}")
            for exit_info in sorted(exits):
                lines.append(f"    -> {exit_info}")
        
        lines.append(f"\n[Current] {self.current_location}")
        return "\n".join(lines)
    
    def get_inventory(self) -> str:
        """Get current inventory."""
        items = self.state.inventory if hasattr(self.state, 'inventory') and self.state.inventory else []
        
        if not items:
            return "Inventory: You are empty-handed."
        
        item_names = []
        for item in items:
            item_str = str(item)
            item_lower = item_str.lower()
            if "parent" in item_lower:
                idx = item_lower.index("parent")
                name = item_str[:idx].strip()
                if ":" in name:
                    name = name.split(":", 1)[1].strip()
                item_names.append(name)
            elif ":" in item_str:
                name = item_str.split(":")[1].strip()
                item_names.append(name)
            else:
                item_names.append(item_str)
        
        return f"Inventory: {', '.join(item_names)}"


# Global game state
_game_state: GameState | None = None


def get_game() -> GameState:
    """Get or initialize the game state."""
    global _game_state
    if _game_state is None:
        _game_state = GameState(INITIAL_GAME)
    return _game_state


# =============================================================================
# MCP Tools
# =============================================================================

@mcp.tool()
def play_action(action: str) -> str:
    """
    Execute a game action in the text adventure.
    
    Args:
        action: The command to execute (e.g., 'north', 'take lamp', 'open mailbox')
    
    Returns:
        The game's response to your action
    """
    game = get_game()
    result = game.take_action(action)
    
    # Add score info
    score_info = f"\n\n[Score: {game.state.score} | Moves: {game.state.moves}]"
    
    if game.state.reward > 0:
        score_info = f"\n\n+{game.state.reward} points! (Total: {game.state.score})"
    
    done_info = ""
    if game.state.done:
        done_info = "\n\nGAME OVER"
    
    return result + score_info + done_info


@mcp.tool()
def memory() -> str:
    """
    Get a summary of the current game state.
    
    Returns location, score, moves, recent actions, and current observation.
    """
    return get_game().get_memory()


@mcp.tool()
def get_map() -> str:
    """
    Get a map showing explored locations and connections.
    
    Useful for navigation and avoiding getting lost.
    """
    return get_game().get_map()


@mcp.tool()
def inventory() -> str:
    """
    Check what items you are currently carrying.
    """
    return get_game().get_inventory()

# Dans mcp_server.py

# @mcp.tool()
# def get_valid_actions_cheat() -> str:
#     """[GOD MODE] Liste des actions syntaxiques via Spacy/Jericho."""
#     game = get_game()
#     # On appelle la méthode de TON wrapper TextAdventureEnv
#     actions = game.env.get_valid_actions()
#     return "ACTIONS VALIDES :\n" + ", ".join(actions)
 

@mcp.tool()
async def get_valid_actions_cheat() -> str:
    """
    [GOD MODE] Returns the list of ALL valid actions currently possible.
    """
    # On utilise le verrou pour empêcher le calcul parallèle
    try:
        game = get_game()
        jericho_env = game.env.env 
        
        # Jericho est synchrone, on l'exécute normalement
        valid_actions = jericho_env.get_valid_actions(use_parallel=False)

        if not valid_actions:
            return "Jericho returned an empty list of actions."
            
        return "VALID ACTIONS:\n" + ", ".join(valid_actions)

    except Exception as e:
        return f"Error retrieving valid actions: {str(e)}"



# @mcp.tool()
# def get_valid_actions_cheat() -> str:
#     """
#     [GOD MODE] Returns the absolute list of logical actions 
#     by scanning the game's internal memory tree.
#     """
#     return get_game().get_heuristic_actions()
        
@mcp.tool()
def detect_objects_cheat() -> str:
    """
    [GOD MODE] Scans the game memory to find interactive objects in the current location.
    Returns their internal game names.
    """
    try:
        # identify_interactive_objects renvoie les objets avec lesquels on peut interagir
        # Note: Cette fonction Jericho n'est pas dispo pour TOUS les jeux, mais souvent pour Zork/LostPig
        objs = env.identify_interactive_objects() 
        return f"OBJECTS IN MEMORY: {objs}"
    except Exception as e:
        return f"Feature not available: {e}"


@mcp.tool()
def cheat_sense_surroundings() -> str:
    try:
        game = get_game()
        j_env = game.env.env
        # player_loc = j_env.get_player_location() # Ex: 93
        player_loc = 164
        
        # On récupère tous les objets du jeu (souvent plusieurs centaines)
        world_objs = j_env.get_world_objects()
        contents = []
        
        for obj in world_objs:
            # On vérifie si le parent est la pièce actuelle
            if obj.parent == player_loc:
                name = str(obj.name).replace('\x00', '').strip()
                # On ignore le joueur lui-même et les noms vides
                if name and "player" not in name.lower():
                    contents.append(f"{name} (ID:{obj.num})")

        return "VISIBLE IN MEMORY: " + (", ".join(contents) if contents else "Empty Room")
    except Exception as e:
        return f"Error: {e}"

@mcp.tool()
def get_game_dictionary() -> list[str]:
    """Récupère tous les mots compris par le jeu."""
    game = get_game()
    # Récupère les objets DictionaryWord
    words = game.env.env.get_dictionary()
    return [w.word for w in words]

@mcp.tool()
def get_action_grammar() -> str:
    """Récupère les modèles de phrases autorisées (ex: 'put X on Y')."""
    game = get_game()
    return game.env.env.bindings.get('grammar', 'Non disponible') 


@mcp.tool()
def cheat_get_valid_exits() -> str:
    """[GOD MODE] Analyse les propriétés de la pièce actuelle pour trouver les sorties."""
    try:
        game = get_game()
        j_env = game.env.env
        # room_id = j_env.get_player_location()
        room_id = 166
        
        # On définit les directions cardinales
        directions = ['north', 'south', 'east', 'west', 'up', 'down', 'ne', 'nw', 'se', 'sw', 'in', 'out']
        valid_exits = []

        # On utilise une méthode de lecture de propriété si elle existe, 
        # sinon on se rabat sur une analyse de l'observation textuelle 'propre'
        for d in directions:
            try:
                # Tentative de lecture directe de la destination pour la direction d
                dest_id = j_env.get_next_location(room_id, d)
                if dest_id > 0:
                    valid_exits.append(d)
            except:
                continue

        # FALLBACK : Si le moteur Jericho est vraiment trop bridé pour les sorties
        if not valid_exits:
            obs = game.state.observation.lower()
            # On cherche des patterns classiques comme "Exit: North" ou "to the south"
            for d in directions:
                if f" {d}" in obs and ("exit" in obs or "lead" in obs or "way" in obs):
                    valid_exits.append(d)
        
        return f"STRICT VALID EXITS: {', '.join(valid_exits) if valid_exits else 'None (Try exploring manually)'}"
    except Exception as e:
        return f"Error: {e}"

@mcp.tool()
def cheat_get_status() -> str:
    """
    [GOD MODE] Returns the internal state: Score, Moves, and Inventory.
    """
    try:
        score = env.get_score()
        # Note: inventory est une liste d'objets Jericho
        inv_objs = env.get_inventory()
        inv_names = [obj.name for obj in inv_objs]
        
        return f"""
SCORE: {score}
INVENTORY (INTERNAL): {inv_names}
"""
    except Exception as e:
        return str(e)


@mcp.tool()
def get_location_info() -> dict:
    """
    Scan la pièce actuelle et le contenu immédiat des objets (Niveau +1).
    Permet au LLM de déduire les types via les noms et la structure.
    """
    try:
        game = get_game()
        env = game.env.env
        
        def clean_name(name):
            return str(name).replace('\x00', '').strip() if name else "Inconnu"

        # 1. Localisation racine
        loc = env.get_player_location()
        room_id = loc.num if hasattr(loc, 'num') else int(loc)
        room_obj = env.get_object(room_id)
        
        world_objs = env.get_world_objects()

        # 2. Collecte des objets de la pièce
        detected_elements = []
        for obj in world_objs:
            # On ne prend que les objets dont le parent est la pièce actuelle
            if obj.parent == room_id:
                name = clean_name(obj.name)
                if "player" in name.lower(): continue

                # Scan de niveau +1 : on regarde juste si cet objet contient des trucs
                sub_contents = []
                for sub_obj in world_objs:
                    if sub_obj.parent == obj.num:
                        sub_contents.append({
                            "id": sub_obj.num,
                            "name": clean_name(sub_obj.name)
                        })

                # On construit une fiche simple
                element = {
                    "id": obj.num,
                    "name": name,
                    "contains_count": len(sub_contents),
                    "contents": sub_contents # Liste des noms/id immédiats
                }
                detected_elements.append(element)

        return {
            "status": "success",
            "location": {
                "id": room_id,
                "name": clean_name(room_obj.name)
            },
            "detected_objects": detected_elements,
            "inventory": [
                {"id": i.num, "name": clean_name(i.name)} 
                for i in env.get_inventory()
            ],
            "world_hash": env.get_world_state_hash()
        }

    except Exception as e:
        return {"status": "error", "message": str(e)}
# @mcp.tool()
# def get_location_info() -> dict:
#     """
#     Retourne la vérité absolue de la RAM : Localisation, Objets présents, 
#     Inventaire et Hash d'état.
#     """
#     try:
#         game = get_game()
#         env = game.env.env
        
#         # 1. Récupération de la localisation précise
#         loc = env.get_player_location()
#         room_id = loc.num if hasattr(loc, 'num') else int(loc)
#         room_obj = env.get_object(room_id)
#         room_name = str(room_obj.name).replace('\x00', '').strip() if room_obj else "Unknown Room"
        
#         # 2. Scan des objets présents dans la pièce (Vérité RAM)
#         world_objs = env.get_world_objects()
#         detected_objects = []
#         for obj in world_objs:
#             # On vérifie si l'objet est physiquement dans la pièce
#             if obj.parent == room_id:
#                 name = str(obj.name).replace('\x00', '').strip()
#                 # On ignore le joueur et les objets sans nom
#                 if name and "player" not in name.lower():
#                     # On stocke le nom et l'ID pour le "Radar" de l'agent
#                     detected_objects.append({"name": name, "id": obj.num})
                
#         # 3. Inventaire technique (Objets portés)
#         inventory = []
#         for obj in env.get_inventory():
#             inv_name = str(obj.name).replace('\x00', '').strip()
#             inventory.append({"name": inv_name, "id": obj.num})
                
#         # 4. Hash d'état pour détecter tout changement atomique
#         world_hash = env.get_world_state_hash()
                
#         return {
#             "status": "success",
#             "location": {
#                 "id": room_id,
#                 "name": room_name
#             },
#             "detected_objects": detected_objects,
#             "inventory": inventory,
#             "world_hash": world_hash
#         }
        
#     except Exception as e:
#         return {
#             "status": "error",
#             "message": str(e),
#             "location": {"id": -1, "name": "Error"}
#         }

@mcp.tool()
def get_current_room_name() -> str:
    """Récupère le nom officiel de la pièce dans la mémoire du jeu."""
    try:
        game = get_game()
        env = game.env.env # Accès Jericho
        
        # 1. On récupère l'ID de la pièce actuelle
        room_id = env.get_player_location()
        
        # 2. On récupère l'objet correspondant
        room_obj = env.get_object(room_id)
        
        # 3. On renvoie son nom (en forçant le string pour éviter les bytes)
        return str(room_obj.name).strip()
    except Exception as e:
        return f"Error: {e}"

@mcp.tool()
def test_dictionary_access() -> str:
    """[TEST] Récupère un échantillon du dictionnaire interne du jeu."""
    try:
        game = get_game()
        # Récupère tous les mots connus du parseur
        words = game.env.env.get_dictionary()
        # On en prend 20 pour ne pas surcharger le log
        sample = [w.word for w in words[:20]]
        return f"DICTIONARY SAMPLE ({len(words)} words total): " + ", ".join(sample)
    except Exception as e:
        return f"Dictionary Test Failed: {e}"

@mcp.tool()
def test_grammar_access() -> str:
    """[TEST] Récupère les modèles d'actions (Grammaire)."""
    try:
        game = get_game()
        # Les bindings contiennent la grammaire des verbes
        bindings = game.env.env.bindings
        grammar = bindings.get('grammar', 'No grammar found')
        # On coupe si c'est trop long pour l'affichage
        return f"GRAMMAR SNIPPET: {grammar[:300]}..."
    except Exception as e:
        return f"Grammar Test Failed: {e}"

@mcp.tool()
def get_object_tree_simple() -> str:
    """[TEST] Liste brute des 5 premiers objets pour voir leur structure."""
    try:
        game = get_game()
        objs = game.env.env.get_world_objects()
        res = []
        for o in objs[:5]:
            res.append(f"Obj{o.num}: {o.name} (Parent:{o.parent})")
        return "OBJECT TREE SAMPLE: " + " | ".join(res)
    except Exception as e:
        return f"Object Tree Failed: {e}"
# =============================================================================
# Main
# =============================================================================

if __name__ == "__main__":
    mcp.run()