Create app.py

app.py

@@ -0,0 +1,302 @@
+import os
+import requests
+import base64
+from dotenv import load_dotenv
+from mcp.server.fastmcp import FastMCP, Image
+
+load_dotenv()
+
+# --- Configuration ---
+# URL of the Flask app API - configurable via environment variable
+FLASK_API_URL = os.getenv("FLASK_API_URL", "http://127.0.0.1:5000")
+
+# --- MCP Server Setup ---
+mcp = FastMCP(
+    name="GeoGuessrAgent",
+    host="127.0.0.1",
+    port=7860,
+)
+
+# --- Game State Management ---
+# Store the current game ID and basic state
+active_game = {}
+
+# --- Flask API Helper Functions ---
+def call_flask_api(endpoint, method='GET', json_data=None):
+    """Helper function to call Flask API endpoints"""
+    url = f"{FLASK_API_URL}{endpoint}"
+    try:
+        if method == 'POST':
+            response = requests.post(url, json=json_data, headers={'Content-Type': 'application/json'}, timeout=30)
+        else:
+            response = requests.get(url, timeout=30)
+        
+        if response.status_code in [200, 201]:
+            return response.json()
+        else:
+            error_msg = f"API call failed: {response.status_code} - {response.text}"
+            print(f"Flask API Error: {error_msg}")
+            raise Exception(error_msg)
+    except requests.exceptions.ConnectionError as e:
+        error_msg = f"Could not connect to Flask API at {FLASK_API_URL}. Make sure the Flask server is running. Error: {str(e)}"
+        print(f"Connection Error: {error_msg}")
+        raise Exception(error_msg)
+    except requests.exceptions.Timeout as e:
+        error_msg = f"Timeout calling Flask API at {FLASK_API_URL}. Error: {str(e)}"
+        print(f"Timeout Error: {error_msg}")
+        raise Exception(error_msg)
+    except Exception as e:
+        error_msg = f"API call error: {str(e)}"
+        print(f"General Error: {error_msg}")
+        raise Exception(error_msg)
+
+def base64_to_image_bytes(base64_string):
+    """Convert base64 string to image bytes"""
+    try:
+        if not base64_string:
+            raise ValueError("Empty base64 string provided")
+        
+        # Remove any data URL prefix if present
+        if base64_string.startswith('data:'):
+            base64_string = base64_string.split(',', 1)[1]
+        
+        # Decode the base64 string
+        image_bytes = base64.b64decode(base64_string)
+        
+        if len(image_bytes) == 0:
+            raise ValueError("Decoded image is empty")
+        
+        print(f"Successfully decoded image: {len(image_bytes)} bytes")
+        return image_bytes
+        
+    except Exception as e:
+        print(f"Error decoding base64 image: {e}")
+        raise ValueError(f"Failed to decode base64 image: {str(e)}")
+
+# --- MCP Tools ---
+
+@mcp.tool()
+def start_game(difficulty: str = "easy", player_name: str = "MCP Agent") -> Image:
+    """
+    Starts a new GeoGuessr game by calling the Flask API.
+    Args:
+        difficulty (str): The difficulty of the game ('easy', 'medium', 'hard').
+        player_name (str): The name of the player/agent.
+    Returns:
+        Image: The first Street View image with compass overlay.
+    """
+    global active_game
+    
+    # Call Flask API to start game
+    game_data = call_flask_api('/start_game', 'POST', {
+        'difficulty': difficulty,
+        'player_name': player_name
+    })
+    
+    # Store game state
+    active_game = {
+        'game_id': game_data['game_id'],
+        'player_name': game_data['player_name'],
+        'game_over': False
+    }
+    
+    # Convert base64 image to bytes and return as Image
+    if game_data.get('streetview_image'):
+        try:
+            image_bytes = base64_to_image_bytes(game_data['streetview_image'])
+            print(f"Successfully started game {active_game['game_id']} for player {active_game['player_name']}")
+            return Image(data=image_bytes, format="jpeg")
+        except Exception as e:
+            print(f"Error processing Street View image: {e}")
+            raise Exception(f"Failed to process Street View image: {str(e)}")
+    else:
+        raise Exception("No Street View image received from the game")
+
+
+
+@mcp.tool()
+def move(direction: str = None, degree: float = None, distance: float = 0.1) -> Image:
+    """
+    Moves the player in a specified direction by calling the Flask API.
+    Args:
+        direction (str, optional): Direction to move (N, NE, E, SE, S, SW, W, NW).
+        degree (float, optional): Precise direction in degrees (0-360).
+        distance (float): Distance to move in kilometers (default: 0.1km = 100m).
+    Returns:
+        Image: The new Street View image with compass overlay.
+    """
+    global active_game
+    if not active_game or not active_game.get('game_id'):
+        raise ValueError("Game not started. Call start_game() first.")
+    if active_game.get('game_over'):
+        raise ValueError("Game is over.")
+    
+    
+    # Prepare move data
+    move_data = {'distance': distance}
+    if direction:
+        move_data['direction'] = direction
+    elif degree is not None:
+        move_data['degree'] = degree
+    else:
+        raise ValueError("Must provide either direction or degree parameter.")
+    
+    # Call Flask API to move
+    game_id = active_game['game_id']
+    move_result = call_flask_api(f'/game/{game_id}/move', 'POST', move_data)
+    
+    # Convert base64 image to bytes and return as Image
+    if move_result.get('streetview_image'):
+        try:
+            image_bytes = base64_to_image_bytes(move_result['streetview_image'])
+            direction_info = move_result.get('moved_direction', 'unknown direction')
+            distance_info = move_result.get('distance_moved_km', 0) * 1000
+            print(f"Successfully moved {direction_info} for {distance_info:.0f}m in game {game_id}")
+            return Image(data=image_bytes, format="jpeg")
+        except Exception as e:
+            print(f"Error processing move Street View image: {e}")
+            raise Exception(f"Failed to process move Street View image: {str(e)}")
+    else:
+        raise Exception("No Street View image received from the move")
+
+@mcp.tool()
+def make_placeholder_guess(lat: float, lng: float) -> dict:
+    """
+    Records a temporary guess for the location (stored locally until final guess).
+    Args:
+        lat (float): The latitude of the guess.
+        lng (float): The longitude of the guess.
+    Returns:
+        dict: A status message.
+    """
+    global active_game
+    if not active_game or not active_game.get('game_id'):
+        raise ValueError("Game not started.")
+    if active_game.get('game_over'):
+        raise ValueError("Game is over.")
+
+    active_game['placeholder_guess'] = {'lat': lat, 'lng': lng}
+    return {"status": "success", "message": f"Placeholder guess recorded: {lat:.6f}, {lng:.6f}"}
+
+
+@mcp.tool()
+def make_final_guess() -> dict:
+    """
+    Makes the final guess for the active game by calling the Flask API.
+    Uses the stored placeholder guess coordinates.
+    Returns:
+        dict: The results of the guess including distance, score, and actual location.
+    """
+    global active_game
+    if not active_game or not active_game.get('game_id'):
+        raise ValueError("Game not started.")
+    if active_game.get('game_over'):
+        raise ValueError("Game is already over.")
+    if 'placeholder_guess' not in active_game:
+        raise ValueError("No placeholder guess was made. Call make_placeholder_guess() first.")
+
+    # Get the placeholder guess
+    guess_location = active_game['placeholder_guess']
+    
+    # Call Flask API to make the final guess
+    game_id = active_game['game_id']
+    guess_result = call_flask_api(f'/game/{game_id}/guess', 'POST', {
+        'lat': guess_location['lat'],
+        'lng': guess_location['lng']
+    })
+
+    # Mark game as over
+    active_game['game_over'] = True
+    
+    return {
+        "distance_km": guess_result['distance_km'],
+        "score": guess_result['score'],
+        "actual_location": guess_result['actual_location'],
+        "guess_location": guess_result['guess_location']
+    }
+
+@mcp.tool()
+def get_game_state() -> dict:
+    """
+    Gets the current game state from the Flask API.
+    Returns:
+        dict: Current game state including moves, actions, and game status.
+    """
+    global active_game
+    if not active_game or not active_game.get('game_id'):
+        raise ValueError("Game not started.")
+    
+    game_id = active_game['game_id']
+    game_state = call_flask_api(f'/game/{game_id}/state')
+    
+    # Don't expose the actual coordinates - keep the guessing challenge
+    state_info = {
+        "game_id": game_state.get('game_id', game_id),
+        "player_name": game_state.get('player_name', active_game.get('player_name')),
+        "moves": game_state.get('moves', 0),
+        "game_over": game_state.get('game_over', False),
+        "total_actions": len(game_state.get('actions', [])),
+        "guesses_made": len(game_state.get('guesses', [])),
+        "placeholder_guess": active_game.get('placeholder_guess')
+    }
+    
+    # Update local game state
+    active_game['game_over'] = game_state.get('game_over', False)
+    
+    return state_info
+
+@mcp.tool()
+def test_connection() -> str:
+    """
+    Simple test to verify MCP server is working.
+    Returns:
+        str: A test message.
+    """
+    return "MCP server is working correctly!"
+
+@mcp.get("/health")
+def health_check():
+    """Health check endpoint for Docker"""
+    return {"status": "healthy", "flask_api_url": FLASK_API_URL}
+
+@mcp.tool()
+def check_flask_api() -> dict:
+    """
+    Checks if the Flask API is available and responding.
+    Returns:
+        dict: Status of the Flask API connection.
+    """
+    try:
+        response = requests.get(f"{FLASK_API_URL}/", timeout=5)
+        if response.status_code == 200:
+            return {
+                "status": "connected",
+                "message": f"Flask API is available at {FLASK_API_URL}",
+                "flask_status_code": response.status_code
+            }
+        else:
+            return {
+                "status": "error",
+                "message": f"Flask API returned status code {response.status_code}",
+                "flask_status_code": response.status_code
+            }
+    except requests.exceptions.ConnectionError:
+        return {
+            "status": "disconnected",
+            "message": f"Cannot connect to Flask API at {FLASK_API_URL}. Make sure app.py is running.",
+            "flask_status_code": None
+        }
+    except Exception as e:
+        return {
+            "status": "error",
+            "message": f"Error checking Flask API: {str(e)}",
+            "flask_status_code": None
+        }
+
+# --- Server Execution ---
+if __name__ == "__main__":
+    print(f"GeoGuessr MCP Server starting...")
+    print(f"Flask API URL: {FLASK_API_URL}")
+    print("Make sure the Flask server (app.py) is running before using this MCP server.")
+    print("Running GeoGuessr MCP server with SSE transport")
+    mcp.run(transport="sse")