Portfolio

FROM ubuntu:latest AS builder
RUN apt update && apt install openjdk-21-jre unzip -y
RUN mkdir -p /app/world
WORKDIR /app/world
ADD ./ServerFiles-3.0.zip /app
RUN unzip /app/ServerFiles-3.0.zip -d /app/world/ && chmod +x /app/world/startserver.sh

RUN echo "eula=true" > /app/world/eula.txt
RUN /bin/bash -c ' \
    ATM10_INSTALL_ONLY=true /app/world/startserver.sh 2>&1 | tee /dev/stderr; \
    echo "Initialization complete. Exiting build stage."; \
    exit 0'

# stage 2 using files as base image
FROM ubuntu:latest AS final
COPY --from=builder /app/world /app/world
COPY server.properties /app/world/server.properties
RUN apt -y update && apt -y install openjdk-21-jre
EXPOSE 25565
EXPOSE 25575
CMD ["/bin/bash", "-c", "/app/world/startserver.sh"]

services:
  mc:
    build:
      context: .
      dockerfile: Dockerfile
      target: final
    volumes:
      - minecraft-world:/app/world
    ports:
      - "25565:25565"
      - "25575:25575"
volumes:
  minecraft-world:
    driver: local

from mcrcon_shell.mcrcon import *  # Corrected import path
import socket

class MinecraftRconClient:
    def __init__(self, host, port, password):
        self.host = host
        self.port = int(port)
        self.password = password
        self.rcon = self.connect()

    def getserver_info(self):
        return {
            "host": self.host,
            "port": self.port,
        }
    
    def connect(self) -> bool:
        """ Connects to the Minecraft server using RCON.
         Returns True if the connection is successful, False otherwise.
        """
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect((self.host, self.port))
        result = login(sock, self.password)
        if not result:
            print("Incorrect rcon password")
            return False
        self.rcon = sock
        return True

    def run(self, cmd) -> str:
        """ returns the output of the command """
        if not self.rcon:
            print("Not connected to the server")
            return
        response = command(self.rcon, cmd)
        print(response)
        return response
    
    def disconnect(self):
        self.rcon.close()
        self.rcon = None
        print("Disconnected from the server")
        

#!/usr/bin/env python3
"""
Minecraft Java Server MCP Server using mclib

This MCP server provides tools for common Minecraft Java server commands
using RCON protocol through the mclib library.
"""

import os
import json
import logging
from typing import Any, Dict, List, Optional, Union
from dataclasses import dataclass
from enum import StrEnum

from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
from mcrcon_shell.mclib import MinecraftRconClient

# Load environment variables
load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Initialize FastMCP
mcp = FastMCP("Minecraft Java Server")

# Global RCON client instance
rcon_client: Optional['RConClient'] = None

