yvachot/GeoGuessrMCP
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor tools module: removed unused tools, commented out legacy registrations, updated mock data in tests, consolidated imports, and standardized naming conventions in the codebase.

Browse source
This commit is contained in:
Yûki VACHOT 2025-11-29 01:24:58 +01:00
parent ec0fe38861
commit 328e597f48
16 changed files with 86 additions and 386 deletions
Download patch file Download diff file Expand all files Collapse all files

4
src/geoguessr_mcp/api/client.py
View file

@ -10,10 +10,10 @@ from typing import Any, Optional
import httpx
from .endpoints import EndpointInfo
from ..auth.session import SessionManager
from ..config import settings
from ..monitoring.schema_manager import schema_registry
from .endpoints import EndpointInfo
from ..monitoring.schema.SchemaRegistry import schema_registry
logger = logging.getLogger(__name__)

9
src/geoguessr_mcp/models/__init__.py
View file

@ -1,11 +1,10 @@
"""Data models for GeoGuessr."""
from Achievement import Achievement
from DailyChallenge import DailyChallenge
from Game import Game
from SeasonStats import SeasonStats
from .Achievement import Achievement
from .DailyChallenge import DailyChallenge
from .Game import Game
from .RoundGuess import RoundGuess
from .SeasonStats import SeasonStats
from .UserProfile import UserProfile
from .UserStats import UserStats

7
src/geoguessr_mcp/monitoring/__init__.py
View file

