Weather API

async def validate_api_key(api_key: str) -> bool:
    """Validate by making a test request to WeatherAPI.com"""
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{WEATHER_API_BASE}/current.json",
                params={"key": api_key, "q": "London"},
                timeout=5.0
            )
            return response.status_code == 200
    except Exception:
        return False

@mcp.tool()
async def get_weather(city: str, forecast_days: int = None) -> dict:
    # Extract key from Authorization header
    api_key = get_api_key_from_request()

    # Validate against WeatherAPI.com
    if not await validate_api_key(api_key):
        return {"error": "Invalid or expired API key", "status": 401}

    # Forward to WeatherAPI.com with validated key
    response = await client.get(
        f"{WEATHER_API_BASE}/current.json",
        params={"key": api_key, "q": city}
    )
    return response.json()

from collections import defaultdict
import time

rate_limits = defaultdict(list)  # {api_key: [timestamps]}

def check_rate_limit(api_key: str, max_requests: int = 100, window_seconds: int = 3600) -> bool:
    now = time.time()

    # Clean old timestamps
    rate_limits[api_key] = [
        ts for ts in rate_limits[api_key]
        if now - ts < window_seconds
    ]

    # Check limit
    if len(rate_limits[api_key]) >= max_requests:
        return False

    # Record this request
    rate_limits[api_key].append(now)
    return True

@mcp.tool()
async def get_weather(city: str, forecast_days: Optional[int] = None) -> dict:
    """Get weather data - current conditions or forecast up to 10 days."""

    api_key = get_api_key_from_request()

    if not check_rate_limit(api_key):
        return {"error": "Rate limit exceeded", "status": 429}

    if not await validate_api_key(api_key):
        return {"error": "Invalid API key", "status": 401}

    # Determine behavior from parameters
    if forecast_days is None:
        # Current weather
        endpoint = f"{WEATHER_API_BASE}/current.json"
        params = {"key": api_key, "q": city, "aqi": "yes"}
    else:
        # Forecast
        endpoint = f"{WEATHER_API_BASE}/forecast.json"
        params = {"key": api_key, "q": city, "days": forecast_days, "aqi": "yes"}

    # Proxy to WeatherAPI.com
    async with httpx.AsyncClient() as client:
        response = await client.get(endpoint, params=params)
        return response.json()

# Clone
git clone https://github.com/InteractionCo/poke-mcp-examples.git
cd poke-mcp-examples/weather-api

# Environment
conda create -n weather-api python=3.12
conda activate weather-api

# Install
pip install -r requirements.txt

# Run (no .env needed - API key comes from Poke)
python src/server.py