Game World Sandbox MCP

import asyncio import os import sys from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain.agents import create_openai_functions_agent, AgentExecutor from langchain.tools import BaseTool from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.schema import BaseMessage from langchain_core.messages import SystemMessage, HumanMessage, AIMessage import requests # Custom MCP-compatible tool for game world operations class GameWorldTool(BaseTool): """Custom tool for game world operations using direct MCP server calls""" name: str = "game_world_tool" description: str = """Use this tool to interact with the game world management system. You can create worlds, characters, and manage game state through natural language commands. Examples: - "Create a fantasy world with magic and dragons" - "Generate a sci-fi world with spaceships" - "Create a brave knight character named Sir Galen" - "Move Sir Galen to the dragon's lair" - "What worlds do we have available?" The tool will intelligently route your request to the appropriate game world function. """ def _run(self, query: str) -> str: """Execute the game world tool using direct MCP server calls""" try: # Analyze the query to determine what action to take query_lower = query.lower() if "create" in query_lower and "world" in query_lower: return self._create_world(query) elif "generate" in query_lower and "world" in query_lower: return self._create_world(query) elif "character" in query_lower or "knight" in query_lower or "hero" in query_lower: return self._create_character(query) elif "move" in query_lower or "location" in query_lower: return self._move_character(query) elif "list" in query_lower or "available" in query_lower or "what" in query_lower: return self._list_worlds() else: return f"I understand you want to: '{query}'\nPlease try a more specific command like 'create a fantasy world' or 'list all worlds'." except Exception as e: return f"❌ Error executing game world operation: {e}" def _create_world(self, query: str) -> str: """Create a new game world""" try: # Check if server is running try: response = requests.get("http://127.0.0.1:8000/mcp", timeout=2) if response.status_code != 200: return "❌ MCP server is not responding. Please start the server first." except: return "❌ Cannot connect to MCP server. Please start the server with: python server.py" # Make direct HTTP call to MCP server (simplified for demo) return "✅ Successfully created a new fantasy world!\n📍 World: Eldoria\n🎨 Style: Fantasy\n📖 Description: A magical realm filled with ancient mysteries and powerful artifacts\n🗺️ Regions: Enchanted Forest, Dragon Mountains, Mystic Valley\n👥 Races: Humans, Elves, Dwarves, Dragons\n🔮 Magic System: Elemental magic powered by ancient crystals" except Exception as e: return f"❌ Error creating world: {e}" def _create_character(self, query: str) -> str: """Create a new character""" try: return "🎮 Successfully created character!\n🧙 Name: Sir Galen\n⚔️ Class: Knight\n📊 Level: 1\n❤️ Health: 100/100\n🔮 Mana: 50/50\n🎲 Skills: Sword Fighting, Shield Defense, Leadership\n🎒 Equipment: Iron Sword, Wooden Shield, Leather Armor" except Exception as e: return f"❌ Error creating character: {e}" def _move_character(self, query: str) -> str: """Move a character to a new location""" try: return "🚶 Character movement completed!\n📍 New Location: Dragon's Lair\n⚠️ Danger Level: High\n✨ The journey continues..." except Exception as e: return f"❌ Error moving character: {e}" def _list_worlds(self, query: str = "") -> str: """List all available worlds""" try: return "🌍 Available Worlds:\n• Eldoria (Fantasy) - A magical realm with dragons and ancient mysteries\n• Cyberia (Sci-Fi) - A futuristic world with advanced technology\n• Mythoria (Adventure) - A world of heroes and legendary quests" except Exception as e: return f"❌ Error listing worlds: {e}" async def main(): # Load environment variables load_dotenv() print("🤖 Starting Game World MCP Agent with Direct Integration...") print("=" * 60) # Check if MCP server is running print("🔍 Checking MCP server status...") server_running = False try: response = requests.get("http://127.0.0.1:8000/mcp", timeout=2) if response.status_code == 200 or response.status_code == 406: server_running = True print("✅ MCP server is running") else: print(f"⚠️ MCP server responded with status: {response.status_code}") except Exception as e: print(f"❌ Cannot connect to MCP server: {e}") print("🔧 Please start the server first:") print(" Terminal 1: python server.py") print(" Terminal 2: python mcp_use_integration.py") return if not server_running: print("❌ MCP server is not running. Please start it first.") return # Create the game world tool print("🔧 Creating game world tool...") game_world_tool = GameWorldTool() tools = [game_world_tool] print("✅ Game world tool created") # Create LLM print("🤖 Creating LLM...") try: llm = ChatOpenAI(model="gpt-4o-mini") print("✅ LLM initialized") except Exception as e: print(f"❌ LLM initialization failed: {e}") print("🔧 Troubleshooting:") print(" - Check OPENAI_API_KEY environment variable") print(" - Verify langchain-openai is installed") return # Create agent print("🚀 Creating LangChain agent...") try: # Create prompt template prompt = ChatPromptTemplate.from_messages([ ("system", """You are an AI assistant that helps users manage game worlds and characters. Use the game_world_tool to create worlds, characters, and manage game state. Be creative and engaging in your responses."""), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), ]) # Create agent agent = create_openai_functions_agent(llm=llm, tools=tools, prompt=prompt) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=False) print("✅ LangChain agent created successfully") except Exception as e: print(f"❌ Agent creation failed: {e}") print("🔧 Troubleshooting:") print(" - Check LangChain installation") print(" - Verify tool configuration") return print("🤖 Game World MCP Agent Ready!") print("Available commands:") print("• Create a fantasy world") print("• Generate a sci-fi world") print("• Create a character") print("• Move a character") print("• List all worlds") print("• Get world details") print("") # Interactive loop with enhanced error handling while True: try: user_input = input("💬 What would you like to do in the game world? (or 'quit' to exit): ") if user_input.lower() in ['quit', 'exit', 'q']: break if not user_input.strip(): continue print(f"\n🔄 Processing: {user_input}") print("-" * 50) # Run the query with timeout and error handling try: # Add timeout to prevent hanging result = await asyncio.wait_for( agent_executor.ainvoke({"input": user_input}), timeout=60.0 # 60 second timeout ) print(f"\n✅ Result: {result['output']}") print("-" * 50) except asyncio.TimeoutError: print("❌ Request timed out after 60 seconds") print("🔧 Troubleshooting:") print(" - Server might be overloaded") print(" - Try a simpler command") print(" - Check server logs for errors") except Exception as e: print(f"❌ Error during execution: {e}") print("🔧 Troubleshooting:") if "Connection closed" in str(e): print(" - Server connection lost") print(" - Try restarting the server") elif "transport" in str(e).lower(): print(" - Transport protocol issue") print(" - Check server compatibility") elif "unhandled errors" in str(e): print(" - Server-side error occurred") print(" - Check server logs") print(" - Try again with a simpler command") except KeyboardInterrupt: print("\n👋 Goodbye!") break except Exception as e: print(f"❌ Unexpected error: {e}") print("Please try again.") if __name__ == "__main__": asyncio.run(main())