@ -1,11 +1,10 @@
"""Monitoring module for API endpoint tracking and schema detection."""
from schema.EndpointSchema import EndpointSchema
from schema.SchemaDetector import SchemaDetector, SchemaField
from schema.SchemaRegistry import SchemaRegistry, schema_registry
from .endpoint.EndpointMonitor import (MONITORED_ENDPOINTS, EndpointMonitor,
endpoint_monitor)
from .schema.EndpointSchema import EndpointSchema
from .schema.SchemaDetector import SchemaDetector, SchemaField
from .schema.SchemaRegistry import SchemaRegistry, schema_registry
__all__ = [
"EndpointMonitor",

6
src/geoguessr_mcp/services/analysis_service.py
View file

@ -9,11 +9,11 @@ import logging
from dataclasses import dataclass, field
from typing import Optional
from ..api.client import GeoGuessrClient
from ..models.Game import Game
from ..monitoring.schema_manager import schema_registry
from .game_service import GameService
from .profile_service import ProfileService
from ..api.client import GeoGuessrClient
from ..models.Game import Game
from ..monitoring.schema.SchemaRegistry import schema_registry
logger = logging.getLogger(__name__)

24
src/geoguessr_mcp/tools/__init__.py
View file

@ -2,30 +2,32 @@
from mcp.server.fastmcp import FastMCP
from ..api.client import GeoguessrClient
from ..api.client import GeoGuessrClient
from ..auth.session import SessionManager
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
# 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):
"""Register all tools with the MCP server."""
# Initialize dependencies
session_manager = SessionManager()
client = GeoguessrClient(session_manager)
client = GeoGuessrClient(session_manager)
# Initialize services
profile_service = ProfileService(client)
game_service = GameService(client)
analysis_service = AnalysisService()
analysis_service = AnalysisService(client)
# 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_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)

131
src/geoguessr_mcp/tools/analysis_tools.py
View file

@ -1,130 +1 @@
@mcp.tool()
async def analyze_recent_games(count: int = 10) -> dict:
"""
Analyze recent games and provide statistics summary.
Fetches recent games from the activity feed and calculates aggregate statistics.
Args:
count: Number of recent games to analyze (default: 10)
"""
async with await get_async_session() as client:
# Get activity feed
feed_response = await client.get(
f"{GEOGUESSR_BASE_URL}/v4/feed/private", params={"count": count * 2, "page": 0}
)
feed_response.raise_for_status()
feed = feed_response.json()
games_analyzed = []
total_score = 0
total_rounds = 0
perfect_rounds = 0
for entry in feed.get("entries", []):
if entry.get("type") == "PlayedGame" and len(games_analyzed) < count:
game_token = entry.get("payload", {}).get("gameToken")
if game_token:
try:
game_response = await client.get(
f"{GEOGUESSR_BASE_URL}/v3/games/{game_token}"
)
if game_response.status_code == 200:
game = game_response.json()
game_info = {
"token": game_token,
"map": game.get("map", {}).get("name", "Unknown"),
"mode": game.get("type", "Unknown"),
"total_score": 0,
"rounds": [],
}
for round_data in game.get("player", {}).get("guesses", []):
round_score = round_data.get("roundScoreInPoints", 0)
game_info["total_score"] += round_score
game_info["rounds"].append(
{
"score": round_score,
"distance": round_data.get("distanceInMeters", 0),
"time": round_data.get("time", 0),
}
)
total_rounds += 1
if round_score == 5000:
perfect_rounds += 1
total_score += game_info["total_score"]
games_analyzed.append(game_info)
except Exception as e:
logger.warning(f"Failed to fetch game {game_token}: {e}")
return {
"games_analyzed": len(games_analyzed),
"total_score": total_score,
"average_score": total_score / len(games_analyzed) if games_analyzed else 0,
"total_rounds": total_rounds,
"perfect_rounds": perfect_rounds,
"perfect_round_percentage": (
(perfect_rounds / total_rounds * 100) if total_rounds > 0 else 0
),
"games": games_analyzed,
}
@mcp.tool()
async def get_performance_summary() -> dict:
"""
Get a comprehensive performance summary combining profile stats,
achievements, and season information.
"""
async with await get_async_session() as client:
results = {}
# Get profile
try:
profile_response = await client.get(f"{GEOGUESSR_BASE_URL}/v3/profiles")
profile_response.raise_for_status()
results["profile"] = profile_response.json()
except Exception as e:
results["profile_error"] = str(e)
# Get stats
try:
stats_response = await client.get(f"{GEOGUESSR_BASE_URL}/v3/profiles/stats")
stats_response.raise_for_status()
results["stats"] = stats_response.json()
except Exception as e:
results["stats_error"] = str(e)
# Get extended stats
try:
extended_response = await client.get(f"{GEOGUESSR_BASE_URL}/v4/stats/me")
extended_response.raise_for_status()
results["extended_stats"] = extended_response.json()
except Exception as e:
results["extended_stats_error"] = str(e)
# Get season stats
try:
season_response = await client.get(f"{GEOGUESSR_BASE_URL}/v4/seasons/active/stats")
season_response.raise_for_status()
results["current_season"] = season_response.json()
except Exception as e:
results["season_error"] = str(e)
# Get achievements
try:
achievements_response = await client.get(
f"{GEOGUESSR_BASE_URL}/v3/profiles/achievements"
)
achievements_response.raise_for_status()
achievements = achievements_response.json()
results["achievements_summary"] = {
"total": len(achievements) if isinstance(achievements, list) else 0,
"achievements": achievements,
}
except Exception as e:
results["achievements_error"] = str(e)
return results
# TODO

181
src/geoguessr_mcp/tools/auth_tools.py
View file

@ -1,180 +1 @@
"""MCP tools for auth operations."""
import logging
from mcp.server.fastmcp import FastMCP
from ..auth.session import SessionManager
logger = logging.getLogger(__name__)
def register_auth_tools(mcp: FastMCP, session_manager: SessionManager):
"""Register auth-related tools."""
@mcp.tool()
async def login(email: str, password: str) -> dict:
"""
Authenticate with GeoGuessr using your email and password.
This creates a session that will be used for all later API calls.
Args:
email: Your GeoGuessr account email
password: Your GeoGuessr account password
Returns:
Session information including username and session token
Note: Your credentials are only used to get an authentication token
from GeoGuessr. They are not stored on the server.
"""
try:
session_token, session = await session_manager.login(email, password)
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.
This 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 to logout",
}
return {"success": False, "message": "No active session"}
@mcp.tool()
async def set_session_token(token: str) -> dict:
"""
Set an existing session token for authentication.
Use this if you have a previously obtained session token.
Args:
token: A valid session token from a previous login
"""
global _current_session_token
session = await session_manager.get_session(token)
if session and session.is_valid():
_current_session_token = token
return {
"success": True,
"message": f"Session set for user {session.username}",
"username": session.username,
}
return {"success": False, "error": "Invalid or expired session token"}
@mcp.tool()
async def set_ncfa_cookie(cookie: str) -> dict:
"""
Directly set the _ncfa cookie for authentication.
Use this if you've manually extracted the cookie from your browser.
Args:
cookie: The _ncfa cookie value from your browser
Note: This sets the cookie as the default for all requests.
"""
global _current_session_token
# Validate the cookie by making a test request
async with httpx.AsyncClient(timeout=30.0) as client:
client.cookies.set("_ncfa", cookie, domain="www.geoguessr.com")
response = await client.get(f"{GEOGUESSR_BASE_URL}/v3/profiles")
if response.status_code != 200:
return {"success": False, "error": "Invalid cookie - authentication failed"}
profile = response.json()
# 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.datetime.now(datetime.UTC) + datetime.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 authentication method.
"""
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 = os.environ.get("GEOGUESSR_NCFA_COOKIE")
if env_cookie:
# Validate the environment cookie
try:
async with httpx.AsyncClient(timeout=30.0) as client:
client.cookies.set("_ncfa", env_cookie, domain="www.geoguessr.com")
response = await client.get(f"{GEOGUESSR_BASE_URL}/v3/profiles")
if response.status_code == 200:
profile = response.json()
return {
"authenticated": True,
"method": "environment_variable",
"username": profile.get("nick", "Unknown"),
"user_id": profile.get("id", "Unknown"),
}
except Exception:
pass
return {
"authenticated": False,
"message": "Not authenticated. Use 'login' with your GeoGuessr credentials or 'set_ncfa_cookie' with a valid cookie.",
}
# TODO

0
src/geoguessr_mcp/tools/competitive_tools.py
View file

1
src/geoguessr_mcp/tools/game_tools.py
View file

@ -0,0 +1 @@
# TODO

27
src/geoguessr_mcp/tools/profile_tools.py
View file

@ -1,26 +1 @@
"""MCP tools for profile operations."""
from mcp.server.fastmcp import FastMCP
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(session_token: str = "") -> dict:
"""Get the current user's profile information."""
profile = await profile_service.get_profile(session_token if session_token else None)
return {
"id": profile.id,
"nick": profile.nick,
"email": profile.email,
"country": profile.country,
"level": profile.level,
}
@mcp.tool()
async def get_my_stats(session_token: str = "") -> dict:
"""Get the current user's statistics."""
return await profile_service.get_stats(session_token if session_token else None)
# TODO

