src/mcp/mcp_tools.py

#!/usr/bin/env python3
"""
MCP Tools for AI Agent Integration
Provides tools for AI agents to interact with the MMORPG game world via MCP
"""

import uuid
import time
import random
from typing import Dict, List
from dataclasses import asdict

from ..core.player import Player


class GradioMCPTools:
    """MCP tools integrated with the game world for AI agent interactions"""
    
    def __init__(self, game_facade):
        """Initialize with game facade for clean architecture access"""
        self.game_facade = game_facade
        self.ai_agents: Dict[str, Player] = {}
    
    def register_ai_agent(self, agent_name: str, mcp_client_id: str = None) -> str:
        """
        Register an AI agent as a player in the MMORPG game world.
        
        Args:
            agent_name: Display name for the AI agent in the game
            mcp_client_id: Optional MCP client identifier (auto-generated if not provided)
            
        Returns:
            The unique agent ID for the registered AI agent
            
        Raises:
            Exception: If the game is full and cannot accommodate more players
        """
        if mcp_client_id is None:
            mcp_client_id = f"ai_{uuid.uuid4().hex[:8]}"
            
        agent_id = f"ai_{uuid.uuid4().hex[:8]}"
          # Use the game facade to register the AI agent
        success = self.game_facade.register_ai_agent(agent_name, agent_id)
        
        if success:
            # Get the player object from the game engine
            player = self.game_facade.game_engine.get_player(agent_id)
            if player:
                player.ai_agent_id = mcp_client_id
                self.ai_agents[mcp_client_id] = player
                return agent_id
        
        raise Exception("Game is full, cannot add AI agent")
    
    def move_ai_agent(self, mcp_client_id: str, direction: str) -> Dict:
        """
        Move an AI agent in the game world in the specified direction.
        
        Args:
            mcp_client_id: The MCP client identifier for the AI agent
            direction: Movement direction (up, down, left, right)
            
        Returns:
            Dictionary containing success status, new position, nearby entities, and world events
        """
        if mcp_client_id not in self.ai_agents:
            return {"error": "AI agent not registered"}
        agent = self.ai_agents[mcp_client_id]
        success = self.game_facade.move_ai_agent(agent.id, direction)
        
        return {
            "success": success,
            "new_position": {"x": agent.x, "y": agent.y},
            "nearby_players": self.get_nearby_entities(agent.id),            "world_events": self.game_facade.get_recent_world_events()
        }
    
    def ai_agent_chat(self, mcp_client_id: str, message: str) -> Dict:
        """
        Send a chat message from an AI agent to the game world.
        
        Args:
            mcp_client_id: The MCP client identifier for the AI agent
            message: The chat message content to send
            
        Returns:
            Dictionary containing success status and confirmation message
        """
        if mcp_client_id not in self.ai_agents:
            return {"error": "AI agent not registered"}
        
        agent = self.ai_agents[mcp_client_id]
        success = self.game_facade.ai_agent_chat(agent.id, message)
        
        return {"success": success, "message": "Chat message sent"}
    
    def get_game_state_for_ai(self, mcp_client_id: str) -> Dict:
        """
        Get comprehensive current game state information for an AI agent.
        
        Args:
            mcp_client_id: The MCP client identifier for the AI agent
            
        Returns:
            Dictionary containing agent status, nearby entities, recent chat, world events, and available NPCs
        """
        if mcp_client_id not in self.ai_agents:
            return {"error": "AI agent not registered"}
        
        agent = self.ai_agents[mcp_client_id]
        
        return {
            "agent_status": asdict(agent),
            "nearby_entities": self.get_nearby_entities(agent.id),
            "recent_chat": self.game_facade.get_recent_chat_messages(),
            "world_events": self.game_facade.get_recent_world_events(),
            "available_npcs": self.game_facade.get_available_npcs()
        }
    
    def get_nearby_entities(self, player_id: str, radius: int = 100) -> List[Dict]:
        """Get entities near a player"""
        player = self.game_facade.game_engine.get_player(player_id)
        if not player:
            return []

        nearby = []

        # Get all players from game engine
        all_players = self.game_facade.game_engine.get_all_players()

        # Check other players
        for other_player in all_players.values():
            if other_player.id == player_id:
                continue

            distance = ((player.x - other_player.x)**2 + (player.y - other_player.y)**2)**0.5
            if distance <= radius:
                nearby.append({
                    "type": "player",
                    "name": other_player.name,
                    "player_type": other_player.type,
                    "position": {"x": other_player.x, "y": other_player.y},
                    "distance": round(distance, 1)
                })

        # Get NPCs from game engine
        all_npcs = self.game_facade.game_engine.get_all_npcs()

        # Check NPCs
        for npc in all_npcs.values():
            distance = ((player.x - npc['x'])**2 + (player.y - npc['y'])**2)**0.5
            if distance <= radius:
                nearby.append({
                    "type": "npc",
                    "name": npc['name'],
                    "npc_id": npc['id'],
                    "position": {"x": npc['x'], "y": npc['y']},
                    "distance": round(distance, 1)
                })

        return nearby

    def send_private_message(self, mcp_client_id: str, target_id: str, message: str) -> Dict:
        """AI agent sends private message to another player or NPC"""
        if mcp_client_id not in self.ai_agents:
            return {"error": "AI agent not registered"}

        agent = self.ai_agents[mcp_client_id]
        success = self.game_facade.send_private_message(
            sender_id=agent.id,
            sender_name=agent.name,
            recipient_id=target_id,
            message=message
        )

        return {"success": success, "message": "Private message sent" if success else "Failed to send message"}

    def get_private_messages(self, mcp_client_id: str, target_id: str) -> Dict:
        """Get private messages between AI agent and target"""
        if mcp_client_id not in self.ai_agents:
            return {"error": "AI agent not registered"}

        agent = self.ai_agents[mcp_client_id]
        messages = self.game_facade.get_private_messages(agent.id, target_id)

        return {
            "success": True,
            "messages": messages,
            "count": len(messages)
        }

    def interact_with_npc(self, mcp_client_id: str, npc_id: str, command: str) -> Dict:
        """AI agent interacts with an NPC"""
        if mcp_client_id not in self.ai_agents:
            return {"error": "AI agent not registered"}

        agent = self.ai_agents[mcp_client_id]

        # Send message to NPC via private message system
        success = self.game_facade.send_private_message(
            sender_id=agent.id,
            sender_name=agent.name,
            recipient_id=npc_id,
            message=command
        )

        if success:
            # Get recent messages to see NPC response
            messages = self.game_facade.get_private_messages(agent.id, npc_id)
            recent_response = messages[-1] if messages else None

            return {
                "success": True,
                "message": "Command sent to NPC",
                "npc_response": recent_response.get('message', '') if recent_response else '',
                "interaction_log": messages[-3:] if len(messages) >= 3 else messages
            }
        else:
            return {"error": "Failed to send command to NPC"}

    def get_world_events(self, mcp_client_id: str) -> Dict:
        """Get recent world events for AI agent"""
        if mcp_client_id not in self.ai_agents:
            return {"error": "AI agent not registered"}

        try:
            events = self.game_facade.get_world_events()
            return {
                "success": True,
                "events": events[-10:],  # Last 10 events
                "count": len(events)
            }
        except Exception as e:
            return {"error": f"Failed to get world events: {str(e)}"}

    def get_agent_info(self, mcp_client_id: str) -> Dict:
        """Get information about the AI agent"""
        if mcp_client_id not in self.ai_agents:
            return {"error": "AI agent not registered"}

        agent = self.ai_agents[mcp_client_id]
        return {
            "success": True,
            "agent_id": agent.id,
            "name": agent.name,
            "type": agent.type,
            "position": {"x": agent.x, "y": agent.y},
            "level": agent.level,
            "hp": f"{agent.hp}/{agent.max_hp}",
            "gold": agent.gold,
            "experience": agent.experience,
            "last_active": agent.last_active,
            "ai_agent_id": getattr(agent, 'ai_agent_id', mcp_client_id)
        }