Documentation/simple_game_client.py

#!/usr/bin/env python3
"""

Simple MMORPG Game Client

========================



A basic client implementation for connecting to the MMORPG game server via MCP.

This client can join the game, move around, chat, and interact with NPCs.



Requirements:

    pip install mcp aiohttp asyncio



Usage:

    python simple_game_client.py



Features:

    - Connect to game server via MCP

    - Register as a player

    - Move around the game world  

    - Send chat messages

    - Interact with NPCs

    - Get game state information

    - Interactive console interface

"""

import asyncio
import json
import uuid
import sys
from typing import Dict, List, Optional

try:
    from mcp import ClientSession
    from mcp.client.sse import sse_client
    from contextlib import AsyncExitStack
except ImportError:
    print("❌ Missing dependencies. Install with: pip install mcp aiohttp")
    sys.exit(1)


class SimpleGameClient:
    """Simple client for connecting to the MMORPG game server"""
    
    def __init__(self, server_url="http://127.0.0.1:7868/gradio_api/mcp/sse"):
        self.server_url = server_url
        self.session = None
        self.connected = False
        self.tools = []
        self.exit_stack = None
        self.client_id = f"client_{uuid.uuid4().hex[:8]}"
        self.agent_id = None
        self.player_name = None
        
        print(f"🎮 Game Client initialized with ID: {self.client_id}")
    
    async def connect(self):
        """Connect to the game server"""
        print(f"🔌 Connecting to {self.server_url}...")
        
        try:
            # Clean up any existing connection
            if self.exit_stack:
                await self.exit_stack.aclose()
            
            self.exit_stack = AsyncExitStack()
            
            # Establish SSE connection to MCP server
            transport = await self.exit_stack.enter_async_context(
                sse_client(self.server_url)
            )
            read_stream, write_callable = transport
            
            # Create MCP client session
            self.session = await self.exit_stack.enter_async_context(
                ClientSession(read_stream, write_callable)
            )
            
            # Initialize the session
            await self.session.initialize()
            
            # Get available tools/commands from server
            response = await self.session.list_tools()
            self.tools = response.tools
            
            self.connected = True
            tool_names = [tool.name for tool in self.tools]
            print(f"✅ Connected successfully!")
            print(f"📋 Available commands: {', '.join(tool_names)}")
            
            return True
            
        except Exception as e:
            print(f"❌ Connection failed: {e}")
            self.connected = False
            return False
    
    async def register_player(self, player_name: str):
        """Register as a new player in the game"""
        if not self.connected:
            print("❌ Not connected to server")
            return False
        
        print(f"👤 Registering player: {player_name}")
        
        try:
            # Find the register tool
            register_tool = self._find_tool(['register', 'agent'])
            if not register_tool:
                print("❌ Register tool not available on server")
                return False
            
            # Register the player
            result = await self.session.call_tool(
                register_tool.name,
                {"name": player_name, "client_id": self.client_id}
            )
            
            # Parse the result
            content = self._extract_content(result)
            print(f"📝 Registration response: {content}")
            
            if "registered" in content.lower() or "success" in content.lower():
                self.player_name = player_name
                # Try to extract agent_id from response
                self._parse_agent_id(content)
                print(f"✅ Successfully registered as: {player_name}")
                return True
            else:
                print(f"❌ Registration failed: {content}")
                return False
                
        except Exception as e:
            print(f"❌ Registration error: {e}")
            return False
    
    async def move_player(self, direction: str):
        """Move the player in the specified direction"""
        if not self._check_ready():
            return False
        
        direction = direction.lower()
        valid_directions = ['north', 'south', 'east', 'west', 'up', 'down']
        
        if direction not in valid_directions:
            print(f"❌ Invalid direction. Use: {', '.join(valid_directions)}")
            return False
        
        try:
            # Find the move tool
            move_tool = self._find_tool(['move', 'agent'])
            if not move_tool:
                print("❌ Move tool not available")
                return False
            
            # Execute the move
            result = await self.session.call_tool(
                move_tool.name,
                {"client_id": self.client_id, "direction": direction}
            )
            
            content = self._extract_content(result)
            print(f"🚶 Move {direction}: {content}")
            return True
            
        except Exception as e:
            print(f"❌ Move error: {e}")
            return False
    
    async def send_chat(self, message: str):
        """Send a chat message to other players"""
        if not self._check_ready():
            return False
        
        if not message.strip():
            print("❌ Cannot send empty message")
            return False
        
        try:
            # Find the chat tool
            chat_tool = self._find_tool(['chat', 'send'])
            if not chat_tool:
                print("❌ Chat tool not available")
                return False
            
            # Send the chat message
            result = await self.session.call_tool(
                chat_tool.name,
                {"client_id": self.client_id, "message": message}
            )
            
            content = self._extract_content(result)
            print(f"💬 Chat sent: {content}")
            return True
            
        except Exception as e:
            print(f"❌ Chat error: {e}")
            return False
    
    async def get_game_state(self):
        """Get current game state information"""
        if not self.connected:
            print("❌ Not connected to server")
            return None
        
        try:
            # Find the game state tool
            state_tool = self._find_tool(['state', 'game'])
            if not state_tool:
                print("❌ Game state tool not available")
                return None
            
            # Get game state
            result = await self.session.call_tool(state_tool.name, {})
            content = self._extract_content(result)
            
            try:
                # Try to parse as JSON
                game_state = json.loads(content)
                return game_state
            except json.JSONDecodeError:
                # Return as text if not JSON
                return {"info": content}
                
        except Exception as e:
            print(f"❌ Game state error: {e}")
            return None
    
    async def interact_with_npc(self, npc_id: str, message: str):
        """Interact with an NPC"""
        if not self._check_ready():
            return False
        
        if not npc_id.strip() or not message.strip():
            print("❌ NPC ID and message are required")
            return False
        
        try:
            # Find the interact tool
            interact_tool = self._find_tool(['interact', 'npc'])
            if not interact_tool:
                print("❌ NPC interaction tool not available")
                return False
            
            # Interact with NPC
            result = await self.session.call_tool(
                interact_tool.name,
                {"client_id": self.client_id, "npc_id": npc_id, "message": message}
            )
            
            content = self._extract_content(result)
            print(f"🤖 {npc_id}: {content}")
            return True
            
        except Exception as e:
            print(f"❌ NPC interaction error: {e}")
            return False
    
    async def list_tools(self):
        """List all available tools on the server"""
        if not self.connected:
            print("❌ Not connected to server")
            return
        
        print("🛠️ Available Tools:")
        for i, tool in enumerate(self.tools, 1):
            print(f"  {i}. {tool.name}")
            if hasattr(tool, 'description') and tool.description:
                print(f"     Description: {tool.description}")
    
    def _find_tool(self, keywords: List[str]):
        """Find a tool by keywords"""
        for keyword in keywords:
            tool = next((t for t in self.tools if keyword in t.name.lower()), None)
            if tool:
                return tool
        return None
    
    def _extract_content(self, result):
        """Extract content from MCP result"""
        try:
            if hasattr(result, 'content') and result.content:
                if isinstance(result.content, list):
                    content_parts = []
                    for item in result.content:
                        if hasattr(item, 'text'):
                            content_parts.append(item.text)
                        else:
                            content_parts.append(str(item))
                    return ''.join(content_parts)
                elif hasattr(result.content, 'text'):
                    return result.content.text
                else:
                    return str(result.content)
            return str(result)
        except Exception as e:
            return f"Error extracting content: {e}"
    
    def _parse_agent_id(self, content: str):
        """Try to extract agent ID from response"""
        lines = content.split('\n')
        for line in lines:
            if any(keyword in line.lower() for keyword in ['agent_id', 'id:', 'player id']):
                parts = line.split(':')
                if len(parts) > 1:
                    self.agent_id = parts[1].strip()
                    print(f"🆔 Agent ID: {self.agent_id}")
                    break
    
    def _check_ready(self):
        """Check if client is ready for game operations"""
        if not self.connected:
            print("❌ Not connected to server")
            return False
        if not self.player_name:
            print("❌ Not registered as a player")
            return False
        return True
    
    async def disconnect(self):
        """Disconnect from the server"""
        if self.exit_stack:
            await self.exit_stack.aclose()
        self.connected = False
        self.player_name = None
        self.agent_id = None
        print("🔌 Disconnected from server")


