Spaces:
Running
Running
| import gradio as gr | |
| import anthropic | |
| import os | |
| import json | |
| import re | |
| import html as html_module | |
| import asyncio | |
| from mcp import ClientSession, StdioServerParameters | |
| from mcp.client.stdio import stdio_client | |
| # Anthropic client | |
| client = anthropic.Anthropic() | |
| # MCP Server configurations | |
| STEAM_MCP_SERVER = StdioServerParameters( | |
| command="python", | |
| args=["tools/steam_mcp_server.py"], | |
| env={ | |
| "STEAM_API_KEY": os.getenv("STEAM_API_KEY", ""), | |
| "ANTHROPIC_API_KEY": os.getenv("ANTHROPIC_API_KEY", ""), | |
| } | |
| ) | |
| IGDB_MCP_SERVER = StdioServerParameters( | |
| command="uvx", | |
| args=["--from", "git+https://github.com/bielacki/igdb-mcp-server.git", "igdb-mcp-server"], | |
| env={ | |
| # IGDB MCP server expects IGDB_CLIENT_ID/SECRET (these are Twitch credentials) | |
| "IGDB_CLIENT_ID": os.getenv("IGDB_CLIENT_ID", ""), | |
| "IGDB_CLIENT_SECRET": os.getenv("IGDB_CLIENT_SECRET", "") | |
| } | |
| ) | |
| # System prompt for the agent | |
| SYSTEM_PROMPT = """You are Dream Game Forge, an autonomous agent that discovers what game a person truly needs. | |
| Your process: | |
| 1. INVESTIGATE: When a user provides their Steam ID, announce that you're connecting to the Steam MCP server, then fetch their library using the get_steam_library tool. Briefly mention the MCP connection in your response (e.g., "Connecting to Steam MCP server to fetch your library..."). Note playtime patterns - what do they sink hundreds of hours into? What do they buy and never play? | |
| 2. ENRICH: For the user's top 3-5 most played games, announce that you're querying the IGDB MCP server for enrichment, then use search_games tool to fetch detailed genre, theme, platform, and rating data. Mention the MCP usage (e.g., "Querying IGDB MCP server for game metadata..."). This provides deeper insight into what truly resonates with them. | |
| 3. ANALYZE: Build a taste profile from their gaming patterns combining Steam playtime data with IGDB genre/theme insights. Identify patterns in mechanics, pacing, and emotional engagement. | |
| 4. COLLABORATE: Present your analysis findings to the user with a summary of their gaming DNA. Then ask: "Anything you'd like me to consider? (play session length, themes to avoid, specific mechanics you enjoy, etc.)" | |
| - WAIT for the user's response before proceeding | |
| - This is a REQUIRED step - do not skip to game generation without user input | |
| - If they say "no" or "nothing", that's fine, proceed to design | |
| 5. IDENTIFY THE GAP: Based on your analysis AND the user's input, determine what game *should* exist for this person but doesn't. Be specific and creative, incorporating their preferences. | |
| 6. DESIGN: Write a mini Game Design Document - core loop, unique mechanic, emotional hook, incorporating user preferences from the collaboration phase. | |
| 7. BUILD: Generate a playable HTML5 Canvas prototype that captures the game concept. | |
| CRITICAL - Game Generation Requirements: | |
| STRUCTURE & POLISH: | |
| - Create a COMPLETE, self-contained HTML file (no external dependencies) | |
| - Use HTML5 Canvas for rendering | |
| - Include all CSS in <style> tags and all JavaScript in <script> tags | |
| - The <title> tag in the HTML MUST exactly match the game name you designed in the Game Design Document | |
| - Wrap the entire HTML in a code block with ```html markers | |
| GAME STATES & FLOW: | |
| - TITLE SCREEN: Display game name with "Click to Start" prompt (clickable anywhere to begin) | |
| - PLAYING STATE: Active gameplay with score tracking, visual feedback, and controls | |
| - GAME OVER SCREEN: Show final score with "Play Again" button to restart | |
| - Implement clear WIN and LOSE conditions with appropriate end screens | |
| - Display brief instructions on title screen (controls, objective) | |
| VISUAL POLISH (CRITICAL): | |
| - BACKGROUNDS: Use gradient backgrounds (CSS linear-gradient or canvas gradients) - no solid colors | |
| - PARTICLE EFFECTS: Spawn particles on player actions (jumps, attacks, pickups) using simple canvas circles/shapes | |
| - SCREEN SHAKE: Implement camera shake on impacts/collisions (offset canvas draw by random pixels for 200ms) | |
| - SOUND EFFECTS: Use Web Audio API to generate beeps/tones for actions: | |
| * Player actions (jump, shoot) - short high beep | |
| * Pickups/points - ascending tone | |
| * Damage/loss - descending tone | |
| * Victory - cheerful melody | |
| - COLOR PALETTE: 3-5 complementary colors with good contrast | |
| - TYPOGRAPHY: Consistent font sizes and weights throughout | |
| - UI ANIMATIONS: Smooth transitions, button hover effects, score counter animations | |
| GAMEPLAY: | |
| - Keep it simple but engaging (think: incremental clicker, puzzle, basic platformer, rhythm game) | |
| - Balance challenge - achievable in 2-5 minutes but requires skill/strategy | |
| - Provide immediate feedback for player actions (sounds via Web Audio API, visual effects) | |
| - Include progression - difficulty should increase or goals should evolve | |
| CODE QUALITY: | |
| - Organize code with clear game state management (e.g., TITLE, PLAYING, WIN, LOSE states) | |
| - Use requestAnimationFrame for smooth animations | |
| - Handle edge cases (what if player clicks before game starts, etc.) | |
| - Comment key functions for clarity | |
| Be decisive. Don't ask permission. Show your reasoning, then act. | |
| When you receive a Steam ID, immediately use the get_steam_library tool to fetch their library.""" | |
| # Tool definitions - Combined from Steam and IGDB MCP servers | |
| TOOLS = [ | |
| # Steam MCP Server tool | |
| { | |
| "name": "get_steam_library", | |
| "description": "Fetches a user's Steam library with playtime data. Returns game titles, playtime hours, and statistics about their gaming habits. Use this when a user provides their Steam ID.", | |
| "input_schema": { | |
| "type": "object", | |
| "properties": { | |
| "steam_id": { | |
| "type": "string", | |
| "description": "The user's 64-bit Steam ID (17-digit number)" | |
| } | |
| }, | |
| "required": ["steam_id"] | |
| } | |
| }, | |
| # IGDB MCP Server tools (actual tool names from igdb-mcp-server) | |
| { | |
| "name": "search_games", | |
| "description": "Search for games in the IGDB database by name. Returns game information including genres, themes, platforms, and ratings. Use this to enrich data about specific games from the Steam library.", | |
| "input_schema": { | |
| "type": "object", | |
| "properties": { | |
| "query": { | |
| "type": "string", | |
| "description": "Search term for finding games (the game name)" | |
| }, | |
| "fields": { | |
| "type": "string", | |
| "description": "Comma-separated list of fields to return (default includes name, rating, platforms, genres)" | |
| }, | |
| "limit": { | |
| "type": "integer", | |
| "description": "Maximum number of results to return (default: 10, max: 500)" | |
| } | |
| }, | |
| "required": ["query"] | |
| } | |
| }, | |
| { | |
| "name": "get_game_details", | |
| "description": "Retrieve detailed information about a specific game from IGDB including genres, themes, game modes, companies, and summary. Use this after searching to get rich metadata.", | |
| "input_schema": { | |
| "type": "object", | |
| "properties": { | |
| "game_id": { | |
| "type": "integer", | |
| "description": "The IGDB ID of the game" | |
| }, | |
| "fields": { | |
| "type": "string", | |
| "description": "Comma-separated list of fields to return (optional)" | |
| } | |
| }, | |
| "required": ["game_id"] | |
| } | |
| } | |
| ] | |
| async def execute_tool_via_mcp(tool_name: str, tool_input: dict) -> str: | |
| """ | |
| Execute a tool via MCP protocol. | |
| Routes to the appropriate MCP server based on tool name: | |
| - Steam tools: Steam MCP Server | |
| - IGDB tools: IGDB MCP Server | |
| Demonstrates proper MCP client usage by: | |
| 1. Connecting to the MCP server via stdio | |
| 2. Initializing a client session | |
| 3. Calling the tool through MCP protocol | |
| 4. Returning the result | |
| """ | |
| try: | |
| # Determine which MCP server to use based on tool name | |
| # IGDB tools: search_games, get_game_details, get_most_anticipated_games, custom_query | |
| # Steam tools: get_steam_library | |
| if tool_name in ["search_games", "get_game_details", "get_most_anticipated_games", "custom_query"]: | |
| server_params = IGDB_MCP_SERVER | |
| server_name = "IGDB" | |
| else: | |
| server_params = STEAM_MCP_SERVER | |
| server_name = "Steam" | |
| # Connect to MCP server using stdio transport | |
| async with stdio_client(server_params) as (read, write): | |
| async with ClientSession(read, write) as session: | |
| # Initialize the MCP session | |
| await session.initialize() | |
| # List available tools from the MCP server | |
| tools = await session.list_tools() | |
| print(f"[MCP] Connected to {server_name} MCP Server. Available tools: {[t.name for t in tools.tools]}") | |
| # Call the tool through MCP protocol | |
| result = await session.call_tool(tool_name, tool_input) | |
| # Extract text content from MCP response | |
| if result.content: | |
| return result.content[0].text | |
| else: | |
| return "No response from MCP server" | |
| except Exception as e: | |
| return f"Error executing tool via MCP ({server_name}): {str(e)}" | |
| def execute_tool(tool_name: str, tool_input: dict) -> str: | |
| """ | |
| Synchronous wrapper for MCP tool execution. | |
| Runs the async MCP client in an event loop. | |
| """ | |
| return asyncio.run(execute_tool_via_mcp(tool_name, tool_input)) | |
| def extract_html_game(text: str) -> str: | |
| """Extract HTML game code from response text.""" | |
| # Look for HTML code blocks with ```html markers | |
| html_pattern = r'```html\s*(<!DOCTYPE html>.*?</html>)\s*```' | |
| match = re.search(html_pattern, text, re.DOTALL | re.IGNORECASE) | |
| if match: | |
| return match.group(1).strip() | |
| # Alternative: look for standalone HTML (from <!DOCTYPE to </html>) | |
| html_standalone = r'(<!DOCTYPE html>.*?</html>)' | |
| match = re.search(html_standalone, text, re.DOTALL | re.IGNORECASE) | |
| if match: | |
| return match.group(1).strip() | |
| return None | |
| def respond(message, history: list[dict[str, str]]): | |
| """Agentic response with tool loop. Returns (response_text, game_html).""" | |
| # Convert Gradio history format to Anthropic format | |
| messages = [] | |
| for msg in history: | |
| messages.append({ | |
| "role": msg["role"], | |
| "content": msg["content"] | |
| }) | |
| messages.append({"role": "user", "content": message}) | |
| # Agentic loop | |
| response_text = "" | |
| max_iterations = 10 | |
| for iteration in range(max_iterations): | |
| response = client.messages.create( | |
| model="claude-sonnet-4-20250514", | |
| max_tokens=8192, # Increased for game generation | |
| system=SYSTEM_PROMPT, | |
| tools=TOOLS, | |
| messages=messages | |
| ) | |
| # Collect text responses | |
| for block in response.content: | |
| if block.type == "text": | |
| response_text += block.text | |
| # Check if we need to execute tools | |
| if response.stop_reason == "tool_use": | |
| # Execute all tool calls | |
| tool_results = [] | |
| for block in response.content: | |
| if block.type == "tool_use": | |
| tool_name = block.name | |
| tool_input = block.input | |
| tool_id = block.id | |
| # Execute the tool | |
| result = execute_tool(tool_name, tool_input) | |
| tool_results.append({ | |
| "type": "tool_result", | |
| "tool_use_id": tool_id, | |
| "content": result | |
| }) | |
| # Add assistant response and tool results to messages | |
| messages.append({ | |
| "role": "assistant", | |
| "content": response.content | |
| }) | |
| messages.append({ | |
| "role": "user", | |
| "content": tool_results | |
| }) | |
| elif response.stop_reason == "end_turn": | |
| # Done - return the accumulated response | |
| break | |
| else: | |
| # Unexpected stop reason | |
| break | |
| # Extract HTML game if present | |
| game_html = extract_html_game(response_text) | |
| return response_text, game_html | |
| # Custom Gradio interface | |
| with gr.Blocks(title="🎮 Dream Game Forge") as demo: | |
| gr.Markdown("# 🎮 Dream Game Forge") | |
| gr.Markdown("""**Start by entering your Steam ID below.** The agent will analyze your library, | |
| then collaborate with you to design your perfect game!""") | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| chatbot = gr.Chatbot(label="Agent Analysis", height=600) | |
| msg = gr.Textbox( | |
| label="Your message", | |
| placeholder="Start: Enter your Steam ID (e.g., 76561198162013924) | Continue: Answer the agent's questions", | |
| lines=2 | |
| ) | |
| submit = gr.Button("Send", variant="primary") | |
| clear = gr.Button("Clear") | |
| with gr.Column(scale=1): | |
| game_display = gr.HTML( | |
| label="Your Game Prototype", | |
| value="<div style='padding: 20px; text-align: center; color: #666;'>Your generated game will appear here...</div>" | |
| ) | |
| def user_submit(message, history): | |
| """Handle user message submission.""" | |
| return "", history + [[message, None]] | |
| def bot_respond(history): | |
| """Generate bot response and game with full conversation context.""" | |
| # Convert Gradio history to Anthropic message format | |
| # Gradio history is [[user_msg, bot_msg], [user_msg, bot_msg], ...] | |
| # We need all previous exchanges except the current pending one | |
| previous_messages = [] | |
| for i, (user_msg, bot_msg) in enumerate(history[:-1]): | |
| previous_messages.append({"role": "user", "content": user_msg}) | |
| if bot_msg: | |
| previous_messages.append({"role": "assistant", "content": bot_msg}) | |
| # Get the current user message | |
| current_message = history[-1][0] | |
| # Call respond with full conversation history | |
| response_text, game_html = respond(current_message, previous_messages) | |
| # Update chat history | |
| history[-1][1] = response_text | |
| # Update game display | |
| if game_html: | |
| # Wrap in iframe for proper rendering with taller height to prevent cutoff | |
| escaped_html = html_module.escape(game_html, quote=True) | |
| game_output = f'<iframe srcdoc="{escaped_html}" width="100%" height="800" frameborder="0" style="border: 2px solid #444; border-radius: 8px; display: block;"></iframe>' | |
| else: | |
| game_output = "<div style='padding: 20px; text-align: center; color: #666;'>No game generated yet. The agent is analyzing your library...</div>" | |
| return history, game_output | |
| # Event handlers | |
| msg.submit(user_submit, [msg, chatbot], [msg, chatbot]).then( | |
| bot_respond, [chatbot], [chatbot, game_display] | |
| ) | |
| submit.click(user_submit, [msg, chatbot], [msg, chatbot]).then( | |
| bot_respond, [chatbot], [chatbot, game_display] | |
| ) | |
| clear.click(lambda: ([], "<div style='padding: 20px; text-align: center; color: #666;'>Your generated game will appear here...</div>"), None, [chatbot, game_display]) | |
| if __name__ == "__main__": | |
| demo.launch() | |