http_server.py

#!/usr/bin/env python3
"""
HTTP wrapper for MCP MUD Game Server for testing
"""

import asyncio
import json
import logging
from typing import Any, Dict
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
import uvicorn

# Import our game systems
from game_engine import GameEngine
from character_system import Character
from dungeon_manager import DungeonManager
from dice_system import DiceSystem
from database_manager import DatabaseManager

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

# Initialize FastAPI app
app = FastAPI(title="MCP MUD Game Server", version="1.0.0")

# Initialize game components
game_engine = GameEngine()  # GameEngine initializes its own components

@app.on_event("startup")
async def startup_event():
    """Initialize database on startup."""
    try:
        game_engine.db_manager.initialize_tables()
        logger.info("Database tables initialized successfully")
    except Exception as e:
        logger.error(f"Failed to initialize database: {e}")

@app.get("/health")
async def health_check():
    """Health check endpoint."""
    return {"status": "healthy", "service": "mcp-mud-game"}

@app.get("/mcp/tools/list")
async def list_tools():
    """List available MCP tools."""
    tools = [
        {
            "name": "create_character",
            "description": "Create a new player character",
            "parameters": ["player_id", "character_name", "character_class"]
        },
        {
            "name": "get_character_status", 
            "description": "Get current character status and stats",
            "parameters": ["player_id"]
        },
        {
            "name": "move_character",
            "description": "Move character in a direction", 
            "parameters": ["player_id", "direction"]
        },
        {
            "name": "examine_object",
            "description": "Examine room, object, or creature",
            "parameters": ["player_id", "target"]
        },
        {
            "name": "attack_target",
            "description": "Attack a target with equipped weapon",
            "parameters": ["player_id", "target", "weapon?"]
        },
        {
            "name": "cast_spell",
            "description": "Cast a spell",
            "parameters": ["player_id", "spell_name", "target?"]
        },
        {
            "name": "use_item",
            "description": "Use an item from inventory",
            "parameters": ["player_id", "item_name", "target?"]
        },
        {
            "name": "attempt_skill_check",
            "description": "Attempt a skill check",
            "parameters": ["player_id", "skill_type", "target", "difficulty?"]
        },
        {
            "name": "rest_character",
            "description": "Rest to recover health and mana",
            "parameters": ["player_id"]
        },
        {
            "name": "generate_room",
            "description": "Generate a new room (DM only)",
            "parameters": ["theme", "difficulty", "connections"]
        },
        {
            "name": "add_encounter",
            "description": "Add encounter to room (DM only)",
            "parameters": ["room_id", "encounter_type"]
        },
        {
            "name": "get_dungeon_overview",
            "description": "Get overview of current dungeon state (DM only)",
            "parameters": []
        },
        {
            "name": "cleanup_broken_connections", 
            "description": "Clean up broken room connections (Admin only)",
            "parameters": []
        }
    ]
    return {"tools": tools}

@app.post("/mcp/tools/call")
async def call_tool(request: Dict[str, Any]):
    """Call an MCP tool."""
    try:
        tool_name = request.get("tool")
        arguments = request.get("arguments", {})
        
        logger.info(f"Calling tool: {tool_name} with arguments: {arguments}")
        
        if tool_name == "create_character":
            result = game_engine.create_character(
                arguments["player_id"],
                arguments["character_name"],
                arguments["character_class"]
            )
            
        elif tool_name == "get_character_status":
            result = game_engine.get_character_status(arguments["player_id"])
            
        elif tool_name == "move_character":
            result = game_engine.move_character(
                arguments["player_id"],
                arguments["direction"]
            )
            
        elif tool_name == "examine_object":
            result = game_engine.examine_object(
                arguments["player_id"],
                arguments["target"]
            )
            
        elif tool_name == "attack_target":
            result = game_engine.attack_target(
                arguments["player_id"],
                arguments["target"],
                arguments.get("weapon")
            )
            
        elif tool_name == "cast_spell":
            result = game_engine.cast_spell(
                arguments["player_id"],
                arguments["spell_name"],
                arguments.get("target")
            )
            
        elif tool_name == "use_item":
            result = game_engine.use_item(
                arguments["player_id"],
                arguments["item_name"],
                arguments.get("target")
            )
            
        elif tool_name == "attempt_skill_check":
            result = game_engine.attempt_skill_check(
                arguments["player_id"],
                arguments["skill_type"],
                arguments["target"],
                arguments.get("difficulty", "medium")
            )
            
        elif tool_name == "rest_character":
            result = game_engine.rest_character(arguments["player_id"])
            
        elif tool_name == "generate_room":
            result = game_engine.dungeon_manager.generate_room(
                arguments["theme"],
                arguments["difficulty"],
                arguments["connections"]
            )
            
        elif tool_name == "add_encounter":
            result = game_engine.dungeon_manager.add_encounter(
                arguments["room_id"],
                arguments["encounter_type"]
            )
            
        elif tool_name == "get_dungeon_overview":
            result = game_engine.dungeon_manager.get_dungeon_overview()
            
        elif tool_name == "cleanup_broken_connections":
            result = game_engine.cleanup_broken_connections()
            
        else:
            raise HTTPException(status_code=400, detail=f"Unknown tool: {tool_name}")
        
        return JSONResponse(content=result)
        
    except Exception as e:
        logger.error(f"Error calling tool {tool_name}: {e}")
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)