Simple_Game_Client_Guide.md

Simple Game Client Guide

This guide shows you how to create a simple client to connect to the MMORPG game server using the Model Context Protocol (MCP). No LLM integration required - just basic game operations.

Table of Contents

1. Overview
2. Prerequisites
3. MCP Server Connection
4. Basic Client Example
5. Available Game Operations
6. Command Reference
7. Error Handling
8. Advanced Features

---

Overview

The MMORPG game server exposes its functionality through MCP (Model Context Protocol), allowing external clients to:

• 🎮 Connect to the game server
• 👤 Join/Leave the game as a player
• 🗺️ Move around the game world
• 💬 Send chat messages to other players
• 🎯 Interact with NPCs and game objects
• 📊 Get game state information

Game Server Details

• MCP Endpoint: http://localhost:7868/gradio_api/mcp/sse (when running locally)
• Protocol: Server-Sent Events (SSE) over HTTP
• Authentication: None required for basic operations
• Data Format: JSON messages following MCP specification

---

Prerequisites

Before building your client, ensure you have:

Required Dependencies

pip install mcp asyncio aiohttp contextlib

Python Version

• Python 3.8 or higher
• Support for async/await syntax

Game Server Running

Make sure the MMORPG game server is running:

cd c:/Users/Chris4K/Projekte/projecthub/projects/MMOP_second_try
python app.py

The server should be accessible at http://localhost:7868 (UI) and the MCP endpoint at http://localhost:7868/gradio_api/mcp/sse.

---

MCP Server Connection

Here's how to establish a connection to the game server:

Basic Connection Setup

import asyncio
from mcp import ClientSession
from mcp.client.sse import sse_client
from contextlib import AsyncExitStack