class MinecraftCommandBuilder:
    """Builder for Minecraft commands"""

    @staticmethod
    def teleport(player: str, x: float, y: float, z: float) -> str:
        """Build teleport command"""
        return f"tp {player} {x} {y} {z}"

    @staticmethod
    def ban_player(player: str, reason: Optional[str] = None) -> str:
        """Build ban command"""
        if reason:
            return f"ban {player} {reason}"
        return f"ban {player}"

    @staticmethod
    def unban_player(player: str) -> str:
        """Build unban command"""
        return f"pardon {player}"

    @staticmethod
    def kick_player(player: str, reason: Optional[str] = None) -> str:
        """Build kick command"""
        if reason:
            return f"kick {player} {reason}"
        return f"kick {player}"

    @staticmethod
    def deop_player(player: str) -> str:
        """Build deop command"""
        return f"deop {player}"

    @staticmethod
    def op_player(player: str) -> str:
        """Build op command"""
        return f"op {player}"

    @staticmethod
    def locate_player(player: str) -> tuple[str, str]:
        """Build locate command"""
        return (f"data get entity {player} Pos", 
                f"data get entity {player} Dimension",)
    
    @staticmethod
    def give_item(player: str, item: str, count: int = 1) -> str:
        """Build give command"""
        return f"give {player} {item} {count}"
    
    @staticmethod
    def summon_entity(entity: str, x: float, y: float, z: float) -> str:
        """Build summon command"""
        return f"summon {entity} {x} {y} {z}"
    
    @staticmethod
    def set_weather(weather: str, duration: int, unit: str) -> str:
        """Build weather command"""
        if unit not in ["d", "s", "t"]:
            unit = "t"  # Default to ticks
        return f"weather {weather} {duration}{unit}"
    
    @staticmethod
    def set_time(time: str) -> str:
        """Build time command"""
        match time:
            case "daytime" | "day":
                return "time set day"
            case "nighttime" | "night":
                return "time set night"
            case "noon":
                return "time set noon"
            case "midnight":
                return "time set midnight"
            
    @staticmethod
    def add_time(increment: int, unit: str) -> str:
        """Build add time command"""
        if unit not in ["d", "s", "t"]:
            unit = "t"
        return f"time add {increment}{unit}"
        
    @staticmethod
    def set_gamemode(gamemode: str, player: Optional[str] = None) -> str:
        """Build gamemode command"""
        if player:
            return f"gamemode {gamemode} {player}"
        return f"gamemode {gamemode}"
    
    @staticmethod
    def chat_message(message: str) -> str:
        """Build chat command"""
        message = "[API] " + message
        return f"say {message}"
    
    HELP = "help"
    LIST = "list"
    SAVE = "save-all"
  
class MinecraftItem(StrEnum):
    """Enum for Minecraft items"""
    STONE = "minecraft:stone"
    GRASS_BLOCK = "minecraft:grass_block"
    DIRT = "minecraft:dirt"
    COBBLESTONE = "minecraft:cobblestone"
    OAK_PLANKS = "minecraft:oak_planks"
    OAK_LOG = "minecraft:oak_log"
    IRON_INGOT = "minecraft:iron_ingot"
    GOLD_INGOT = "minecraft:gold_ingot"
    DIAMOND = "minecraft:diamond"
    EMERALD = "minecraft:emerald"
    WOODEN_SWORD = "minecraft:wooden_sword"
    WOODEN_AXE = "minecraft:wooden_axe"
    WOODEN_PICKAXE = "minecraft:wooden_pickaxe"
    WOODEN_SHOVEL = "minecraft:wooden_shovel"
    IRON_SWORD = "minecraft:iron_sword"
    IRON_AXE = "minecraft:iron_axe"
    IRON_PICKAXE = "minecraft:iron_pickaxe"
    IRON_SHOVEL = "minecraft:iron_shovel"
    GOLD_SWORD = "minecraft:gold_sword"
    GOLD_AXE = "minecraft:gold_axe"
    GOLD_PICKAXE = "minecraft:gold_pickaxe"
    GOLD_SHOVEL = "minecraft:gold_shovel"
    DIAMOND_SWORD = "minecraft:diamond_sword"
    DIAMOND_AXE = "minecraft:diamond_axe"
    DIAMOND_PICKAXE = "minecraft:diamond_pickaxe"
    DIAMOND_SHOVEL = "minecraft:diamond_shovel"

class MinecraftEntity(StrEnum):
    """Enum for Minecraft entities"""
    PLAYER = "minecraft:player"
    CREEPER = "minecraft:creeper"
    SKELETON = "minecraft:skeleton"
    SPIDER = "minecraft:spider"
    ZOMBIE = "minecraft:zombie"
    ENDERMAN = "minecraft:enderman"
    COW = "minecraft:cow"
    SHEEP = "minecraft:sheep"
    PIG = "minecraft:pig"
    CHICKEN = "minecraft:chicken"
    VILLAGER = "minecraft:villager"

