""" Student MCP Server for Text Adventure Games 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. """ 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 # ============================================================================= # Create the MCP Server # ============================================================================= mcp = FastMCP("Student 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 """ 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 = self._extract_location(self.state.observation) if self.state else "Unknown" self.current_objective = "Explore the game world and maximize score." self.interactives: dict = {} # Interactive objects and their locations 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 initialize(self, game: str = "zork1"): """Initialize or reset the game.""" self.game_name = game self.env = TextAdventureEnv(game) self.state = self.env.reset() # TODO: Reset your state tracking here self.history = [] self.explored_locations = {} self.current_location = self._extract_location(self.state.observation) self.current_objective = "Explore the game world and maximize score." self.interactives: dict = {} return self.state.observation def step(self, action: str) -> str: """Execute an action and return the result.""" if self.env is None: self.initialize() self.state = self.env.step(action) result = self.state.observation # TODO: Update your state tracking here self.history.append((action, result)) if len(self.history) > 20: self.history = self.history[-20:] 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 def get_memory(self) -> str: """Get a summary of current game state.""" recent = self.history if self.history else [] recent_str = "\n".join([f" > {a} -> {r}..." 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} All actions and observations: {recent_str} Current Observation: {self.state.observation} Current Objective: {self.current_objective}""" 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)}" def get_interactives(self) -> str: """Get interactive objects and their locations.""" if not self.interactives: return "Interactives: No interactive objects identified yet." lines = ["Interactive objects and Locations:"] for obj, loc in sorted(self.interactives.items()): lines.append(f"\n* {obj} (at {loc})") return "\n".join(lines) def add_interactive(self, obj: str, location: str = None): """Add an interactive object, automatically adds its location.""" if location is None: location = self.current_location self.interactives[obj] = location def get_current_objective(self) -> str: """Get the current objective.""" return self.current_objective def set_current_objective(self, objective: str): """Set a new current objective.""" self.current_objective = objective def get_score(self) -> int: """Get current score.""" return self.state.score if self.state else 0 def get_moves(self) -> int: """Get number of moves taken.""" return self.state.moves if self.state else 0 # Global game manager _game = GameManager() 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 # ============================================================================= # MCP Tools - IMPLEMENT THESE # ============================================================================= @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. Args: 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 , drop , open , examine - Other: look, inventory, read , turn on lamp """ game = get_game() # TODO: You might want to add action validation here # TODO: You might want to include score changes in the response result = game.step(action) # Optional: Append 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 # TODO: Implement additional tools to help your agent @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() @mcp.tool() def get_interactives() -> str: """ Get interactive objects and their locations. """ return get_game().get_interactives() @mcp.tool() def add_interactive(object: str, location: str = None) -> str: """ Add an interactive object and its location. """ get_game().add_interactive(object, location) return f"Added interactive: {object} at location: {location}" @mcp.tool() def get_current_objective() -> str: """ Get the current objective. """ return get_game().get_current_objective() @mcp.tool() def set_current_objective(objective: str) -> str: """ Set a new current objective. """ get_game().set_current_objective(objective) return f"Objective updated to: {objective}" @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" # ============================================================================= # Run the server # ============================================================================= if __name__ == "__main__": # This runs the server with stdio transport (for MCP clients) mcp.run()