class GameClient:
    def __init__(self, server_url="http://localhost: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 = None
    
    async def connect(self):
        """Connect to the game server"""
        try:
            # Clean up any existing connection
            if self.exit_stack:
                await self.exit_stack.aclose()
            
            self.exit_stack = AsyncExitStack()
            
            # Establish SSE connection
            transport = await self.exit_stack.enter_async_context(
                sse_client(self.server_url)
            )
            read_stream, write_callable = transport
            
            # Create MCP 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
            response = await self.session.list_tools()
            self.tools = response.tools
            
            self.connected = True
            print(f"✅ Connected to game server!")
            print(f"Available commands: {[tool.name for tool in self.tools]}")
            
            return True
            
        except Exception as e:
            print(f"❌ Connection failed: {e}")
            self.connected = False
            return False
    
    async def disconnect(self):
        """Disconnect from the game server"""
        if self.exit_stack:
            await self.exit_stack.aclose()
        self.connected = False
        print("🔌 Disconnected from game server")

Connection Test

async def test_connection():
    client = GameClient()
    if await client.connect():
        print("Connection successful!")
        await client.disconnect()
    else:
        print("Connection failed!")

# Run the test
asyncio.run(test_connection())

---

Basic Client Example

Here's a complete working client that can perform basic game operations:

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

class SimpleGameClient:
    def __init__(self, server_url="http://localhost: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 = str(uuid.uuid4())[:8]
        self.agent_id = None
        self.player_name = None
    
    async def connect(self):
        """Connect to the game server"""
        try:
            if self.exit_stack:
                await self.exit_stack.aclose()
            
            self.exit_stack = AsyncExitStack()
            
            transport = await self.exit_stack.enter_async_context(
                sse_client(self.server_url)
            )
            read_stream, write_callable = transport
            
            self.session = await self.exit_stack.enter_async_context(
                ClientSession(read_stream, write_callable)
            )
            
            await self.session.initialize()
            response = await self.session.list_tools()
            self.tools = response.tools
            
            self.connected = True
            return True
            
        except Exception as e:
            print(f"❌ Connection failed: {e}")
            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
        
        try:
            # Find the register tool
            register_tool = next((t for t in self.tools if 'register' in t.name), None)
            if not register_tool:
                print("❌ Register tool not available")
                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)
            if "registered" in content.lower():
                self.player_name = player_name
                # Extract agent_id from response if available
                if "agent_id" in content or "ID:" in content:
                    # Parse agent ID from response
                    lines = content.split('\n')
                    for line in lines:
                        if "agent_id" in line.lower() or "id:" in line:
                            parts = line.split(':')
                            if len(parts) > 1:
                                self.agent_id = parts[1].strip()
                                break
                
                print(f"✅ Registered as player: {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.connected or not self.agent_id:
            print("❌ Not connected or not registered")
            return False
        
        try:
            # Find the move tool
            move_tool = next((t for t in self.tools if 'move' in t.name), None)
            if not move_tool:
                print("❌ Move tool not available")
                return False
            
            # Move the player
            result = await self.session.call_tool(
                move_tool.name,
                {"client_id": self.client_id, "direction": direction}
            )
            
            content = self._extract_content(result)
            print(f"🚶 Move result: {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.connected or not self.agent_id:
            print("❌ Not connected or not registered")
            return False
        
        try:
            # Find the chat tool
            chat_tool = next((t for t in self.tools if 'chat' in t.name), None)
            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 = next((t for t in self.tools if 'state' in t.name or 'game' in t.name), None)
            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.connected or not self.agent_id:
            print("❌ Not connected or not registered")
            return False
        
        try:
            # Find the interact tool
            interact_tool = next((t for t in self.tools if 'interact' in t.name or 'npc' in t.name), None)
            if not interact_tool:
                print("❌ Interact 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 {npc_id}: {content}")
            return True
            
        except Exception as e:
            print(f"❌ Interaction error: {e}")
            return False
    
    def _extract_content(self, result):
        """Extract content from MCP result"""
        if hasattr(result, 'content') and result.content:
            if isinstance(result.content, list):
                return ''.join(str(item) for item in result.content)
            return str(result.content)
        return str(result)
    
    async def disconnect(self):
        """Disconnect from the server"""
        if self.exit_stack:
            await self.exit_stack.aclose()
        self.connected = False
        print("🔌 Disconnected from server")

# Import required modules at the top
from mcp import ClientSession
from mcp.client.sse import sse_client
from contextlib import AsyncExitStack

---

Available Game Operations

The game server exposes these main operations through MCP tools:

1. Player Management

• register_ai_agent - Register a new player
• move_agent - Move player in game world
• get_game_state - Get current world state

2. Communication

• send_chat - Send chat messages
• interact_with_npc - Talk to NPCs

3. Game Information

• get_game_state - World state and player list
• Player proximity - Detect nearby players/NPCs

Tool Parameters

Each tool expects specific parameters:

# Register Player
{
    "name": "PlayerName",
    "client_id": "unique_client_id"
}

# Move Player
{
    "client_id": "your_client_id",
    "direction": "north|south|east|west"
}

# Send Chat
{
    "client_id": "your_client_id", 
    "message": "Hello world!"
}

# Interact with NPC
{
    "client_id": "your_client_id",
    "npc_id": "npc_identifier",
    "message": "Your message to NPC"
}

---

Command Reference

Basic Usage Example

async def main():
    # Create and connect client
    client = SimpleGameClient()
    
    if not await client.connect():
        print("Failed to connect!")
        return
    
    # Register as a player
    if not await client.register_player("MyPlayer"):
        print("Failed to register!")
        return
    
    # Get current game state
    state = await client.get_game_state()
    print(f"Game state: {state}")
    
    # Move around
    await client.move_player("north")
    await client.move_player("east")
    
    # Send a chat message
    await client.send_chat("Hello everyone!")
    
    # Interact with an NPC
    await client.interact_with_npc("weather_oracle", "What's the weather in Berlin?")
    
    # Disconnect
    await client.disconnect()

# Run the client
asyncio.run(main())

Interactive Console Client

async def interactive_client():
    client = SimpleGameClient()
    
    print("🎮 MMORPG Simple Client")
    print("Commands: connect, register <name>, move <direction>, chat <message>, state, quit")
    
    while True:
        command = input("> ").strip().split()
        
        if not command:
            continue
            
        cmd = command[0].lower()
        
        if cmd == "connect":
            if await client.connect():
                print("✅ Connected!")
            else:
                print("❌ Connection failed!")
                
        elif cmd == "register" and len(command) > 1:
            name = " ".join(command[1:])
            if await client.register_player(name):
                print(f"✅ Registered as {name}")
            else:
                print("❌ Registration failed!")
                
        elif cmd == "move" and len(command) > 1:
            direction = command[1]
            await client.move_player(direction)
            
        elif cmd == "chat" and len(command) > 1:
            message = " ".join(command[1:])
            await client.send_chat(message)
            
        elif cmd == "state":
            state = await client.get_game_state()
            print(f"📊 Game State: {json.dumps(state, indent=2)}")
            
        elif cmd == "npc" and len(command) > 2:
            npc_id = command[1]
            message = " ".join(command[2:])
            await client.interact_with_npc(npc_id, message)
            
        elif cmd == "quit":
            await client.disconnect()
            break
            
        else:
            print("❓ Unknown command")

# Run interactive client
asyncio.run(interactive_client())

---

Error Handling

Common Issues and Solutions

1. Connection Errors

async def robust_connect(client, max_retries=3):
    for attempt in range(max_retries):
        try:
            if await client.connect():
                return True
            print(f"Attempt {attempt + 1} failed, retrying...")
            await asyncio.sleep(2)
        except Exception as e:
            print(f"Connection attempt {attempt + 1} error: {e}")
    
    print("❌ All connection attempts failed")
    return False

2. Tool Not Available

def find_tool_safe(tools, keywords):
    """Safely find a tool by keywords"""
    for keyword in keywords:
        tool = next((t for t in tools if keyword in t.name.lower()), None)
        if tool:
            return tool
    return None

# Usage
move_tool = find_tool_safe(client.tools, ['move', 'agent', 'player'])
if not move_tool:
    print("❌ No movement tool available")

3. Response Parsing

def safe_parse_response(result):
    """Safely parse MCP response"""
    try:
        content = client._extract_content(result)
        
        # Try JSON first
        try:
            return json.loads(content)
        except json.JSONDecodeError:
            # Return as structured text
            return {"message": content, "type": "text"}
            
    except Exception as e:
        return {"error": str(e), "type": "error"}

---

Advanced Features

1. Automatic Reconnection

class RobustGameClient(SimpleGameClient):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.auto_reconnect = True
        self.reconnect_delay = 5
    
    async def _ensure_connected(self):
        """Ensure connection is active, reconnect if needed"""
        if not self.connected and self.auto_reconnect:
            print("🔄 Attempting to reconnect...")
            await self.connect()
            if self.player_name:
                await self.register_player(self.player_name)
    
    async def move_player(self, direction):
        await self._ensure_connected()
        return await super().move_player(direction)

2. Event Monitoring

class EventMonitorClient(SimpleGameClient):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.event_handlers = {}
    
    def on_chat_message(self, handler):
        """Register handler for chat messages"""
        self.event_handlers['chat'] = handler
    
    def on_player_move(self, handler):
        """Register handler for player movements"""
        self.event_handlers['move'] = handler
    
    async def poll_events(self):
        """Poll for game events"""
        while self.connected:
            try:
                state = await self.get_game_state()
                # Process state changes and trigger handlers
                # Implementation depends on game state format
                await asyncio.sleep(1)
            except Exception as e:
                print(f"Event polling error: {e}")
                await asyncio.sleep(5)

3. Batch Operations

async def batch_moves(client, moves):
    """Execute multiple moves in sequence"""
    for direction in moves:
        result = await client.move_player(direction)
        if not result:
            print(f"❌ Failed to move {direction}")
            break
        await asyncio.sleep(0.5)  # Small delay between moves

# Usage
await batch_moves(client, ["north", "north", "east", "south"])

4. Configuration Management

import json

class ConfigurableClient(SimpleGameClient):
    def __init__(self, config_file="client_config.json"):
        # Load configuration
        try:
            with open(config_file, 'r') as f:
                config = json.load(f)
        except FileNotFoundError:
            config = self.default_config()
            
        super().__init__(config.get('server_url', 'http://localhost:7860/gradio_api/mcp/sse'))
        self.config = config
    
    def default_config(self):
        return {
            "server_url": "http://localhost:7860/gradio_api/mcp/sse",
            "player_name": "DefaultPlayer",
            "auto_reconnect": True,
            "move_delay": 0.5
        }

---

Quick Start Script

Save this as quick_client.py for immediate testing:

#!/usr/bin/env python3
"""
Quick Game Client - Connect and play immediately
Usage: python quick_client.py [player_name]
"""

import asyncio
import sys
from simple_game_client import SimpleGameClient

async def quick_play(player_name="TestPlayer"):
    client = SimpleGameClient()
    
    print(f"🎮 Connecting as {player_name}...")
    
    # Connect and register
    if not await client.connect():
        print("❌ Failed to connect!")
        return
    
    if not await client.register_player(player_name):
        print("❌ Failed to register!")
        return
    
    print("✅ Connected and registered successfully!")
    
    # Basic game session
    await client.send_chat(f"Hello! {player_name} has joined the game!")
    
    # Move around a bit
    moves = ["north", "east", "south", "west"]
    for move in moves:
        print(f"🚶 Moving {move}...")
        await client.move_player(move)
        await asyncio.sleep(1)
    
    # Get final state
    state = await client.get_game_state()
    print(f"📊 Final state: {state}")
    
    await client.disconnect()
    print("👋 Session ended!")

if __name__ == "__main__":
    player_name = sys.argv[1] if len(sys.argv) > 1 else "TestPlayer"
    asyncio.run(quick_play(player_name))

Run with:

python quick_client.py MyPlayerName

---

Next Steps

1. Customize the client for your specific needs
2. Add game-specific features like inventory management
3. Implement advanced AI for automated gameplay
4. Create GUI interfaces using tkinter, PyQt, or web frameworks
5. Build monitoring tools for game statistics

This guide provides the foundation for any client application that needs to interact with the MMORPG game server!