Fix middleware schema caching error #7

Merged
NyxiumYuuki merged 13 commits from claude/fix-middleware-schema-cache-01SXnBSQQ8ptLgqViF4UsUYx into master 2025-12-01 03:03:24 +01:00
36 changed files with 780 additions and 132 deletions

View file

@ -54,7 +54,8 @@ MONITORING_ENABLED=true
MONITORING_INTERVAL_HOURS=24
# Directory to store schema cache (persisted between restarts)
SCHEMA_CACHE_DIR=/app/data/schemas
# Default: ./data/schemas (local dev) or /app/data/schemas (Docker)
SCHEMA_CACHE_DIR=./data/schemas
# =============================================================================
# Logging Configuration

581
DEVELOPMENT.md Normal file
View file

@ -0,0 +1,581 @@
# Development Guide
This guide covers local development setup, testing, and debugging the GeoGuessr MCP Server.
## Table of Contents
- [Development Setup](#development-setup)
- [Running Locally](#running-locally)
- [Testing with MCP Inspector](#testing-with-mcp-inspector)
- [Testing with Claude Desktop](#testing-with-claude-desktop)
- [Running Tests](#running-tests)
- [Code Quality](#code-quality)
- [Debugging](#debugging)
- [Common Issues](#common-issues)
## Development Setup
### Prerequisites
- Python 3.11 or higher
- `uv` (recommended) or `pip`
- Git
- A GeoGuessr account (for authenticated endpoints)
### 1. Clone the Repository
```bash
git clone https://github.com/NyxiumYuuki/GeoGuessrMCP.git
cd GeoGuessrMCP
```
### 2. Create Virtual Environment
**Using uv (recommended):**
```bash
uv venv
source .venv/bin/activate # On Linux/macOS
# OR
.venv\Scripts\activate # On Windows
```
**Using venv:**
```bash
python -m venv .venv
source .venv/bin/activate # On Linux/macOS
# OR
.venv\Scripts\activate # On Windows
```
### 3. Install Dependencies
**Development dependencies (includes testing tools):**
```bash
uv pip install -e ".[dev]"
# OR
pip install -e ".[dev]"
```
**Production dependencies only:**
```bash
uv pip install -e .
# OR
pip install -e .
```
### 4. Configure Environment Variables
```bash
cp .env.example .env
```
Edit `.env` and configure:
**Required for authenticated endpoints:**
```bash
GEOGUESSR_NCFA_COOKIE=your_cookie_here
```
**Optional development settings:**
```bash
# Disable authentication for local testing
MCP_AUTH_ENABLED=false
# Enable debug logging
LOG_LEVEL=DEBUG
# Local schema cache
SCHEMA_CACHE_DIR=./data/schemas
# Server configuration
MCP_HOST=0.0.0.0
MCP_PORT=8000
MCP_TRANSPORT=streamable-http
```
## Running Locally
### Start the Server
```bash
python -m src.geoguessr_mcp.main
```
The server will start on `http://0.0.0.0:8000` by default.
**Expected output:**
```
INFO - Starting GeoGuessr MCP Server on 0.0.0.0:8000 with streamable-http transport
INFO - MCP server authentication is DISABLED - server is publicly accessible
INFO - Started server process [12345]
INFO - Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
```
### Enable Debug Logging
Set `LOG_LEVEL=DEBUG` in `.env` to see detailed request/response logs:
```bash
LOG_LEVEL=DEBUG
```
This enables:
- Request method and path logging
- Request headers logging
- Authentication flow details
- MCP protocol messages
- Schema detection changes
## Testing with MCP Inspector
**MCP Inspector** is a web-based tool for testing MCP servers interactively. It's the recommended way to develop and debug MCP servers.
### 1. Install MCP Inspector
Download from: [https://github.com/modelcontextprotocol/inspector](https://github.com/modelcontextprotocol/inspector)
Or run with npx:
```bash
npx @modelcontextprotocol/inspector
```
MCP Inspector will start on `http://localhost:6274` by default.
### 2. Configure Connection
#### Without Authentication
In MCP Inspector:
1. Select **"Server-Sent Events (SSE)"** or **"Streamable HTTP"** transport
2. Set URL: `http://localhost:8000/mcp`
3. Click **"Connect"**
#### With Authentication
In MCP Inspector:
1. Select **"Streamable HTTP"** transport
2. Set URL: `http://localhost:8000/mcp`
3. Click **"Custom Headers"**
4. Add header:
- **Name:** `Authorization`
- **Value:** `Bearer YOUR_API_KEY`
5. Click **"Connect"**
**Generate API key:**
```bash
openssl rand -hex 32
```
Add to `.env`:
```bash
MCP_AUTH_ENABLED=true
MCP_API_KEYS=your-generated-key-here
```
### 3. Test the Connection
After connecting, you should see:
- ✅ **Connection Status:** Connected
- ✅ **Protocol Version:** 2025-06-18
- ✅ **Server Info:** GeoGuessr MCP v1.22.0
### 4. Explore Available Tools
In the **"Tools"** tab, you should see:
**Authentication Tools:**
- `login(email, password)` - Login with GeoGuessr credentials
- `logout()` - Logout current session
- `set_ncfa_cookie(cookie)` - Set authentication cookie manually
- `get_current_session_token()` - Get current session status
**Profile Tools:**
- `get_profile(username)` - Get user profile
- `get_current_user_profile()` - Get your profile
- `get_user_stats(user_id)` - Get detailed stats
**Game Tools:**
- `get_game(game_id)` - Get game details
- `get_recent_games(count)` - Get recent games
- `get_activity_feed(count)` - Get activity feed
**Analysis Tools:**
- `get_performance_summary()` - Comprehensive performance overview
- `analyze_recent_games(count)` - Analyze recent gameplay
- `get_strategy_recommendations()` - Get improvement tips
**Monitoring Tools:**
- `check_api_status()` - Check API endpoint availability
- `list_available_endpoints()` - List all monitored endpoints
- `get_endpoint_schema(path)` - Get schema for specific endpoint
- `explore_endpoint(path)` - Manually explore new endpoints
### 5. Test a Tool
Try calling a simple tool:
1. Select `get_current_session_token()` from the tools list
2. Click **"Call Tool"**
3. Check the response
**Expected response (not authenticated):**
```json
{
"has_token": false,
"message": "No authentication cookie set. Use login() or set_ncfa_cookie()."
}
```
**Expected response (authenticated):**
```json
{
"has_token": true,
"user_id": "your-user-id",
"username": "your-username"
}
```
### 6. Debugging Connection Issues
If you see errors in MCP Inspector, check the server logs:
**Common issues:**
**❌ OPTIONS 405 Method Not Allowed**
- **Cause:** CORS middleware not configured
- **Fix:** Ensure middleware is properly applied (should be fixed in latest version)
**❌ POST 401 Unauthorized**
- **Cause:** Missing or invalid API key
- **Fix:** Add correct `Authorization: Bearer YOUR_KEY` header
**❌ POST 400 Bad Request**
- **Cause:** Session ID not maintained (CORS headers not exposed)
- **Fix:** Ensure `expose_headers` includes `mcp-session-id` (should be fixed in latest version)
**❌ Connection timeout**
- **Cause:** Server not running or firewall blocking
- **Fix:** Check server is running on `localhost:8000`
## Testing with Claude Desktop
For testing the full MCP integration with Claude Desktop:
### 1. Configure Claude Desktop
**macOS:** `~/Library/Application Support/Claude/claude_desktop_config.json`
**Windows:** `%APPDATA%\Claude\claude_desktop_config.json`
**Without authentication:**
```json
{
"mcpServers": {
"geoguessr-local": {
"type": "streamable-http",
"url": "http://localhost:8000/mcp"
}
}
}
```
**With authentication:**
```json
{
"mcpServers": {
"geoguessr-local": {
"type": "streamable-http",
"url": "http://localhost:8000/mcp",
"headers": {
"Authorization": "Bearer YOUR_API_KEY"
}
}
}
}
```
### 2. Restart Claude Desktop
After modifying the config, completely quit and restart Claude Desktop.
### 3. Test in Claude
In a new conversation:
```
Can you check my GeoGuessr profile?
```
Claude should automatically use the MCP server tools to fetch your profile.
## Running Tests
### Unit Tests
```bash
pytest src/tests/unit/ -v
```
### Integration Tests
**Requires authentication:**
```bash
# Set GEOGUESSR_NCFA_COOKIE in .env first
pytest src/tests/integration/ -v
```
### All Tests with Coverage
```bash
pytest --cov=src/geoguessr_mcp src/tests/ -v
```
### Coverage Report
```bash
pytest --cov=src/geoguessr_mcp --cov-report=html src/tests/
open htmlcov/index.html # View coverage report
```
## Code Quality
### Format Code with Black
```bash
black src/ tests/
```
**Configuration:** Line length 100 (see `pyproject.toml`)
### Lint with Ruff
```bash
ruff check src/ tests/
```
**Auto-fix:**
```bash
ruff check --fix src/ tests/
```
### Type Check with MyPy
```bash
mypy src/
```
### Pre-commit Hooks
Install pre-commit hooks to automatically run checks:
```bash
pre-commit install
```
This will run Black, Ruff, and MyPy before each commit.
## Debugging
### Enable Debug Mode
Set in `.env`:
```bash
LOG_LEVEL=DEBUG
```
Restart the server to see:
- All HTTP requests with headers
- Authentication flow
- MCP protocol messages
- Schema detection and changes
### Debug Authentication
Check if your GeoGuessr cookie is valid:
```bash
# In Python console
from src.geoguessr_mcp.auth import session_manager
import asyncio
async def test():
token = session_manager.get_current_session_token()
print(f"Token: {token}")
asyncio.run(test())
```
### Debug API Calls
Test API endpoints directly:
```bash
# Install httpx
pip install httpx
# Test endpoint
python -c "
import httpx
import asyncio
async def test():
async with httpx.AsyncClient() as client:
response = await client.get(
'https://www.geoguessr.com/api/v3/profiles',
cookies={'_ncfa': 'YOUR_COOKIE'}
)
print(response.json())
asyncio.run(test())
"
```
### Debug Schema Detection
Check schema cache:
```bash
ls -la data/schemas/
cat data/schemas/schemas.json | python -m json.tool
```
Force schema refresh:
```bash
rm -rf data/schemas/*
# Restart server - schemas will be regenerated
```
### View Server Logs
**Real-time logs:**
```bash
tail -f server.log # If using file logging
```
**Docker logs:**
```bash
docker compose logs -f geoguessr-mcp
```
## Common Issues
### Issue: Module not found errors
**Solution:**
```bash
# Ensure you're in the project root
pip install -e ".[dev]"
```
### Issue: Permission denied on schema cache
**Solution:**
```bash
mkdir -p data/schemas
chmod 755 data/schemas
```
Or use a different path in `.env`:
```bash
SCHEMA_CACHE_DIR=/tmp/geoguessr-schemas
```
### Issue: Port 8000 already in use
**Solution:**
```bash
# Use different port
echo "MCP_PORT=8001" >> .env
```
Or kill the process using port 8000:
```bash
# Linux/macOS
lsof -ti:8000 | xargs kill -9
# Windows
netstat -ano | findstr :8000
taskkill /PID <PID> /F
```
### Issue: CORS errors in browser
**Solution:**
Ensure CORS middleware is properly configured with `expose_headers`:
```python
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
expose_headers=["mcp-session-id", "mcp-protocol-version"],
)
```
### Issue: Session not maintained (400 errors)
**Symptoms:**
- First POST succeeds (200 OK)
- Second POST fails (400 Bad Request)
- Logs show new session created each time
**Solution:**
The `mcp-session-id` header must be exposed via CORS (fixed in latest version).
Check server logs for:
```
expose_headers=["mcp-session-id", "mcp-protocol-version"]
```
## Contributing
### Development Workflow
1. Create a feature branch
2. Make changes
3. Run tests: `pytest`
4. Run code quality checks: `black`, `ruff`, `mypy`
5. Commit with descriptive message
6. Push and create pull request
### Code Style
- **Formatting:** Black (line length 100)
- **Linting:** Ruff
- **Type hints:** Required for all functions
- **Docstrings:** Google style for public APIs
- **Tests:** Required for new features
### Project Structure
```
src/geoguessr_mcp/
├── api/ # GeoGuessr API client
├── auth/ # Authentication & session management
├── middleware/ # MCP server middleware (auth, CORS)
├── models/ # Data models (Profile, Stats, Games)
├── monitoring/ # API monitoring & schema detection
│ ├── endpoint/ # Endpoint monitoring
│ └── schema/ # Schema detection & registry
├── services/ # Business logic services
├── tools/ # MCP tool definitions
├── config.py # Configuration management
└── main.py # Application entry point
```
## Resources
- [MCP Documentation](https://modelcontextprotocol.io/)
- [FastMCP Guide](https://github.com/jlowin/fastmcp)
- [MCP Inspector](https://github.com/modelcontextprotocol/inspector)
- [GeoGuessr API (unofficial)](https://www.geoguessr.com/api)
- [Starlette Middleware](https://www.starlette.io/middleware/)
## Support
For issues and questions:
- Check this guide first
- Review server logs with `LOG_LEVEL=DEBUG`
- Check existing issues on GitHub
- Create a new issue with logs and reproduction steps

View file

@ -4,8 +4,8 @@ GeoGuessr API Endpoints Registry.
Centralized endpoint definitions with metadata for dynamic discovery and routing.
"""
from collections.abc import Callable
from dataclasses import dataclass
from typing import Callable, Optional
from ..config import settings
@ -19,7 +19,7 @@ class EndpointInfo:
description: str = ""
auth_required: bool = True
use_game_server: bool = False
params_builder: Optional[Callable[..., dict]] = None
params_builder: Callable[..., dict] | None = None
class Endpoints:

View file

@ -10,16 +10,15 @@ Classes:
"""
import logging
from typing import Optional
import httpx
from .dynamic_response import DynamicResponse
from .endpoints import EndpointInfo
from ..auth import get_current_user_context
from ..auth.session import SessionManager
from ..config import settings
from ..monitoring.schema.schema_registry import schema_registry
from .dynamic_response import DynamicResponse
from .endpoints import EndpointInfo
logger = logging.getLogger(__name__)
@ -45,7 +44,7 @@ class GeoGuessrClient:
async def _get_authenticated_client(
self,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> httpx.AsyncClient:
"""
Get an authenticated HTTP client.
@ -79,9 +78,9 @@ class GeoGuessrClient:
async def request(
self,
endpoint: EndpointInfo,
session_token: Optional[str] = None,
params: Optional[dict] = None,
json_data: Optional[dict] = None,
session_token: str | None = None,
params: dict | None = None,
json_data: dict | None = None,
**kwargs,
) -> DynamicResponse:
"""
@ -154,8 +153,8 @@ class GeoGuessrClient:
async def get(
self,
endpoint: EndpointInfo,
session_token: Optional[str] = None,
params: Optional[dict] = None,
session_token: str | None = None,
params: dict | None = None,
**kwargs,
) -> DynamicResponse:
"""Make a GET request."""
@ -164,8 +163,8 @@ class GeoGuessrClient:
async def post(
self,
endpoint: EndpointInfo,
session_token: Optional[str] = None,
json_data: Optional[dict] = None,
session_token: str | None = None,
json_data: dict | None = None,
**kwargs,
) -> DynamicResponse:
"""Make a POST request."""
@ -174,9 +173,9 @@ class GeoGuessrClient:
async def get_raw(
self,
path: str,
session_token: Optional[str] = None,
session_token: str | None = None,
use_game_server: bool = False,
params: Optional[dict] = None,
params: dict | None = None,
) -> DynamicResponse:
"""
Make a raw GET request to any path.

View file

@ -7,7 +7,6 @@ where each API key can have its own GeoGuessr session.
import asyncio
import logging
from typing import Optional
from ..config import settings
from .session import SessionManager, UserSession
@ -146,7 +145,7 @@ class MultiUserSessionManager:
return True
async def get_session_for_api_key(self, api_key: str) -> Optional[UserSession]:
async def get_session_for_api_key(self, api_key: str) -> UserSession | None:
"""
Get the active session for a specific API key.

View file

@ -7,12 +7,11 @@ the authenticated user making the request.
"""
from contextvars import ContextVar
from typing import Optional
from .user_context import UserContext
# Context variable to store the current user context
_current_user_context: ContextVar[Optional[UserContext]] = ContextVar(
_current_user_context: ContextVar[UserContext | None] = ContextVar(
"current_user_context", default=None
)
@ -30,7 +29,7 @@ def set_current_user_context(context: UserContext) -> None:
_current_user_context.set(context)
def get_current_user_context() -> Optional[UserContext]:
def get_current_user_context() -> UserContext | None:
"""
Get the current user context.

View file

@ -7,7 +7,6 @@ import logging
import secrets
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from typing import Optional
import httpx
@ -25,7 +24,7 @@ class UserSession:
username: str
email: str
created_at: datetime = field(default_factory=datetime.now)
expires_at: Optional[datetime] = None
expires_at: datetime | None = None
def is_valid(self) -> bool:
"""Check if the session is still valid."""
@ -37,10 +36,10 @@ class UserSession:
class SessionManager:
"""Manages user sessions for the MCP server."""
def __init__(self, default_cookie: Optional[str] = None):
def __init__(self, default_cookie: str | None = None):
self._sessions: dict[str, UserSession] = {}
self._user_sessions: dict[str, str] = {}
self._default_cookie: Optional[str] = default_cookie or settings.DEFAULT_NCFA_COOKIE
self._default_cookie: str | None = default_cookie or settings.DEFAULT_NCFA_COOKIE
self._lock = asyncio.Lock()
@staticmethod
@ -111,7 +110,7 @@ class SessionManager:
return session_token, session
@staticmethod
def _extract_ncfa_cookie(response: httpx.Response) -> Optional[str]:
def _extract_ncfa_cookie(response: httpx.Response) -> str | None:
"""Extract _ncfa cookie from response."""
# Try cookies jar first
for cookie in response.cookies.jar:
@ -160,7 +159,7 @@ class SessionManager:
return True
return False
async def get_session(self, session_token: Optional[str] = None) -> Optional[UserSession]:
async def get_session(self, session_token: str | None = None) -> UserSession | None:
"""
Get a session by token or return default if available.
@ -203,7 +202,7 @@ class SessionManager:
logger.info("Default NCFA cookie updated")
@staticmethod
async def validate_cookie(cookie: str) -> Optional[dict]:
async def validate_cookie(cookie: str) -> dict | None:
"""
Validate a cookie by making a test request.

View file

@ -6,7 +6,6 @@ is making a request and their associated GeoGuessr session.
"""
from dataclasses import dataclass
from typing import Optional
from .session import UserSession
@ -23,7 +22,7 @@ class UserContext:
api_key: str
"""The API key used to authenticate this request"""
session: Optional[UserSession] = None
session: UserSession | None = None
"""The GeoGuessr session for this user (if logged in)"""
@property
@ -41,7 +40,7 @@ class UserContext:
return f"User-{hash(self.api_key) % 10000:04d}"
@property
def ncfa_cookie(self) -> Optional[str]:
def ncfa_cookie(self) -> str | None:
"""Get the NCFA cookie for this user."""
if self.session:
return self.session.ncfa_cookie

View file

@ -2,7 +2,6 @@
import os
from dataclasses import dataclass, field
from typing import Optional
@dataclass
@ -18,7 +17,7 @@ class Settings:
GEOGUESSR_DOMAIN_NAME: str = "geoguessr.com"
GEOGUESSR_API_URL: str = "https://www.geoguessr.com/api"
GAME_SERVER_URL: str = "https://game-server.geoguessr.com/api"
DEFAULT_NCFA_COOKIE: Optional[str] = field(
DEFAULT_NCFA_COOKIE: str | None = field(
default_factory=lambda: os.getenv("GEOGUESSR_NCFA_COOKIE")
)
@ -30,14 +29,14 @@ class Settings:
default_factory=lambda: int(os.getenv("MONITORING_INTERVAL_HOURS", "24"))
)
SCHEMA_CACHE_DIR: str = field(
default_factory=lambda: os.getenv("SCHEMA_CACHE_DIR", "/app/data/schemas")
default_factory=lambda: os.getenv("SCHEMA_CACHE_DIR", "./data/schemas")
)
# Authentication Configuration
MCP_AUTH_ENABLED: bool = field(
default_factory=lambda: os.getenv("MCP_AUTH_ENABLED", "false").lower() == "true"
)
MCP_API_KEYS: Optional[str] = field(default_factory=lambda: os.getenv("MCP_API_KEYS"))
MCP_API_KEYS: str | None = field(default_factory=lambda: os.getenv("MCP_API_KEYS"))
# Logging Configuration
LOG_LEVEL: str = field(default_factory=lambda: os.getenv("LOG_LEVEL", "INFO"))

View file

@ -9,7 +9,9 @@ import logging
import sys
from mcp.server.fastmcp import FastMCP
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.middleware.cors import CORSMiddleware
from starlette.requests import Request
from .config import settings
from .middleware import AuthenticationMiddleware
@ -21,16 +23,33 @@ logging.basicConfig(
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger(__name__)
class RequestLoggingMiddleware(BaseHTTPMiddleware):
"""Log request details for debugging."""
async def dispatch(self, request: Request, call_next):
"""Log request and response details."""
logger.debug(f"Request: {request.method} {request.url.path}")
logger.debug(f"Headers: {dict(request.headers)}")
response = await call_next(request)
if response.status_code >= 400:
logger.warning(
f"Error response: {request.method} {request.url.path} -> {response.status_code}"
)
return response
def main():
"""Main entry point for the server."""
# Create the MCP server instance
mcp = FastMCP(
"GeoGuessr Analyzer",
"GeoGuessr MCP",
instructions="""
MCP server for analyzing GeoGuessr game statistics and optimizing gameplay strategy.
@ -62,22 +81,62 @@ def main():
# Register all tools
register_all_tools(mcp)
# Setup authentication middleware if enabled
if settings.MCP_AUTH_ENABLED:
logger.info("Setting up authentication middleware")
# Wrap the streamable_http_app method to inject middleware
_original_streamable_http_app = mcp.streamable_http_app
# Récupérez l'application ASGI via streamable_http_app
mcp_app = mcp.streamable_http_app()
def _streamable_http_app_with_middleware():
"""Wrap app creation to inject middleware."""
app = _original_streamable_http_app()
# Ajoutez les middlewares
mcp_app.add_middleware(
# Add request logging middleware for debugging (first in chain)
if settings.LOG_LEVEL == "DEBUG":
app.add_middleware(RequestLoggingMiddleware)
# Always add CORS middleware for browser compatibility
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
expose_headers=["mcp-session-id", "mcp-protocol-version"],
)
mcp_app.add_middleware(AuthenticationMiddleware)
# Add authentication middleware if enabled
if settings.MCP_AUTH_ENABLED:
app.add_middleware(AuthenticationMiddleware)
return app
# Replace the method with our wrapper
mcp.streamable_http_app = _streamable_http_app_with_middleware
# Also wrap sse_app for SSE transport
if hasattr(mcp, "sse_app"):
_original_sse_app = mcp.sse_app
def _sse_app_with_middleware():
"""Wrap SSE app creation to inject middleware."""
app = _original_sse_app()
if settings.LOG_LEVEL == "DEBUG":
app.add_middleware(RequestLoggingMiddleware)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
expose_headers=["mcp-session-id", "mcp-protocol-version"],
)
if settings.MCP_AUTH_ENABLED:
app.add_middleware(AuthenticationMiddleware)
return app
mcp.sse_app = _sse_app_with_middleware
logger.info(
f"Starting GeoGuessr MCP Server on {settings.HOST}:{settings.PORT} "
@ -98,7 +157,7 @@ def main():
"Users will need to login or provide a cookie."
)
# Run the server
# Run the server - middleware will be applied via our wrapper
mcp.run(transport=settings.TRANSPORT)

View file

@ -6,7 +6,6 @@ and attaches user context for multi-user support.
"""
import logging
from typing import Optional
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
@ -26,7 +25,7 @@ class AuthenticationMiddleware(BaseHTTPMiddleware):
Authorization header with a Bearer token matching one of the configured API keys.
"""
def __init__(self, app, valid_api_keys: Optional[set[str]] = None):
def __init__(self, app, valid_api_keys: set[str] | None = None):
super().__init__(app)
self.valid_api_keys = valid_api_keys or settings.get_api_keys()
self.enabled = settings.MCP_AUTH_ENABLED
@ -54,6 +53,11 @@ class AuthenticationMiddleware(BaseHTTPMiddleware):
if request.url.path == "/health":
return await call_next(request)
# Skip authentication for OPTIONS requests (CORS preflight)
# OPTIONS requests don't include Authorization headers by design
if request.method == "OPTIONS":
return await call_next(request)
# Check for Authorization header
auth_header = request.headers.get("Authorization")

View file

@ -1,7 +1,6 @@
"""Achievement-related data models."""
from dataclasses import dataclass
from typing import Optional
@dataclass
@ -12,9 +11,9 @@ class Achievement:
name: str
description: str = ""
unlocked: bool = False
unlocked_at: Optional[str] = None
unlocked_at: str | None = None
progress: float = 0.0
icon_url: Optional[str] = None
icon_url: str | None = None
@classmethod
def from_api_response(cls, data: dict) -> "Achievement":

View file

@ -1,7 +1,6 @@
"""DailyChallenge-related data models."""
from dataclasses import dataclass, field
from typing import Optional
@dataclass
@ -13,7 +12,7 @@ class DailyChallenge:
date: str = ""
time_limit: int = 0
completed: bool = False
score: Optional[int] = None
score: int | None = None
raw_data: dict = field(default_factory=dict)
@classmethod

View file

@ -1,7 +1,6 @@
"""Game-related data models."""
from dataclasses import dataclass, field
from typing import Optional
from .round_guess import RoundGuess
@ -15,7 +14,7 @@ class Game:
mode: str
total_score: int
rounds: list[RoundGuess] = field(default_factory=list)
created_at: Optional[str] = None
created_at: str | None = None
finished: bool = False
raw_data: dict = field(default_factory=dict)

View file

@ -1,7 +1,6 @@
"""RoundGuess-related data models."""
from dataclasses import dataclass
from typing import Optional
@dataclass
@ -12,8 +11,8 @@ class RoundGuess:
score: int
distance_meters: float
time_seconds: int
lat: Optional[float] = None
lng: Optional[float] = None
lat: float | None = None
lng: float | None = None
country: str = ""
@classmethod

View file

@ -1,7 +1,6 @@
"""UserProfile-related data models."""
from dataclasses import dataclass, field
from typing import Optional
@dataclass
@ -16,7 +15,7 @@ class UserProfile:
created: str = ""
is_verified: bool = False
is_pro: bool = False
avatar_url: Optional[str] = None
avatar_url: str | None = None
raw_data: dict = field(default_factory=dict)
@classmethod

View file

@ -12,16 +12,16 @@ Classes:
"""
import asyncio
import contextlib
import logging
from datetime import UTC, datetime
from typing import Optional
import httpx
from ...config import settings
from ..schema.schema_registry import SchemaRegistry, schema_registry
from .endpoint_definition import EndpointDefinition
from .endpoint_monitoring_result import MonitoringResult
from ..schema.schema_registry import SchemaRegistry, schema_registry
from ...config import settings
logger = logging.getLogger(__name__)
@ -119,14 +119,14 @@ class EndpointMonitor:
def __init__(
self,
registry: Optional[SchemaRegistry] = None,
ncfa_cookie: Optional[str] = None,
registry: SchemaRegistry | None = None,
ncfa_cookie: str | None = None,
):
self.registry = registry or schema_registry
self.ncfa_cookie = ncfa_cookie or settings.DEFAULT_NCFA_COOKIE
self.results: list[MonitoringResult] = []
self._running = False
self._task: Optional[asyncio.Task] = None
self._task: asyncio.Task | None = None
async def check_endpoint(
self,
@ -286,10 +286,8 @@ class EndpointMonitor:
self._running = False
if self._task:
self._task.cancel()
try:
with contextlib.suppress(asyncio.CancelledError):
await self._task
except asyncio.CancelledError:
pass
logger.info("Stopped periodic monitoring")
async def _monitoring_loop(self) -> None:

View file

@ -8,7 +8,6 @@ an endpoint, including its availability, response time, and any errors encounter
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Optional
@dataclass
@ -20,5 +19,5 @@ class MonitoringResult:
response_code: int
response_time_ms: float
schema_changed: bool
error_message: Optional[str] = None
error_message: str | None = None
timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))

View file

@ -11,7 +11,7 @@ schema information.
import logging
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any, Optional
from typing import Any
from .schema_field import SchemaField
@ -29,8 +29,8 @@ class EndpointSchema:
schema_hash: str = ""
response_code: int = 200
is_available: bool = True
error_message: Optional[str] = None
sample_response: Optional[dict] = None
error_message: str | None = None
sample_response: dict | None = None
def to_dict(self) -> dict:
"""Convert to dictionary for serialization."""

View file

@ -6,7 +6,7 @@ a schema field, including its name, type, and other relevant metadata.
"""
from dataclasses import dataclass
from typing import Any, Optional
from typing import Any
@dataclass
@ -16,6 +16,6 @@ class SchemaField:
name: str
field_type: str
nullable: bool = False
nested_schema: Optional[dict] = None
nested_schema: dict | None = None
example_value: Any = None
description: str = ""

View file

@ -15,11 +15,11 @@ import logging
import tempfile
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Optional
from typing import Any
from ...config import settings
from .endpoint_schema import EndpointSchema
from .schema_detector import SchemaDetector
from ...config import settings
logger = logging.getLogger(__name__)
@ -32,7 +32,7 @@ class SchemaRegistry:
to track changes over time and adapt automatically.
"""
def __init__(self, cache_dir: Optional[str] = None):
def __init__(self, cache_dir: str | None = None):
self.cache_dir = Path(cache_dir or settings.SCHEMA_CACHE_DIR)
# Try to create the cache directory, fall back to temp if permission denied
@ -71,6 +71,16 @@ class SchemaRegistry:
for endpoint, schema_data in data.items():
self.schemas[endpoint] = EndpointSchema.from_dict(schema_data)
logger.info(f"Loaded {len(self.schemas)} cached schemas")
except json.JSONDecodeError as e:
logger.warning(
f"Failed to load cached schemas due to corrupted JSON: {e}. "
f"Removing corrupted cache file."
)
try:
schema_file.unlink()
logger.info(f"Removed corrupted schema cache file: {schema_file}")
except Exception as rm_error:
logger.error(f"Failed to remove corrupted cache file: {rm_error}")
except Exception as e:
logger.warning(f"Failed to load cached schemas: {e}")
@ -83,6 +93,16 @@ class SchemaRegistry:
self.schema_history[endpoint] = [
EndpointSchema.from_dict(h) for h in history
]
except json.JSONDecodeError as e:
logger.warning(
f"Failed to load schema history due to corrupted JSON: {e}. "
f"Removing corrupted history file."
)
try:
history_file.unlink()
logger.info(f"Removed corrupted schema history file: {history_file}")
except Exception as rm_error:
logger.error(f"Failed to remove corrupted history file: {rm_error}")
except Exception as e:
logger.warning(f"Failed to load schema history: {e}")
@ -167,7 +187,7 @@ class SchemaRegistry:
)
self._save_schemas()
def get_schema(self, endpoint: str) -> Optional[EndpointSchema]:
def get_schema(self, endpoint: str) -> EndpointSchema | None:
"""Get the current schema for an endpoint."""
return self.schemas.get(endpoint)

View file

@ -7,13 +7,12 @@ dynamic data handling and LLM-friendly output formatting.
import logging
from dataclasses import dataclass, field
from typing import Optional
from .game_service import GameService
from .profile_service import ProfileService
from ..api import GeoGuessrClient
from ..models import Game
from ..monitoring import schema_registry
from .game_service import GameService
from .profile_service import ProfileService
logger = logging.getLogger(__name__)
@ -61,8 +60,8 @@ class AnalysisService:
def __init__(
self,
client: GeoGuessrClient,
game_service: Optional[GameService] = None,
profile_service: Optional[ProfileService] = None,
game_service: GameService | None = None,
profile_service: ProfileService | None = None,
):
self.client = client
self.game_service = game_service or GameService(client)
@ -155,7 +154,7 @@ class AnalysisService:
async def analyze_recent_games(
self,
count: int = 10,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> dict:
"""
Analyze recent games and provide statistics summary.
@ -181,7 +180,7 @@ class AnalysisService:
async def get_performance_summary(
self,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> dict:
"""
Get a comprehensive performance summary.
@ -246,7 +245,7 @@ class AnalysisService:
async def get_strategy_recommendations(
self,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> dict:
"""
Generate strategy recommendations based on performance analysis.

View file

@ -5,9 +5,8 @@ Handles game history, details, and competitive data with dynamic schema support.
"""
import logging
from typing import Optional
from ..api import Endpoints, DynamicResponse, GeoGuessrClient
from ..api import DynamicResponse, Endpoints, GeoGuessrClient
from ..models import DailyChallenge, Game, SeasonStats
logger = logging.getLogger(__name__)
@ -22,7 +21,7 @@ class GameService:
async def get_game_details(
self,
game_token: str,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> tuple[Game, DynamicResponse]:
"""
Get details for a specific game.
@ -45,7 +44,7 @@ class GameService:
async def get_unfinished_games(
self,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> DynamicResponse:
"""Get list of unfinished games."""
return await self.client.get(Endpoints.GAMES.GET_UNFINISHED_GAMES, session_token)
@ -53,7 +52,7 @@ class GameService:
async def get_streak_game(
self,
game_token: str,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> DynamicResponse:
"""Get streak game details."""
endpoint = Endpoints.GAMES.get_streak_game(game_token)
@ -63,7 +62,7 @@ class GameService:
self,
count: int = 10,
page: int = 0,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> DynamicResponse:
"""
Get the activity feed.
@ -82,7 +81,7 @@ class GameService:
async def get_recent_games(
self,
count: int = 10,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> list[Game]:
"""
Get recent games from the activity feed.
@ -122,7 +121,7 @@ class GameService:
async def get_season_stats(
self,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> tuple[SeasonStats, DynamicResponse]:
"""Get active season statistics."""
response = await self.client.get(
@ -138,7 +137,7 @@ class GameService:
async def get_daily_challenge(
self,
day: str = "today",
session_token: Optional[str] = None,
session_token: str | None = None,
) -> tuple[DailyChallenge, DynamicResponse]:
"""
Get daily challenge.
@ -162,7 +161,7 @@ class GameService:
async def get_battle_royale(
self,
game_id: str,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> DynamicResponse:
"""Get battle royale game details."""
endpoint = Endpoints.GAME_SERVER.get_battle_royale(game_id)
@ -171,7 +170,7 @@ class GameService:
async def get_duel(
self,
duel_id: str,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> DynamicResponse:
"""Get duel game details."""
endpoint = Endpoints.GAME_SERVER.get_duel(duel_id)
@ -179,7 +178,7 @@ class GameService:
async def get_tournaments(
self,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> DynamicResponse:
"""Get tournament information."""
return await self.client.get(Endpoints.GAME_SERVER.GET_TOURNAMENTS, session_token)

View file

@ -6,9 +6,8 @@ dynamic schema adaptation.
"""
import logging
from typing import Optional
from ..api import DynamicResponse, GeoGuessrClient, Endpoints
from ..api import DynamicResponse, Endpoints, GeoGuessrClient
from ..models import Achievement, UserProfile, UserStats
logger = logging.getLogger(__name__)
@ -22,7 +21,7 @@ class ProfileService:
async def get_profile(
self,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> tuple[UserProfile, DynamicResponse]:
"""
Get current user's profile.
@ -40,7 +39,7 @@ class ProfileService:
async def get_stats(
self,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> tuple[UserStats, DynamicResponse]:
"""
Get user statistics.
@ -58,7 +57,7 @@ class ProfileService:
async def get_extended_stats(
self,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> DynamicResponse:
"""
Get extended statistics.
@ -69,7 +68,7 @@ class ProfileService:
async def get_achievements(
self,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> tuple[list[Achievement], DynamicResponse]:
"""
Get user achievements.
@ -96,7 +95,7 @@ class ProfileService:
async def get_public_profile(
self,
user_id: str,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> tuple[UserProfile, DynamicResponse]:
"""Get another user's public profile."""
endpoint = Endpoints.PROFILES.get_public_profile(user_id)
@ -110,14 +109,14 @@ class ProfileService:
async def get_user_maps(
self,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> DynamicResponse:
"""Get user's custom maps."""
return await self.client.get(Endpoints.PROFILES.GET_USER_MAPS, session_token)
async def get_comprehensive_profile(
self,
session_token: Optional[str] = None,
session_token: str | None = None,
) -> dict:
"""
Get a comprehensive profile combining multiple endpoints.

View file

@ -2,17 +2,17 @@
from mcp.server.fastmcp import FastMCP
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 .monitoring_tools import register_monitoring_tools
from .profile_tools import register_profile_tools
def register_all_tools(mcp: FastMCP) -> dict:

View file

@ -11,8 +11,8 @@ 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
from .auth_tools import get_current_session_token
def register_analysis_tools(mcp: FastMCP, analysis_service: AnalysisService):

View file

@ -13,8 +13,8 @@ Functions:
from mcp.server.fastmcp import FastMCP
from .auth_tools import get_current_session_token
from ..services.game_service import GameService
from .auth_tools import get_current_session_token
def register_game_tools(mcp: FastMCP, game_service: GameService):
@ -99,7 +99,7 @@ def register_game_tools(mcp: FastMCP, game_service: GameService):
"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)),
"maps_played": list({g.map_name for g in games}),
},
}

View file

@ -10,8 +10,8 @@ evolution.
from mcp.server.fastmcp import FastMCP
from .auth_tools import get_current_session_token
from ..monitoring import endpoint_monitor, schema_registry
from .auth_tools import get_current_session_token
def register_monitoring_tools(mcp: FastMCP):

View file

@ -12,8 +12,8 @@ from the underlying service API. Tools return structured data for easy consumpti
from mcp.server.fastmcp import FastMCP
from .auth_tools import get_current_session_token
from ..services.profile_service import ProfileService
from .auth_tools import get_current_session_token
def register_profile_tools(mcp: FastMCP, profile_service: ProfileService):

View file

@ -9,7 +9,7 @@ from geoguessr_mcp.api import GeoGuessrClient
from geoguessr_mcp.api.dynamic_response import DynamicResponse
from geoguessr_mcp.auth import SessionManager, UserSession
from geoguessr_mcp.config import settings
from geoguessr_mcp.models import RoundGuess, Game
from geoguessr_mcp.models import Game, RoundGuess
from geoguessr_mcp.services import AnalysisService, GameService, ProfileService

View file

@ -10,7 +10,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
from geoguessr_mcp.api import DynamicResponse, GeoGuessrClient, EndpointInfo, Endpoints
from geoguessr_mcp.api import DynamicResponse, EndpointInfo, Endpoints, GeoGuessrClient
from geoguessr_mcp.config import settings

View file

@ -187,7 +187,7 @@ class TestAuthenticationFlow:
assert "expired_user" not in session_manager._user_sessions
@pytest.mark.asyncio
async def test_default_cookie_fallback(self, session_manager):
async def test_default_cookie_fallback(self):
"""Test falling back to default cookie when no session exists."""
# Create manager with default cookie
manager_with_default = SessionManager(default_cookie="default_test_cookie")

View file

@ -26,8 +26,8 @@ class TestMultiUserSessionManager:
@pytest.mark.asyncio
async def test_get_user_context_reuses_existing_manager(self, manager):
"""Test that getting context for existing API key reuses the same manager."""
context1 = await manager.get_user_context("existing_key")
context2 = await manager.get_user_context("existing_key")
await manager.get_user_context("existing_key")
await manager.get_user_context("existing_key")
# Should use the same manager instance
assert manager._user_managers["existing_key"] is manager._user_managers["existing_key"]
@ -36,9 +36,9 @@ class TestMultiUserSessionManager:
@pytest.mark.asyncio
async def test_multiple_api_keys_get_separate_managers(self, manager):
"""Test that different API keys get separate session managers."""
context1 = await manager.get_user_context("key1")
context2 = await manager.get_user_context("key2")
context3 = await manager.get_user_context("key3")
await manager.get_user_context("key1")
await manager.get_user_context("key2")
await manager.get_user_context("key3")
assert len(manager._user_managers) == 3
assert manager._user_managers["key1"] is not manager._user_managers["key2"]
@ -61,10 +61,11 @@ class TestMultiUserSessionManager:
assert session is None
@pytest.mark.asyncio
async def test_login_user_creates_manager_if_not_exists(self, manager):
async def test_login_user_creates_manager_if_not_exists(self):
"""Test that login_user creates a manager if it doesn't exist."""
# This test requires mocking the HTTP client for GeoGuessr API
# We'll mark it as a placeholder for now
# TODO: Add test for this
pytest.skip("Requires mocking GeoGuessr API")
@pytest.mark.asyncio

View file

@ -1,10 +1,9 @@
"""Tests for UserContext class."""
import pytest
from datetime import UTC, datetime, timedelta
from geoguessr_mcp.auth.session import UserSession
from geoguessr_mcp.auth.user_context import UserContext
from datetime import datetime, timedelta, UTC
class TestUserContext:

View file

@ -11,7 +11,7 @@ feeds, recent games, season statistics, and daily challenges.
import pytest
from geoguessr_mcp.models import Game, SeasonStats, DailyChallenge
from geoguessr_mcp.models import DailyChallenge, Game, SeasonStats
class TestGameService:

View file

@ -9,7 +9,7 @@ operations.
import pytest
from geoguessr_mcp.models import UserProfile, UserStats, Achievement
from geoguessr_mcp.models import Achievement, UserProfile, UserStats
class TestProfileService: