Spaces:
Runtime error
Runtime error
| #!/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) | |
| } | |