mcp_server.py

"""
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 os
import sys
import logging

# IMPORTANT:
# MCP stdio requires that stdout contains ONLY JSON-RPC messages.
# Any logs/prints to stdout will break the client.
# We force Python logging to stderr and silence noisy loggers.
logging.getLogger().handlers.clear()
logging.basicConfig(stream=sys.stderr, level=logging.ERROR)
for name in ["mcp", "fastmcp"]:
    logging.getLogger(name).setLevel(logging.ERROR)

# 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.history = []
    
    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 = []
        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)
        
        # TODO: Update your state tracking here
        # self.history.append((action, self.state.observation))
        # Update location tracking, etc.
        self.history.append((action, self.state.observation))
        if len(self.history) > 40:
            self.history = self.history[-40:]
        
        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
# =============================================================================

@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 <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()}]"
    result += f"\n[Score: {game.get_score()} | Moves: {game.get_moves()}]"

    if game.state and getattr(game.state, "done", False):
        result += "\nGAME OVER"
    
    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()
    loc = (game.state.observation.split("\n", 1)[0] if game.state else "").strip()
    recent = "\n".join([f"{a} -> {o[:60].replace(chr(10), ' ')}" for a, o in game.history[-6:]]) or "(none)"
    return (
        f"Location: {loc}\n"
        f"Score: {game.get_score()}\n"
        f"Moves: {game.get_moves()}\n"
        f"Recent actions:\n{recent}"
    )


@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 + f"\n[Score: {game.get_score()} | Moves: {game.get_moves()}]"


@mcp.tool()
def valid_actions() -> str:
    # Keep it simple and safe: return empty if you don't have Jericho wired.
    # Returning empty is better than crashing the server.
    return ""


# =============================================================================
# Run the server
# =============================================================================

if __name__ == "__main__":
    # This runs the server with stdio transport (for MCP clients)
    mcp.run()