34
src/tests/conftest.py
View file

@ -32,6 +32,7 @@ def mock_profile_data():
"isVerified": True,
"level": 50,
"rating": {"rating": 1500, "deviation": 100},
"isProUser": True,
}
@ -45,7 +46,38 @@ def mock_game_data():
"player": {
"guesses": [
{"roundScoreInPoints": 5000, "distanceInMeters": 0, "time": 10},
{"roundScoreInPoints": 4500, "distanceInMeters": 100, "time": 15},
{"roundScoreInPoints": 4500, "distanceInMeters": 120, "time": 15},
{"roundScoreInPoints": 3800, "distanceInMeters": 100, "time": 20},
{"roundScoreInPoints": 4900, "distanceInMeters": 100, "time": 25},
{"roundScoreInPoints": 5000, "distanceInMeters": 100, "time": 35},
]
},
"state": "finished",
}
@pytest.fixture
def mock_stats_data():
"""Standard user stats response data."""
return {
"games": 100,
"totalRounds": 500,
"score": 2250000,
"perfectGames": 10,
"winRate": 0.65,
"bestStreak": 25,
}
@pytest.fixture
def mock_season_stats_data():
"""Standard season stats response data."""
return {
"id": "season-2024-1",
"name": "Season 1 2024",
"position": 150,
"elo": 1850,
"games": 45,
"wins": 30,
"tier": "Gold",
}

10
src/tests/unit/auth/test_session.py
View file

@ -78,7 +78,7 @@ class TestSessionManager:
"""Tests for SessionManager."""
@pytest.mark.asyncio
async def test_login_success(self, mock_profile_response):
async def test_login_success(self, mock_profile_data):
"""Test successful login flow."""
manager = SessionManager()
@ -101,7 +101,7 @@ class TestSessionManager:
# Mock profile response
profile_response = MagicMock()
profile_response.status_code = 200
profile_response.json.return_value = mock_profile_response
profile_response.json.return_value = mock_profile_data
mock_client.post = AsyncMock(return_value=login_response)
mock_client.get = AsyncMock(return_value=profile_response)
@ -113,7 +113,7 @@ class TestSessionManager:
assert session_token is not None
assert len(session_token) > 0
assert session.ncfa_cookie == "test_ncfa_cookie_value"
assert session.user_id == "test-user-id-123"
assert session.user_id == "test-user-id"
assert session.username == "TestPlayer"
assert session.is_valid()
@ -154,7 +154,7 @@ class TestSessionManager:
await manager.login("test@example.com", "password")
@pytest.mark.asyncio
async def test_logout(self, mock_profile_response):
async def test_logout(self, mock_profile_data):
"""Test logout functionality."""
manager = SessionManager()
@ -174,7 +174,7 @@ class TestSessionManager:
profile_response = MagicMock()
profile_response.status_code = 200
profile_response.json.return_value = mock_profile_response
profile_response.json.return_value = mock_profile_data
mock_client.post = AsyncMock(return_value=login_response)
mock_client.get = AsyncMock(return_value=profile_response)

