Spaces:
Sleeping
Sleeping
| """ | |
| 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 = "" | |
| 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.state.observation | |
| return self.state.observation | |
| def step(self, action: str) -> str: | |
| """Execute an action and return the result.""" | |
| if self.env is None: | |
| self.initialize() | |
| # TODO: Update your state tracking here | |
| # self.history.append((action, self.state.observation)) | |
| # Update location tracking, etc. | |
| prev_location = self.current_location | |
| self.state = self.env.step(action) | |
| observation = self.state.observation | |
| self.history.append((action, observation)) | |
| if prev_location not in self.explored_locations: | |
| self.explored_locations[prev_location] = set() | |
| self.explored_locations[prev_location].add(observation) | |
| self.current_location = observation | |
| return self.state.observation | |
| 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 | |
| # ============================================================================= | |
| 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 <item>, drop <item>, open <thing>, examine <thing> | |
| - Other: look, inventory, read <thing>, 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 | |
| result += f"\n[Score: {game.get_score()} | Moves: {game.get_moves()}]" | |
| return result | |
| # TODO: Implement additional tools to help your agent | |
| def memory() -> str: | |
| game = get_game() | |
| lines = [] | |
| lines.append(f"Game: {game.game_name}") | |
| lines.append(f"Score: {game.get_score()}") | |
| lines.append(f"Moves: {game.get_moves()}") | |
| lines.append(f"Current location: {game.current_location}") | |
| if game.history: | |
| lines.append("Recent history:") | |
| for action, obs in game.history[-5:]: | |
| lines.append(f"- {action} -> {obs}") | |
| else: | |
| lines.append("No actions taken yet.") | |
| return "\n".join(lines) | |
| def inventory() -> str: | |
| game = get_game() | |
| result = game.step("inventory") | |
| return result | |
| def get_map() -> str: | |
| game = get_game() | |
| if not game.explored_locations: | |
| return "No locations explored yet." | |
| lines = [] | |
| for location, neighbors in game.explored_locations.items(): | |
| lines.append(f"{location}:") | |
| for n in neighbors: | |
| lines.append(f" -> {n}") | |
| return "\n".join(lines) | |
| 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() | |