@dataclass
class RConClient:
    """RConClient for Minecraft Java Server
    This class handles the connection to the Minecraft server using RCON protocol.
    It allows sending commands and receiving responses from the server.
    It manages the socket connection and provides methods to interact with the server.
    """

    host: str
    port: int
    password: str
    name: str = "default"

    def __post_init__(self):
        """Initialize connection with the provided parameters"""
        self.host = os.getenv('MC_HOST', "localhost")
        self.port = int(os.getenv('MC_PORT', "25575"))
        self.password = os.getenv('MC_PASSWORD', "your_password")
        self.connected: bool = False
        self.client = MinecraftRconClient(self.host, self.port, self.password)

    def __repr__(self):
        return f"RConClient(name={self.name}, host={self.host}, port={self.port})"
    
    def _connect(self):
        """Establish a connection to the Minecraft server using RCON"""
        try:
            if not self.connected:
                self.connected = self.client.connect()
                if self.connected:
                    logger.info(f"Connected to Minecraft server: {self.name} at {self.host}:{self.port}")
                else:
                    logger.error(f"Failed to connect to Minecraft server: {self.name}")
                    raise ConnectionError(f"Failed to connect to Minecraft server: {self.name}")
        except Exception as e:
            logger.error(f"Failed to connect to Minecraft server: {self.name}. Error: {e}")
            self.connected = False
            raise ConnectionError(f"Failed to connect to Minecraft server: {self.name}. Error: {e}")

    def _send_command(self, command: str) -> Union[str, None]:
        """Send a command to the Minecraft server and return the response"""
        if not self.connected:
            logger.error("Not connected to the Minecraft server.")
            return "Error: Not connected to the Minecraft server."

        try:
            logger.info(f"Sending command to Minecraft server: {self.name} - {command}")
            response = self.client.run(command)
            logger.info(f"Command sent: {command}, Response type: {type(response)}, Response: {response}")
            
            # Handle different response types
            if response is None:
                return "No response from server"
            elif isinstance(response, str):
                return response if response.strip() else "Empty response from server"
            else:
                return str(response)
                
        except Exception as e:
            logger.error(f"Failed to send command '{command}': {e}")
            return f"Command failed: {str(e)}"
    
    def _close_connection(self):
        """Close the connection to the Minecraft server"""
        if self.connected:
            try:
                self.client.disconnect()
                logger.info(f"Disconnected from Minecraft server: {self.name}")
            except Exception as e:
                logger.error(f"Failed to disconnect from Minecraft server: {self.name}. Error: {e}")
            finally:
                self.connected = False
        else:
            logger.warning("No active connection to close.")

    def send_server_command(self, command: str, batch_count: int = 1) -> list[Optional[str]]:
        """Send a command to the Minecraft server and return the response"""
        responses = []
        try:
            self._connect()
            for i in range(batch_count):
                logger.info(f"Executing command batch {i + 1}/{batch_count}: {command}")
                response = self._send_command(command)
                responses.append(response)
        except Exception as e:
            logger.error(f"Error in send_server_command: {e}")
            responses.append(f"Error: {str(e)}")
        finally:
            self._close_connection()
        return responses
    

class ItemValidator:
    """Validator for Minecraft items"""
    
    @staticmethod
    def validate_item(item: str) -> tuple[bool, str]:
        """Validate if an item exists in Minecraft
        
        Returns:
            tuple[bool, str]: (is_valid, validated_item_or_error_message)
        """
        # Check if it's already a valid enum value
        try:
            minecraft_item = MinecraftItem(item)
            return True, minecraft_item.value
        except ValueError:
            pass
        
        # Check if it's a valid item without minecraft: prefix
        if not item.startswith("minecraft:"):
            try:
                minecraft_item = MinecraftItem(f"minecraft:{item}")
                return True, minecraft_item.value
            except ValueError:
                pass
        
        # Check case-insensitive match
        item_upper = item.upper()
        for minecraft_item in MinecraftItem:
            if minecraft_item.name == item_upper:
                return True, minecraft_item.value
            if minecraft_item.value.upper() == item_upper:
                return True, minecraft_item.value
        
        return False, f"Invalid item: {item}"
    
    @staticmethod
    def get_available_items() -> List[str]:
        """Get list of all available items"""
        return [item.value for item in MinecraftItem]
    
    @staticmethod
    def search_items(query: str) -> List[str]:
        """Search for items containing the query string"""
        query_lower = query.lower()
        matches = []
        for item in MinecraftItem:
            if query_lower in item.name.lower() or query_lower in item.value.lower():
                matches.append(item.value)
        return matches

