Building a Secure MCP Server for Weather

import logging
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware

logger = logging.getLogger(__name__)

# Global variable to store the current request context
_current_request = None

def set_current_request(request: Request):
    """Set the current request for tools to access"""
    global _current_request
    _current_request = request

def get_api_key():
    """Get the current user's API key"""
    global _current_request

    if _current_request and hasattr(_current_request.state, "api_key"):
        return _current_request.state.api_key

    return None

class AuthMiddleware(BaseHTTPMiddleware):
    """Middleware for Handling Authentication and Authorization"""

    async def dispatch(self, request: Request, call_next):
        # Set the current request context for tools to access
        set_current_request(request)

        # Only require authentication for actual MCP tool calls
        if (
            request.url.path == "/mcp"
            and request.method == "POST"
            and "application/json" in request.headers.get("content-type", "")
        ):
            # Extract and store the API key from request headers
            api_key = request.headers.get("X-API-KEY")

            if api_key:
                # Store API key in request state for tools to access
                request.state.api_key = api_key
                logger.info(f"API key stored: {api_key}")
            else:
                request.state.api_key = None
                logger.info("No valid X-API-KEY header found for MCP tool call")
                raise HTTPException(
                    status_code=401,
                    detail="X-API-KEY header required for MCP tool calls",
                )
        else:
            # For non-tool-call requests, just log but don't require authentication
            logger.info(
                f"Non-tool-call request to {request.url.path} ({request.method}) - no authentication required"
            )
            request.state.api_key = None

        # Continue with the request
        response = await call_next(request)
        return response

import logging
import requests
from typing import Dict, Optional, Tuple
from mcp.server.fastmcp import FastMCP
from core.middleware.auth import get_api_key

logger = logging.getLogger(__name__)

def get_coordinates_from_location(location: str) -> Optional[Tuple[float, float]]:
    """
    Get latitude and longitude coordinates from a location name using Open-Meteo's geocoding API.
    """
    try:
        # Use Open-Meteo's geocoding API
        geocoding_url = "https://geocoding-api.open-meteo.com/v1/search"
        params = {"name": location, "count": 1, "language": "en", "format": "json"}

        response = requests.get(geocoding_url, params=params, timeout=10)
        response.raise_for_status()

        data = response.json()

        if "results" in data and len(data["results"]) > 0:
            result = data["results"][0]
            latitude = result["latitude"]
            longitude = result["longitude"]
            logger.info(f"Found coordinates for '{location}': {latitude}, {longitude}")
            return latitude, longitude
        else:
            logger.info(f"No coordinates found for location: {location}")
            return None

    except Exception as e:
        logger.error(f"Error parsing geocoding response for '{location}': {e}")
        return None

def get_weather_by_coordinates(latitude: float, longitude: float) -> Optional[Dict]:
    """
    Get current weather data using latitude and longitude coordinates.
    """
    try:
        weather_url = "https://api.open-meteo.com/v1/forecast"
        params = {
            "latitude": latitude,
            "longitude": longitude,
            "current_weather": "true",
        }

        response = requests.get(weather_url, params=params, timeout=10)
        response.raise_for_status()

        data = response.json()
        logger.info(f"Weather data retrieved for coordinates {latitude}, {longitude}")
        return data

    except Exception as e:
        logger.error(f"Error fetching weather data: {e}")
        return None

def register_weather_tools(mcp: FastMCP):
    """Register weather-related tools with the MCP server"""

    @mcp.tool()
    def get_weather(location: str) -> Dict:
        """
        Get current weather for a location.

        Args:
            location: The location name (e.g., "London", "New York", "Tokyo")

        Returns:
            Dictionary containing weather information
        """
        # Get the current user's token
        api_key = get_api_key()
        logger.info(f"Tool 'get_weather' called for location: {location}")

        # Get coordinates from location name
        coordinates = get_coordinates_from_location(location)
        if not coordinates:
            return {
                "error": f"Could not find coordinates for location: {location}",
                "success": False,
            }

        latitude, longitude = coordinates

        # Get weather data using coordinates
        weather_data = get_weather_by_coordinates(latitude, longitude)
        if not weather_data:
            return {
                "error": f"Could not fetch weather data for {location}",
                "success": False,
            }

        # Extract current weather information
        current_weather = weather_data.get("current_weather", {})
        weather_units = weather_data.get("current_weather_units", {})

        return {
            "location": location,
            "coordinates": {"latitude": latitude, "longitude": longitude},
            "temperature": f"{current_weather.get('temperature')} {weather_units.get('temperature', '°C')}",
            "windspeed": f"{current_weather.get('windspeed')} {weather_units.get('windspeed', 'km/h')}",
            "winddirection": f"{current_weather.get('winddirection')} {weather_units.get('winddirection', '°')}",
            "weathercode": current_weather.get("weathercode"),
            "time": current_weather.get("time"),
            "success": True,
        }

import argparse
import logging
from mcp.server.fastmcp import FastMCP
import uvicorn

# Import core modules
from core.middleware.auth import AuthMiddleware
from core.tools.weather import register_weather_tools

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Create an MCP server
mcp = FastMCP("Weather MCP Server")

def register_all_components(mcp: FastMCP):
    """Register all MCP components"""
    register_weather_tools(mcp)
    logger.info("All MCP components registered successfully")

# Register components
register_all_components(mcp)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Run MCP server")
    parser.add_argument("--host", default="0.0.0.0", help="Host to bind to")
    parser.add_argument("--port", type=int, default=8000, help="Port to bind to")

    args = parser.parse_args()

    # Create the HTTP app and add middleware
    app = mcp.streamable_http_app()
    app.add_middleware(AuthMiddleware)

    # Run as HTTP server
    print(f"Starting MCP HTTP server on {args.host}:{args.port}")
    print("Authentication middleware enabled - Authorization header required")
    uvicorn.run(app, host=args.host, port=args.port)