Add tools for MCP module: register comprehensive toolsets (auth, profile, game, analysis, monitoring) with enhanced functionality and asynchronous operations. Integrate session handling, API schema analysis, and performance insights.

This commit is contained in:
Yûki VACHOT 2025-11-29 02:27:46 +01:00
parent 1b7963c239
commit 126d04ab0f
6 changed files with 873 additions and 21 deletions

View file

@ -1,33 +1,57 @@
"""Register all MCP tools.""" """MCP Tools registration module."""
from mcp.server.fastmcp import FastMCP 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 ..auth.session import SessionManager
from ..config import settings
from ..services.analysis_service import AnalysisService from ..services.analysis_service import AnalysisService
from ..services.game_service import GameService from ..services.game_service import GameService
from ..services.profile_service import ProfileService from ..services.profile_service import ProfileService
# from .analysis_tools import register_analysis_tools def register_all_tools(mcp: FastMCP) -> dict:
# from .auth_tools import register_auth_tools """
# from .game_tools import register_game_tools Register all MCP tools with the server.
# from .profile_tools import register_profile_tools
Returns:
def register_all_tools(mcp: FastMCP): Dictionary with initialized services for potential reuse
"""Register all tools with the MCP server.""" """
# Initialize dependencies # Initialize core dependencies
session_manager = SessionManager() session_manager = SessionManager(default_cookie=settings.DEFAULT_NCFA_COOKIE)
client = GeoGuessrClient(session_manager) client = GeoGuessrClient(session_manager)
# Initialize services # Initialize services
profile_service = ProfileService(client) profile_service = ProfileService(client)
game_service = GameService(client) game_service = GameService(client)
analysis_service = AnalysisService(client) analysis_service = AnalysisService(client, game_service, profile_service)
# Register tools # Register all tool groups
# register_auth_tools(mcp, session_manager) register_auth_tools(mcp, session_manager)
# register_profile_tools(mcp, profile_service) register_profile_tools(mcp, profile_service)
# register_game_tools(mcp, game_service) register_game_tools(mcp, game_service)
# register_analysis_tools(mcp, analysis_service, 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",
]

View file

@ -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)

View file

@ -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

View file

@ -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,
}

View file

@ -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,
}

View file

@ -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,
}