def get_client() -> RConClient:
    """Get or create the RCON client"""
    global rcon_client
    if rcon_client is None:
        rcon_client = RConClient(
            host=os.getenv('MC_HOST', "localhost"),
            port=int(os.getenv('MC_PORT', "25575")),
            password=os.getenv('MC_PASSWORD', ""),
            name="default"
        )
    return rcon_client

@mcp.tool()
def give_item_to_player(player: str, item: str, count: int = 1) -> str:
    """Give an item to a player with comprehensive validation"""
    if not player.strip():
        return "Player name cannot be empty"
    
    is_valid, result = ItemValidator.validate_item(item)
    if not is_valid:
        return result  # Return error message
    
    command = MinecraftCommandBuilder.give_item(player, result, count)
    response = get_client().send_server_command(command)[0]
    return response or f"Failed to give {item} to {player}"

@mcp.tool()
def search_minecraft_items(query: str) -> str:
    """Search for Minecraft items by name"""
    if not query.strip():
        return "Search query cannot be empty"
    
    matches = ItemValidator.search_items(query)
    if not matches:
        return f"No items found matching '{query}'"
    
    return f"Items matching '{query}':\n" + "\n".join(matches)

@mcp.tool()
def list_all_items() -> str:
    """List all available Minecraft items"""
    items = ItemValidator.get_available_items()
    return f"Available items ({len(items)} total):\n" + "\n".join(items)

@mcp.tool()
def validate_minecraft_item(item: str) -> str:
    """Validate if an item exists in Minecraft"""
    is_valid, result = ItemValidator.validate_item(item)
    if is_valid:
        return f"✅ Valid item: {result}"
    else:
        # Suggest similar items
        suggestions = ItemValidator.search_items(item.split(":")[-1])
        suggestion_text = f"\nSuggestions: {suggestions[:5]}" if suggestions else ""
        return f"❌ {result}{suggestion_text}"
    
@mcp.tool()
def teleport_player(player: str, x: float, y: float, z: float) -> str:
    """Teleport a player to specified coordinates"""
    if not player.strip():
        return "Player name cannot be empty"
    
    command = MinecraftCommandBuilder.teleport(player, x, y, z)
    response = get_client().send_server_command(command)[0]
    return response or f"Failed to teleport {player} to {x}, {y}, {z}"

@mcp.tool()
def summon_entity(entity: str, x: float, y: float, z: float, count: int = 1) -> str:
    """Summon an entity at specified coordinates"""
    responses = []
    if not entity.strip():
        return "Entity name cannot be empty"
    else:
        try:
            MinecraftEntity(entity)  # Validate entity
        except ValueError:
            entity = "minecraft:" + entity if not entity.startswith("minecraft:") else entity
            try:
                MinecraftEntity(entity)  # Validate again with prefix
            except ValueError:
                return f"Invalid entity: {entity}. Use a valid Minecraft entity name."
            return f"Invalid entity: {entity}. Use a valid Minecraft entity name."
        
    command = MinecraftCommandBuilder.summon_entity(entity, x, y, z)
    for i in range(count):
        response = get_client().send_server_command(command)[0]
        if response:
            responses.append(f"Summoned {entity} at {x}, {y}, {z} (Batch {i + 1})")
        else:
            responses.append(f"Failed to summon {entity} at {x}, {y}, {z} (Batch {i + 1})")
    return "\n".join(responses)

