Update mcp_server.py

mcp_server.py

@@ -1,27 +1,8 @@
 """
-Student MCP Server for Text Adventure Games
+Example: MCP Server for Text Adventures
 
-This is your MCP server submission. Implement the tools that your agent
-will use to play text adventure games.
-
-Required tool:
-    play_action(action: str) -> str
-        Execute a game command and return the result.
-
-Recommended tools:
-    memory() -> str
-        Return current game state, score, and recent history.
-    
-    inventory() -> str  
-        Return the player's current inventory.
-    
-    get_map() -> str
-        Return a map of explored locations.
-
-Test your server with:
-    fastmcp dev submission_template/mcp_server.py
-
-Then open the MCP Inspector in your browser to test the tools interactively.
+A complete MCP server that exposes text adventure games via tools.
+This demonstrates a full-featured server with memory, mapping, and inventory.
 """
 
 import sys
@@ -31,179 +12,205 @@ import os
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 from fastmcp import FastMCP
-from games.zork_env import TextAdventureEnv
-
+from games.zork_env import TextAdventureEnv, list_available_games
 
-# =============================================================================
-# Create the MCP Server
-# =============================================================================
 
-mcp = FastMCP("Student Text Adventure Server")
+# Get game from environment variable (default: zork1)
+INITIAL_GAME = os.environ.get("GAME", "zork1")
 
+# Create the MCP server
+mcp = FastMCP("Text Adventure Server")
 
-# =============================================================================
-# Game State Management
-# =============================================================================
 
-class GameManager:
-    """
-    Manages the text adventure game state.
-    
-    TODO: Extend this class to track:
-    - Action history (for memory tool)
-    - Explored locations (for mapping)
-    - Current score and moves
-    """
+class GameState:
+    """Manages the text adventure game state and exploration data."""
     
-    def __init__(self):
-        self.env: TextAdventureEnv = None
-        self.state = None
-        self.game_name: str = ""
-        # TODO: Add more state tracking
-        # self.history: list[tuple[str, str]] = []
-        # self.explored_locations: dict[str, set[str]] = {}
-        # self.current_location: str = ""
-    
-    def initialize(self, game: str = "zork1"):
-        """Initialize or reset the game."""
+    def __init__(self, game: str = "zork1"):
         self.game_name = game
         self.env = TextAdventureEnv(game)
         self.state = self.env.reset()
-        # TODO: Reset your state tracking here
-        return self.state.observation
+        self.history: list[tuple[str, str]] = []
+        self.explored_locations: dict[str, set[str]] = {}
+        self.current_location: str = self._extract_location(self.state.observation)
     
-    def step(self, action: str) -> str:
-        """Execute an action and return the result."""
-        if self.env is None:
-            self.initialize()
-        
+    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 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:]
         
-        # TODO: Update your state tracking here
-        # self.history.append((action, self.state.observation))
-        # Update location tracking, etc.
+        # 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 self.state.observation
+        return result
+    
+    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_score(self) -> int:
-        """Get current score."""
-        return self.state.score if self.state else 0
+    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_moves(self) -> int:
-        """Get number of moves taken."""
-        return self.state.moves if self.state else 0
+    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 manager
-_game = GameManager()
+# Global game state
+_game_state: GameState | None = None
 
 
-def get_game() -> GameManager:
-    """Get or initialize the game manager."""
-    global _game
-    if _game.env is None:
-        # Get game from environment variable (set by evaluator)
-        game = os.environ.get("GAME", "zork1")
-        _game.initialize(game)
-    return _game
+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 - IMPLEMENT THESE
+# MCP Tools
 # =============================================================================
 
 @mcp.tool()
 def play_action(action: str) -> str:
     """
-    Execute a game command and return the result.
-    
-    This is the main tool for interacting with the game.
+    Execute a game action in the text adventure.
     
     Args:
-        action: The command to execute (e.g., "north", "take lamp", "open mailbox")
-        
+        action: The command to execute (e.g., 'north', 'take lamp', 'open mailbox')
+    
     Returns:
-        The game's response to the action
-        
-    Valid commands include:
-        - Movement: north, south, east, west, up, down, enter, exit
-        - Objects: take <item>, drop <item>, open <thing>, examine <thing>
-        - Other: look, inventory, read <thing>, turn on lamp
+        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}]"
     
-    # TODO: You might want to add action validation here
-    # TODO: You might want to include score changes in the response
+    if game.state.reward > 0:
+        score_info = f"\n\n+{game.state.reward} points! (Total: {game.state.score})"
     
-    result = game.step(action)
+    done_info = ""
+    if game.state.done:
+        done_info = "\n\nGAME OVER"
     
-    # Optional: Append score info
-    # result += f"\n[Score: {game.get_score()} | Moves: {game.get_moves()}]"
+    return result + score_info + done_info
+
+import concurrent.futures
+
+@mcp.tool()
+def available_actions() -> str:
+    """Get the list of valid actions at the current location."""
+    game = get_game()
+    with concurrent.futures.ThreadPoolExecutor() as executor:
+        future = executor.submit(game.env.get_valid_actions)
+        try:
+            actions = future.result(timeout=5)
+            return ", ".join(actions)
+        except concurrent.futures.TimeoutError:
+            return "Timed out. Common actions: look, north, south, east, west, up, down, take, open, examine"
+
+
+@mcp.tool()
+def memory() -> str:
+    """
+    Get a summary of the current game state.
     
-    return result
-
-
-# TODO: Implement additional tools to help your agent
-
-# @mcp.tool()
-# def memory() -> str:
-#     """
-#     Get the current game state summary.
-#     
-#     Returns:
-#         A summary including current location, score, moves, and recent history
-#     """
-#     game = get_game()
-#     # TODO: Return useful state information
-#     pass
-
-
-# @mcp.tool()
-# def inventory() -> str:
-#     """
-#     Check what the player is carrying.
-#     
-#     Returns:
-#         List of items in the player's inventory
-#     """
-#     game = get_game()
-#     result = game.step("inventory")
-#     return result
-
-
-# @mcp.tool()
-# def get_map() -> str:
-#     """
-#     Get a map of explored locations.
-#     
-#     Returns:
-#         A text representation of explored locations and connections
-#     """
-#     game = get_game()
-#     # TODO: Return map of explored locations
-#     pass
-
-
-# @mcp.tool()
-# def get_valid_actions() -> str:
-#     """
-#     Get a list of likely valid actions from the current location.
-#     
-#     Returns:
-#         List of actions that might work here
-#     """
-#     # This is a hint: Jericho provides get_valid_actions()
-#     game = get_game()
-#     if game.env and game.env.env:
-#         valid = game.env.env.get_valid_actions()
-#         return "Valid actions: " + ", ".join(valid[:20])
-#     return "Could not determine valid actions"
+    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()  
+
+
+@mcp.tool()
+def get_location() -> str:
+    """Get the current player location reliably using the game engine."""
+    game = get_game()
+    return game.state.location
 
 # =============================================================================
-# Run the server
+# Main
 # =============================================================================
 
 if __name__ == "__main__":
-    # This runs the server with stdio transport (for MCP clients)
-    mcp.run()
+    mcp.run()