class InteractiveGameClient:
    """Interactive console interface for the game client"""
    
    def __init__(self):
        self.client = SimpleGameClient()
        self.running = True
    
    async def start(self):
        """Start the interactive client"""
        print("🎮 MMORPG Simple Game Client")
        print("=" * 40)
        self.show_help()
        
        while self.running:
            try:
                command = input("\n> ").strip()
                if command:
                    await self.process_command(command)
            except KeyboardInterrupt:
                print("\n\n👋 Goodbye!")
                break
            except EOFError:
                break
        
        await self.client.disconnect()
    
    async def process_command(self, command: str):
        """Process a user command"""
        parts = command.split()
        if not parts:
            return
        
        cmd = parts[0].lower()
        args = parts[1:]
        
        if cmd == "help":
            self.show_help()
            
        elif cmd == "connect":
            await self.client.connect()
            
        elif cmd == "register":
            if args:
                name = " ".join(args)
                await self.client.register_player(name)
            else:
                print("❌ Usage: register <player_name>")
                
        elif cmd == "move":
            if args:
                direction = args[0]
                await self.client.move_player(direction)
            else:
                print("❌ Usage: move <direction>")
                print("   Directions: north, south, east, west")
                
        elif cmd == "chat":
            if args:
                message = " ".join(args)
                await self.client.send_chat(message)
            else:
                print("❌ Usage: chat <message>")
                
        elif cmd == "state":
            state = await self.client.get_game_state()
            if state:
                print("📊 Game State:")
                print(json.dumps(state, indent=2))
            
        elif cmd == "npc":
            if len(args) >= 2:
                npc_id = args[0]
                message = " ".join(args[1:])
                await self.client.interact_with_npc(npc_id, message)
            else:
                print("❌ Usage: npc <npc_id> <message>")
                print("   Example: npc weather_oracle What's the weather in Berlin?")
                
        elif cmd == "tools":
            await self.client.list_tools()
            
        elif cmd in ["quit", "exit", "q"]:
            self.running = False
            
        else:
            print(f"❓ Unknown command: {cmd}")
            print("   Type 'help' for available commands")
    
    def show_help(self):
        """Show available commands"""
        print("\n📋 Available Commands:")
        print("  help                    - Show this help")
        print("  connect                 - Connect to game server")
        print("  register <name>         - Register as a player")
        print("  move <direction>        - Move player (north/south/east/west)")
        print("  chat <message>          - Send chat message")
        print("  npc <id> <message>      - Talk to an NPC")
        print("  state                   - Get game state")
        print("  tools                   - List available server tools")
        print("  quit                    - Exit the client")
        print("\n💡 Example session:")
        print("  > connect")
        print("  > register MyPlayer")
        print("  > move north")
        print("  > chat Hello everyone!")
        print("  > npc weather_oracle What's the weather?")


async def quick_demo():
    """Quick demo showing basic functionality"""
    print("🚀 Running Quick Demo...")
    
    client = SimpleGameClient()
    
    # Connect
    if not await client.connect():
        print("❌ Demo failed - could not connect")
        return
    
    # Register
    if not await client.register_player("DemoPlayer"):
        print("❌ Demo failed - could not register")
        return
    
    # Send greeting
    await client.send_chat("Hello! Demo player here!")
    
    # Move around
    moves = ["north", "east", "south", "west"]
    for direction in moves:
        await client.move_player(direction)
        await asyncio.sleep(1)
    
    # Get state
    state = await client.get_game_state()
    if state:
        print(f"📊 Current game state: {json.dumps(state, indent=2)}")
    
    # Try NPC interaction
    await client.interact_with_npc("weather_oracle", "Hello there!")
    
    await client.disconnect()
    print("✅ Demo completed!")


def main():
    """Main entry point"""
    if len(sys.argv) > 1 and sys.argv[1] == "demo":
        # Run quick demo
        asyncio.run(quick_demo())
    else:
        # Run interactive client
        client = InteractiveGameClient()
        asyncio.run(client.start())


if __name__ == "__main__":
    main()