@mcp.tool()
def set_weather(weather: str, duration: int, unit: str) -> str:
    """Set the weather in the Minecraft world"""
    if weather not in ["clear", "rain", "thunder"]:
        return "Invalid weather type. Use 'clear', 'rain', or 'thunder'."
    
    command = MinecraftCommandBuilder.set_weather(weather, duration, unit)
    response = get_client().send_server_command(command)[0]
    return response or f"Failed to set weather to {weather} for {duration}{unit}"

@mcp.tool()
def add_time(increment: int, unit: str) -> str:
    """Add time to the Minecraft world"""
    if unit not in ["d", "s", "t"]:
        return "Invalid time unit. Use 'd' for days, 's' for seconds, or 't' for ticks."
    
    command = MinecraftCommandBuilder.add_time(increment, unit)
    response = get_client().send_server_command(command)[0]
    return response or f"Failed to add {increment}{unit} to the game time"

@mcp.tool()
def get_player_location(player: str) -> str:
    """Get the location of a player"""
    if not player.strip():
        return "Player name cannot be empty"
    
    pos_command, dim_command = MinecraftCommandBuilder.locate_player(player)
    pos_response = get_client().send_server_command(pos_command)[0]
    dim_response = get_client().send_server_command(dim_command)[0]
    
    if pos_response and dim_response:
        return f"{player} is at {pos_response} in dimension {dim_response}"
    else:
        return f"Failed to get location for player {player}"

@mcp.tool()
def deop_player(player: str) -> str:
    """De-op a player"""
    if not player.strip():
        return "Player name cannot be empty"
    
    command = MinecraftCommandBuilder.deop_player(player)
    response = get_client().send_server_command(command)[0]
    return response or f"Failed to de-op {player}"

@mcp.tool()
def op_player(player: str) -> str:
    """Op a player"""
    if not player.strip():
        return "Player name cannot be empty"
    
    command = MinecraftCommandBuilder.op_player(player)
    response = get_client().send_server_command(command)[0]
    return response or f"Failed to op {player}"

@mcp.tool()
def chat_message(message: str) -> str:
    """Send a chat message to the Minecraft server"""
    if not message.strip():
        return "Message cannot be empty"
    
    command = MinecraftCommandBuilder.chat_message(message)
    response = get_client().send_server_command(command)[0]
    return response or f"Failed to send message: {message}"

@mcp.tool()
def ban_player(player: str, reason: Optional[str] = None) -> str:
    """Ban a player from the Minecraft server"""
    if not player.strip():
        return "Player name cannot be empty"
    
    command = MinecraftCommandBuilder.ban_player(player, reason)
    response = get_client().send_server_command(command)[0]
    return response or f"Failed to ban {player}"

@mcp.tool()
def unban_player(player: str) -> str:
    """Unban a player from the Minecraft server"""
    if not player.strip():
        return "Player name cannot be empty"
    
    command = MinecraftCommandBuilder.unban_player(player)
    response = get_client().send_server_command(command)[0]
    return response or f"Failed to unban {player}"

@mcp.tool()
def save_world() -> str:
    """Save the Minecraft world"""
    command = MinecraftCommandBuilder.SAVE
    response = get_client().send_server_command(command)[0]
    return response or "Failed to save the world"

@mcp.tool()
def list_players() -> str:
    """List all players currently online on the Minecraft server"""
    command = MinecraftCommandBuilder.LIST
    response = get_client().send_server_command(command)[0]
    if response:
        return f"Online players:\n{response}"
    else:
        return "Failed to retrieve player list"

@mcp.tool()
def test_connection() -> str:
    """Test the connection to the Minecraft server and return debug info"""
    try:
        client = get_client()
        logger.info(f"Testing connection to {client.host}:{client.port}")
        
        # Test with a simple command
        command = "help"
        response = client.send_server_command(command)[0]
        
        if response:
            return f"✅ Connection successful!\nServer response: {response[:100]}..."
        else:
            return f"❌ Connection failed - no response from server\nHost: {client.host}:{client.port}"
            
    except Exception as e:
        return f"❌ Connection error: {str(e)}"

def main():
    """Main function to run the MCP server"""
    mcp.run()

if __name__ == "__main__":
    main()