diff --git a/src/geoguessr_mcp/tools/__init__.py b/src/geoguessr_mcp/tools/__init__.py index b891c79..e9e0917 100644 --- a/src/geoguessr_mcp/tools/__init__.py +++ b/src/geoguessr_mcp/tools/__init__.py @@ -1,33 +1,57 @@ -"""Register all MCP tools.""" +"""MCP Tools registration module.""" from mcp.server.fastmcp import FastMCP -from ..api.client import GeoGuessrClient +from .analysis_tools import register_analysis_tools +from .auth_tools import register_auth_tools +from .game_tools import register_game_tools +from .monitoring_tools import register_monitoring_tools +from .profile_tools import register_profile_tools +from ..api.geoguessr_client import GeoGuessrClient from ..auth.session import SessionManager +from ..config import settings from ..services.analysis_service import AnalysisService from ..services.game_service import GameService from ..services.profile_service import ProfileService -# from .analysis_tools import register_analysis_tools -# from .auth_tools import register_auth_tools -# from .game_tools import register_game_tools -# from .profile_tools import register_profile_tools +def register_all_tools(mcp: FastMCP) -> dict: + """ + Register all MCP tools with the server. - -def register_all_tools(mcp: FastMCP): - """Register all tools with the MCP server.""" - # Initialize dependencies - session_manager = SessionManager() + Returns: + Dictionary with initialized services for potential reuse + """ + # Initialize core dependencies + session_manager = SessionManager(default_cookie=settings.DEFAULT_NCFA_COOKIE) client = GeoGuessrClient(session_manager) # Initialize services profile_service = ProfileService(client) game_service = GameService(client) - analysis_service = AnalysisService(client) + analysis_service = AnalysisService(client, game_service, profile_service) - # Register tools - # register_auth_tools(mcp, session_manager) - # register_profile_tools(mcp, profile_service) - # register_game_tools(mcp, game_service) - # register_analysis_tools(mcp, analysis_service, game_service) + # Register all tool groups + register_auth_tools(mcp, session_manager) + register_profile_tools(mcp, profile_service) + register_game_tools(mcp, game_service) + register_analysis_tools(mcp, analysis_service) + register_monitoring_tools(mcp) + + return { + "session_manager": session_manager, + "client": client, + "profile_service": profile_service, + "game_service": game_service, + "analysis_service": analysis_service, + } + + +__all__ = [ + "register_all_tools", + "register_auth_tools", + "register_profile_tools", + "register_game_tools", + "register_analysis_tools", + "register_monitoring_tools", +] diff --git a/src/geoguessr_mcp/tools/analysis_tools.py b/src/geoguessr_mcp/tools/analysis_tools.py index 4640904..9482810 100644 --- a/src/geoguessr_mcp/tools/analysis_tools.py +++ b/src/geoguessr_mcp/tools/analysis_tools.py @@ -1 +1,69 @@ -# TODO +""" +This module provides tools for analyzing and improving game performance +by registering multiple analysis-related functions to a given `FastMCP` +instance. These tools include functionalities for analyzing recent games, +retrieving performance summaries, and generating strategy recommendations. + +The functions leverage an external analysis service to compute detailed +statistics and insights based on gameplay data and user profiles. Each tool +offers asynchronous execution for efficient performance. +""" + +from mcp.server.fastmcp import FastMCP + +from .auth_tools import get_current_session_token +from ..services.analysis_service import AnalysisService + + +def register_analysis_tools(mcp: FastMCP, analysis_service: AnalysisService): + """Register analysis-related tools.""" + + @mcp.tool() + async def analyze_recent_games(count: int = 10) -> dict: + """ + Analyze recent games and provide statistics summary. + + Fetches recent games and calculates aggregate statistics including + average scores, perfect round rates, and performance trends. + + Args: + count: Number of recent games to analyze (default: 10) + + Returns: + Comprehensive analysis with statistics and individual game data + """ + session_token = get_current_session_token() + return await analysis_service.analyze_recent_games(count, session_token) + + @mcp.tool() + async def get_performance_summary() -> dict: + """ + Get a comprehensive performance summary. + + Combines profile stats, achievements, season information, and + recent game analysis into a single overview. Useful for understanding + overall account status and progress. + + Returns: + Aggregated performance data from multiple API endpoints + """ + session_token = get_current_session_token() + return await analysis_service.get_performance_summary(session_token) + + @mcp.tool() + async def get_strategy_recommendations() -> dict: + """ + Get personalized strategy recommendations. + + Analyzes gameplay patterns and provides actionable recommendations + for improving performance. Considers factors like: + - Perfect round rate + - Time management + - Score trends + - Weak areas + + Returns: + Analysis summary and prioritized recommendations + """ + session_token = get_current_session_token() + return await analysis_service.get_strategy_recommendations(session_token) diff --git a/src/geoguessr_mcp/tools/auth_tools.py b/src/geoguessr_mcp/tools/auth_tools.py index 4640904..0c47757 100644 --- a/src/geoguessr_mcp/tools/auth_tools.py +++ b/src/geoguessr_mcp/tools/auth_tools.py @@ -1 +1,180 @@ -# TODO +""" +Tools for handling authentication and session management with GeoGuessr. + +This module provides utilities for login, logout, setting authentication cookies, +and checking the current authentication status. Its primary purpose is to ensure +interactions with the GeoGuessr API are authenticated securely and conveniently. + +The module integrates with FastMCP to expose authentication methods as tools +and uses the `SessionManager` for session storage and validation. + +Functions +--------- + +- register_auth_tools: Registers a set of authentication tools with FastMCP. +- get_current_session_token: Returns the currently active session token. + +""" + +import logging +import secrets +from datetime import datetime, timedelta, UTC +from typing import Optional + +from mcp.server.fastmcp import FastMCP + +from ..auth.session import SessionManager, UserSession +from ..config import settings + +logger = logging.getLogger(__name__) + +# Global session token storage +_current_session_token: Optional[str] = None + + +def register_auth_tools(mcp: FastMCP, session_manager: SessionManager): + """Register authentication-related tools.""" + + @mcp.tool() + async def login(email: str, password: str) -> dict: + """ + Authenticate with GeoGuessr using email and password. + + Creates a session that will be used for all later API calls. + Credentials are only used to get an authentication token and are + not stored on the server. + + Args: + email: Your GeoGuessr account email + password: Your GeoGuessr account password + + Returns: + Session information including username and session token + """ + global _current_session_token + + try: + session_token, session = await session_manager.login(email, password) + _current_session_token = session_token + + return { + "success": True, + "message": f"Successfully logged in as {session.username}", + "username": session.username, + "user_id": session.user_id, + "session_token": session_token, + "expires_at": session.expires_at.isoformat() if session.expires_at else None, + } + except ValueError as e: + return {"success": False, "error": str(e)} + except Exception as e: + logger.error(f"Login error: {e}") + return {"success": False, "error": f"An unexpected error occurred: {str(e)}"} + + @mcp.tool() + async def logout() -> dict: + """ + Logout from the current GeoGuessr session. + + Invalidates the current session token. + """ + global _current_session_token + + if _current_session_token: + success = await session_manager.logout(_current_session_token) + _current_session_token = None + return { + "success": success, + "message": "Successfully logged out" if success else "No active session found", + } + + return {"success": False, "message": "No active session"} + + @mcp.tool() + async def set_ncfa_cookie(cookie: str) -> dict: + """ + Set the _ncfa cookie for authentication. + + Use this if you've manually extracted the cookie from your browser. + The cookie will be validated before being accepted. + + Args: + cookie: The _ncfa cookie value from your browser + """ + global _current_session_token + + # Validate the cookie + profile = await session_manager.validate_cookie(cookie) + + if not profile: + return {"success": False, "error": "Invalid cookie - authentication failed"} + + # Create a session from the cookie + session = UserSession( + ncfa_cookie=cookie, + user_id=profile.get("id", ""), + username=profile.get("nick", ""), + email="manual@cookie", + expires_at=datetime.now(UTC) + timedelta(days=30), + ) + + # Store as a session + session_token = secrets.token_urlsafe(32) + async with session_manager._lock: + session_manager._sessions[session_token] = session + session_manager._user_sessions[session.user_id] = session_token + + _current_session_token = session_token + + return { + "success": True, + "message": f"Cookie set successfully. Authenticated as {session.username}", + "username": session.username, + "user_id": session.user_id, + "session_token": session_token, + } + + @mcp.tool() + async def get_auth_status() -> dict: + """ + Check the current authentication status. + + Returns information about the current session or available + authentication methods. + """ + global _current_session_token + + # Check for active session + if _current_session_token: + session = await session_manager.get_session(_current_session_token) + if session and session.is_valid(): + return { + "authenticated": True, + "method": "session", + "username": session.username, + "user_id": session.user_id, + "expires_at": session.expires_at.isoformat() if session.expires_at else None, + } + + # Check for environment variable + env_cookie = settings.DEFAULT_NCFA_COOKIE + if env_cookie: + profile = await session_manager.validate_cookie(env_cookie) + if profile: + return { + "authenticated": True, + "method": "environment_variable", + "username": profile.get("nick", "Unknown"), + "user_id": profile.get("id", "Unknown"), + } + + return { + "authenticated": False, + "message": "Not authenticated. Use 'login' with credentials or 'set_ncfa_cookie' with a valid cookie.", + "available_methods": ["login(email, password)", "set_ncfa_cookie(cookie)"], + } + + +def get_current_session_token() -> Optional[str]: + """Get the current session token for use by other tools.""" + return _current_session_token diff --git a/src/geoguessr_mcp/tools/game_tools.py b/src/geoguessr_mcp/tools/game_tools.py index 4640904..f567e39 100644 --- a/src/geoguessr_mcp/tools/game_tools.py +++ b/src/geoguessr_mcp/tools/game_tools.py @@ -1 +1,240 @@ -# TODO +""" +Provide functionality to register game-related tools with the FastMCP server. + +The module defines tools that can query various game-related details, including +specific game information, activity feeds, recent games, unfinished games, +season statistics, and details about game modes like daily challenges, +Battle Royale, duels, and tournaments. + +Functions: + register_game_tools: Registers multiple tools to interact with the + game service and retrieve or manage game-related data. +""" + +from mcp.server.fastmcp import FastMCP + +from .auth_tools import get_current_session_token +from ..services.game_service import GameService + + +def register_game_tools(mcp: FastMCP, game_service: GameService): + """Register game-related tools.""" + + @mcp.tool() + async def get_game_details(game_token: str) -> dict: + """ + Get detailed information about a specific game. + + Args: + game_token: The game's token/ID + + Returns: + Detailed game information including all rounds and scores + """ + session_token = get_current_session_token() + game, response = await game_service.get_game_details(game_token, session_token) + + return { + "game": game.to_dict(), + "available_fields": response.available_fields, + "raw_summary": response.summarize(max_depth=2), + } + + @mcp.tool() + async def get_activity_feed(count: int = 10, page: int = 0) -> dict: + """ + Get the user's activity feed. + + Shows recent games, achievements, and other activities. + + Args: + count: Number of items to fetch (default: 10) + page: Page number for pagination (default: 0) + + Returns: + Activity feed entries with dynamic schema information + """ + session_token = get_current_session_token() + response = await game_service.get_activity_feed(count, page, session_token) + + if not response.is_success: + return {"success": False, "error": str(response.data)} + + # Extract and categorize entries + entries = response.data.get("entries", []) + categorized = {} + + for entry in entries: + entry_type = entry.get("type", "unknown") + if entry_type not in categorized: + categorized[entry_type] = [] + categorized[entry_type].append(entry) + + return { + "success": True, + "total_entries": len(entries), + "entry_types": list(categorized.keys()), + "entries_by_type": { + t: len(e) for t, e in categorized.items() + }, + "recent_entries": entries[:5], # First 5 for context + "available_fields": response.available_fields, + } + + @mcp.tool() + async def get_recent_games(count: int = 10) -> dict: + """ + Get recent games with full details. + + Args: + count: Number of games to retrieve (default: 10) + + Returns: + List of recent games with scores and round details + """ + session_token = get_current_session_token() + games = await game_service.get_recent_games(count, session_token) + + return { + "games_found": len(games), + "games": [g.to_dict() for g in games], + "summary": { + "total_score": sum(g.total_score for g in games), + "average_score": sum(g.total_score for g in games) / len(games) if games else 0, + "maps_played": list(set(g.map_name for g in games)), + }, + } + + @mcp.tool() + async def get_unfinished_games() -> dict: + """ + Get list of games that haven't been completed. + + Returns: + List of unfinished games that can be resumed + """ + session_token = get_current_session_token() + response = await game_service.get_unfinished_games(session_token) + + return { + "success": response.is_success, + "data": response.data if response.is_success else None, + "available_fields": response.available_fields, + } + + @mcp.tool() + async def get_season_stats() -> dict: + """ + Get current competitive season statistics. + + Returns: + Season ranking, rating, games played, and division info + """ + session_token = get_current_session_token() + + try: + stats, response = await game_service.get_season_stats(session_token) + + return { + "success": True, + "season_stats": { + "rank": stats.rank, + "rating": stats.rating, + "games_played": stats.games_played, + "wins": stats.wins, + "division": stats.division, + }, + "available_fields": response.available_fields, + "raw_summary": response.summarize(), + } + except Exception as e: + return {"success": False, "error": str(e)} + + @mcp.tool() + async def get_daily_challenge(day: str = "today") -> dict: + """ + Get information about the daily challenge. + + Args: + day: "today", "yesterday", or a specific date + + Returns: + Daily challenge details including map and time limit + """ + session_token = get_current_session_token() + + try: + challenge, response = await game_service.get_daily_challenge(day, session_token) + + return { + "success": True, + "challenge": { + "token": challenge.token, + "map": challenge.map_name, + "date": challenge.date, + "time_limit": challenge.time_limit, + "completed": challenge.completed, + "score": challenge.score, + }, + "available_fields": response.available_fields, + } + except Exception as e: + return {"success": False, "error": str(e)} + + @mcp.tool() + async def get_battle_royale(game_id: str) -> dict: + """ + Get Battle Royale game details. + + Args: + game_id: The Battle Royale game ID + + Returns: + Game details including players and standings + """ + session_token = get_current_session_token() + response = await game_service.get_battle_royale(game_id, session_token) + + return { + "success": response.is_success, + "data": response.summarize() if response.is_success else None, + "available_fields": response.available_fields, + "schema_description": response.schema_description, + } + + @mcp.tool() + async def get_duel(duel_id: str) -> dict: + """ + Get Duel game details. + + Args: + duel_id: The Duel game ID + + Returns: + Duel details including opponent and results + """ + session_token = get_current_session_token() + response = await game_service.get_duel(duel_id, session_token) + + return { + "success": response.is_success, + "data": response.summarize() if response.is_success else None, + "available_fields": response.available_fields, + } + + @mcp.tool() + async def get_tournaments() -> dict: + """ + Get tournament information. + + Returns: + Available tournaments and their details + """ + session_token = get_current_session_token() + response = await game_service.get_tournaments(session_token) + + return { + "success": response.is_success, + "data": response.summarize() if response.is_success else None, + "available_fields": response.available_fields, + } diff --git a/src/geoguessr_mcp/tools/monitoring_tools.py b/src/geoguessr_mcp/tools/monitoring_tools.py new file mode 100644 index 0000000..dec84c6 --- /dev/null +++ b/src/geoguessr_mcp/tools/monitoring_tools.py @@ -0,0 +1,197 @@ +""" +Module to register monitoring tools for analyzing API endpoints and schemas. + +This module provides a set of monitoring tools that allow checking the status +and availability of API endpoints, exploring their response schemas, and +tracking changes to API structures. These tools are integrated with the FastMCP +framework and can be used to comprehensively monitor API performance and +evolution. +""" + +from mcp.server.fastmcp import FastMCP + +from .auth_tools import get_current_session_token +from ..monitoring import endpoint_monitor, schema_registry + + +def register_monitoring_tools(mcp: FastMCP): + """Register monitoring-related tools.""" + + @mcp.tool() + async def check_api_status() -> dict: + """ + Check the availability and status of all monitored API endpoints. + + Runs a full check of all known GeoGuessr API endpoints and reports + their availability, response times, and any schema changes detected. + + Returns: + Comprehensive API status report including + - Number of available/unavailable endpoints + - Response times + - Recent schema changes + - Error details for failed endpoints + """ + # Update monitor with current auth + session_token = get_current_session_token() + if session_token: + from ..auth.session import SessionManager + session_manager = SessionManager() + session = await session_manager.get_session(session_token) + if session: + endpoint_monitor.ncfa_cookie = session.ncfa_cookie + + await endpoint_monitor.run_full_check() + return endpoint_monitor.get_monitoring_report() + + @mcp.tool() + async def get_endpoint_schema(endpoint: str) -> dict: + """ + Get the current schema information for a specific API endpoint. + + Provides detailed information about the response format of an endpoint, + including all fields, their types, and example values. This is useful + for understanding what data is available from each endpoint. + + Args: + endpoint: The API endpoint path (e.g., "/v3/profiles") + + Returns: + Schema information including + - Field names and types + - Whether the endpoint is currently available + - When the schema was last updated + - Sample response structure + """ + schema = schema_registry.get_schema(endpoint) + + if not schema: + return { + "found": False, + "message": f"No schema information available for {endpoint}", + "available_endpoints": schema_registry.get_available_endpoints(), + } + + return { + "found": True, + "endpoint": schema.endpoint, + "method": schema.method, + "is_available": schema.is_available, + "last_updated": schema.last_updated.isoformat(), + "response_code": schema.response_code, + "field_count": len(schema.fields), + "fields": { + name: { + "type": field.field_type, + "nullable": field.nullable, + "has_nested": field.nested_schema is not None, + } + for name, field in list(schema.fields.items())[:30] # Limit for context + }, + "error_message": schema.error_message, + } + + @mcp.tool() + async def list_available_endpoints() -> dict: + """ + List all known API endpoints and their current status. + + Returns a summary of all monitored endpoints including which ones + are currently available and when they were last checked. + + Returns: + List of endpoints with availability status + """ + summary = schema_registry.get_schema_summary() + + return { + "total_endpoints": summary["total_endpoints"], + "available_count": summary["available_endpoints"], + "endpoints": { + ep: { + "available": info["available"], + "field_count": info["field_count"], + "last_updated": info["last_updated"], + } + for ep, info in summary["endpoints"].items() + }, + } + + @mcp.tool() + async def get_schema_changes() -> dict: + """ + Get information about recent schema changes detected in the API. + + Shows endpoints where the response format has changed since the + last check. This is useful for understanding API evolution. + + Returns: + List of endpoints with schema changes and change details + """ + changes = [] + + for endpoint, history in schema_registry.schema_history.items(): + if len(history) > 0: + current = schema_registry.get_schema(endpoint) + previous = history[-1] if history else None + + if current and previous: + changes.append({ + "endpoint": endpoint, + "current_hash": current.schema_hash, + "previous_hash": previous.schema_hash, + "current_fields": len(current.fields), + "previous_fields": len(previous.fields), + "changed_at": current.last_updated.isoformat(), + }) + + return { + "total_changes_tracked": len(changes), + "changes": changes, + } + + @mcp.tool() + async def explore_endpoint(path: str, use_game_server: bool = False) -> dict: + """ + Explore an unknown or new API endpoint. + + Makes a request to the specified endpoint and analyzes the response + to discover its schema. Useful for discovering new endpoints or + testing endpoint availability. + + Args: + path: The API endpoint path to explore (e.g., "/v3/new-endpoint") + use_game_server: Whether to use the game server URL instead of main API + + Returns: + Response analysis including discovered schema and sample data + """ + session_token = get_current_session_token() + + from ..api.dynamic_response import GeoGuessrClient + from ..auth.session import SessionManager + + session_manager = SessionManager() + client = GeoGuessrClient(session_manager) + + try: + response = await client.get_raw( + path, + session_token, + use_game_server=use_game_server, + ) + + return { + "success": response.is_success, + "status_code": response.status_code, + "response_time_ms": round(response.response_time_ms, 2), + "discovered_fields": response.available_fields, + "schema_description": response.schema_description, + "data_preview": response.summarize(max_depth=2), + } + except Exception as e: + return { + "success": False, + "error": str(e), + "path": path, + } diff --git a/src/geoguessr_mcp/tools/profile_tools.py b/src/geoguessr_mcp/tools/profile_tools.py index 4640904..588aaa2 100644 --- a/src/geoguessr_mcp/tools/profile_tools.py +++ b/src/geoguessr_mcp/tools/profile_tools.py @@ -1 +1,146 @@ -# TODO +""" +This module defines the `register_profile_tools` function, which registers a +suite of profile-related functionalities as tools in a FastMCP application. + +It includes tools for retrieving user profile data, game statistics, achievements, +and other related functionality. These tools are dynamically registered with an +instance of FastMCP, allowing the client to interact with profile and game meta-data. + +The tools are designed to handle asynchronous operations and adapt to changes +from the underlying service API. Tools return structured data for easy consumption. +""" + +from mcp.server.fastmcp import FastMCP + +from .auth_tools import get_current_session_token +from ..services.profile_service import ProfileService + + +def register_profile_tools(mcp: FastMCP, profile_service: ProfileService): + """Register profile-related tools.""" + + @mcp.tool() + async def get_my_profile() -> dict: + """ + Get the current user's profile information. + + Returns profile data including username, level, country, and more. + The response format adapts to API changes automatically. + """ + session_token = get_current_session_token() + profile, response = await profile_service.get_profile(session_token) + + return { + "profile": profile.to_dict(), + "available_fields": response.available_fields, + "raw_data_preview": response.summarize(max_depth=1), + } + + @mcp.tool() + async def get_my_stats() -> dict: + """ + Get the current user's game statistics. + + Returns statistics like games played, average score, win rate, etc. + """ + session_token = get_current_session_token() + stats, response = await profile_service.get_stats(session_token) + + return { + "stats": stats.to_dict(), + "available_fields": response.available_fields, + } + + @mcp.tool() + async def get_extended_stats() -> dict: + """ + Get extended statistics not shown on the profile page. + + Returns additional metrics and detailed breakdowns. + Response format is dynamic - check available_fields for current structure. + """ + session_token = get_current_session_token() + response = await profile_service.get_extended_stats(session_token) + + return { + "data": response.data if response.is_success else None, + "success": response.is_success, + "available_fields": response.available_fields, + "schema_description": response.schema_description, + } + + @mcp.tool() + async def get_achievements() -> dict: + """ + Get all achievements for the current user. + + Returns list of achievements with unlocked status and progress. + """ + session_token = get_current_session_token() + achievements, response = await profile_service.get_achievements(session_token) + + unlocked = [a for a in achievements if a.unlocked] + locked = [a for a in achievements if not a.unlocked] + + return { + "summary": { + "total": len(achievements), + "unlocked": len(unlocked), + "locked": len(locked), + "completion_rate": f"{len(unlocked) / len(achievements) * 100:.1f}%" if achievements else "0%", + }, + "unlocked_achievements": [ + {"name": a.name, "description": a.description, "unlocked_at": a.unlocked_at} + for a in unlocked[:20] # Limit for context + ], + "locked_achievements": [ + {"name": a.name, "description": a.description, "progress": a.progress} + for a in locked[:10] # Show some locked ones + ], + } + + @mcp.tool() + async def get_comprehensive_profile() -> dict: + """ + Get a comprehensive profile summary combining multiple data sources. + + Aggregates profile, stats, achievements, and more into a single response. + Useful for getting a complete overview of the user's account. + """ + session_token = get_current_session_token() + return await profile_service.get_comprehensive_profile(session_token) + + @mcp.tool() + async def get_user_maps() -> dict: + """ + Get maps created by the current user. + + Returns list of custom maps with their details. + """ + session_token = get_current_session_token() + response = await profile_service.get_user_maps(session_token) + + return { + "success": response.is_success, + "data": response.summarize() if response.is_success else None, + "available_fields": response.available_fields, + } + + @mcp.tool() + async def get_public_profile(user_id: str) -> dict: + """ + Get another user's public profile. + + Args: + user_id: The user's ID to look up + + Returns: + Public profile information for the specified user + """ + session_token = get_current_session_token() + profile, response = await profile_service.get_public_profile(user_id, session_token) + + return { + "profile": profile.to_dict(), + "available_fields": response.available_fields, + }