Spaces:
Sleeping
Sleeping
| """ | |
| Student MCP Server for Text Adventure Games | |
| """ | |
| import sys | |
| 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 | |
| mcp = FastMCP("Student Text Adventure Server") | |
| class GameManager: | |
| def __init__(self): | |
| self.env: TextAdventureEnv = None | |
| self.state = None | |
| self.game_name: str = "" | |
| self.history: list[tuple[str, str]] = [] | |
| self.explored_locations: dict[str, set[str]] = {} | |
| self.current_location: str = "" | |
| self.failed_actions: set[str] = set() | |
| def initialize(self, game: str = "zork1"): | |
| self.game_name = game | |
| self.env = TextAdventureEnv(game) | |
| self.state = self.env.reset() | |
| self.history = [] | |
| self.explored_locations = {} | |
| self.failed_actions = set() | |
| self.current_location = self._extract_location(self.state.observation) | |
| return self.state.observation | |
| def _extract_location(self, observation: str) -> str: | |
| lines = observation.strip().split('\n') | |
| return lines[0] if lines else "Unknown" | |
| def step(self, action: str) -> str: | |
| if self.env is None: | |
| self.initialize() | |
| prev_score = self.state.score | |
| self.state = self.env.step(action) | |
| result = self.state.observation | |
| # Track failed actions (no score gain, no location change) | |
| new_location = self._extract_location(result) | |
| if self.state.score == prev_score and new_location == self.current_location: | |
| self.failed_actions.add(action.lower().strip()) | |
| # Update map | |
| if action.lower().strip() 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 | |
| self.history.append((action, result)) | |
| if len(self.history) > 50: | |
| self.history = self.history[-50:] | |
| return result | |
| def get_score(self) -> int: | |
| return self.state.score if self.state else 0 | |
| def get_moves(self) -> int: | |
| return self.state.moves if self.state else 0 | |
| _game = GameManager() | |
| def get_game() -> GameManager: | |
| global _game | |
| if _game.env is None: | |
| game = os.environ.get("GAME", "zork1") | |
| _game.initialize(game) | |
| return _game | |
| def play_action(action: str) -> str: | |
| """ | |
| Execute a game command and return the result. | |
| Args: | |
| action: The command to execute (e.g., "north", "take lamp", "open mailbox") | |
| Returns: | |
| The game's response to the action | |
| """ | |
| game = get_game() | |
| result = game.step(action) | |
| score_info = f"\n\n[Score: {game.get_score()} | Moves: {game.get_moves()}]" | |
| if game.state.reward > 0: | |
| score_info = f"\n\n+{game.state.reward} points! (Total: {game.get_score()})" | |
| done_info = "\n\nGAME OVER" if game.state.done else "" | |
| return result + score_info + done_info | |
| def memory() -> str: | |
| """ | |
| Get current game state: location, score, moves, and recent action history. | |
| """ | |
| game = get_game() | |
| recent = game.history[-5:] if game.history else [] | |
| recent_str = "\n".join([f" > {a} -> {r[:60]}..." for a, r in recent]) if recent else " (none yet)" | |
| failed_str = ", ".join(sorted(game.failed_actions)[:20]) if game.failed_actions else "none" | |
| return f"""Current State: | |
| - Location: {game.current_location} | |
| - Score: {game.get_score()} points | |
| - Moves: {game.get_moves()} | |
| - Game: {game.game_name} | |
| Recent Actions: | |
| {recent_str} | |
| Failed Actions (DO NOT REPEAT THESE): | |
| {failed_str} | |
| Current Observation: | |
| {game.state.observation}""" | |
| def get_valid_actions() -> str: | |
| """ | |
| Get a list of valid actions available in the current game state. | |
| Use this to avoid wasting steps on invalid commands. | |
| """ | |
| game = get_game() | |
| try: | |
| valid = game.env.get_valid_actions() | |
| # Filter out already-failed actions | |
| filtered = [a for a in valid if a.lower().strip() not in game.failed_actions] | |
| return "Valid actions: " + ", ".join(filtered[:25]) | |
| except Exception: | |
| return "Could not determine valid actions" | |
| def inventory() -> str: | |
| """Check what items you are currently carrying.""" | |
| game = get_game() | |
| result = game.step("inventory") | |
| return result | |
| def get_map() -> str: | |
| """Get a map of explored locations and their connections.""" | |
| game = get_game() | |
| if not game.explored_locations: | |
| return "No locations explored yet. Try moving around!" | |
| lines = ["Explored Locations and Exits:"] | |
| for loc, exits in sorted(game.explored_locations.items()): | |
| lines.append(f"\n* {loc}") | |
| for exit_info in sorted(exits): | |
| lines.append(f" -> {exit_info}") | |
| lines.append(f"\n[Current] {game.current_location}") | |
| return "\n".join(lines) | |
| if __name__ == "__main__": | |
| mcp.run() |