12
src/tests/unit/models/test_game.py
View file

@ -13,11 +13,11 @@ from geoguessr_mcp.models.RoundGuess import RoundGuess
class TestGame:
"""Tests for Game model."""
def test_from_api_response(self, mock_game_response):
def test_from_api_response(self, mock_game_data):
"""Test creating game from API response."""
game = Game.from_api_response(mock_game_response)
game = Game.from_api_response(mock_game_data)
assert game.token == "ABC123XYZ"
assert game.token == "ABC123"
assert game.map_name == "World"
assert game.mode == "standard"
assert game.finished is True
@ -52,11 +52,11 @@ class TestGame:
assert guess.distance_meters == 150.5
assert guess.time_seconds == 25
def test_to_dict(self, mock_game_response):
def test_to_dict(self, mock_game_data):
"""Test serializing game to dict."""
game = Game.from_api_response(mock_game_response)
game = Game.from_api_response(mock_game_data)
result = game.to_dict()
assert result["token"] == "ABC123XYZ"
assert result["token"] == "ABC123"
assert len(result["rounds"]) == 5
assert result["total_score"] > 0

4
src/tests/unit/models/test_season_stats.py
View file

@ -16,9 +16,9 @@ from geoguessr_mcp.models import SeasonStats
class TestSeasonStats:
"""Tests for SeasonStats model."""
def test_from_api_response(self, mock_season_stats_response):
def test_from_api_response(self, mock_season_stats_data):
"""Test creating season stats from API response."""
stats = SeasonStats.from_api_response(mock_season_stats_response)
stats = SeasonStats.from_api_response(mock_season_stats_data)
assert stats.season_id == "season-2024-1"
assert stats.season_name == "Season 1 2024"

14
src/tests/unit/models/test_user_profile.py
View file

@ -11,14 +11,14 @@ from geoguessr_mcp.models import UserProfile
class TestUserProfile:
"""Tests for UserProfile model."""
def test_from_api_response(self, mock_profile_response):
def test_from_api_response(self, mock_profile_data):
"""Test creating profile from API response."""
profile = UserProfile.from_api_response(mock_profile_response)
profile = UserProfile.from_api_response(mock_profile_data)
assert profile.id == "test-user-id-123"
assert profile.id == "test-user-id"
assert profile.nick == "TestPlayer"
assert profile.email == "test@example.com"
assert profile.country == "US"
assert profile.country == "FR"
assert profile.level == 50
assert profile.is_verified is True
assert profile.is_pro is True
@ -33,11 +33,11 @@ class TestUserProfile:
assert profile.email == ""
assert profile.level == 0
def test_to_dict(self, mock_profile_response):
def test_to_dict(self, mock_profile_data):
"""Test serializing profile to dict."""
profile = UserProfile.from_api_response(mock_profile_response)
profile = UserProfile.from_api_response(mock_profile_data)
result = profile.to_dict()
assert result["id"] == "test-user-id-123"
assert result["id"] == "test-user-id"
assert result["nick"] == "TestPlayer"
assert "raw_data" not in result

8
src/tests/unit/models/test_user_stats.py
View file

@ -13,9 +13,9 @@ from geoguessr_mcp.models import UserStats
class TestUserStats:
"""Tests for UserStats model."""
def test_from_api_response(self, mock_stats_response):
def test_from_api_response(self, mock_stats_data):
"""Test creating stats from API response."""
stats = UserStats.from_api_response(mock_stats_response)
stats = UserStats.from_api_response(mock_stats_data)
assert stats.games_played == 100
assert stats.rounds_played == 500
@ -42,9 +42,9 @@ class TestUserStats:
assert stats.perfect_games == 5
assert stats.streak_best == 15
def test_to_dict(self, mock_stats_response):
def test_to_dict(self, mock_stats_data):
"""Test serializing stats to dict."""
stats = UserStats.from_api_response(mock_stats_response)
stats = UserStats.from_api_response(mock_stats_data)
result = stats.to_dict()
assert result["games_played"] == 100