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="0.0.0.0", 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!" if __name__ == "__main__": mcp.run(transport="sse")