tools/weather.py

from mcp.server.fastmcp import FastMCP
import requests
import os
from typing import Dict

mcp = FastMCP("Weather")

@mcp.tool()
def get_current_weather(location: str) -> Dict:
    """Get current weather conditions for a specific location.
    
    Args:
        location (str): The city name, state, and/or country. Can be in various formats:
                       - "London" (city only)
                       - "New York, NY" (city, state)
                       - "Tokyo, Japan" (city, country)
                       - "Paris, France" (city, country)
                       - "Miami, FL, USA" (city, state, country)
    
    Returns:
        Dict: Weather information containing:
            - location (str): Confirmed location name from API
            - temperature (str): Current temperature with unit (e.g., "22°C")
            - condition (str): Weather description (e.g., "clear sky", "light rain")
            - humidity (str): Humidity percentage (e.g., "65%")
            - wind_speed (str): Wind speed with unit (e.g., "3.2 m/s")
            - pressure (str): Atmospheric pressure (e.g., "1013 hPa")
            - status (str): "real_data" for API data, "mock_data" for fallback, "error" for failures
            - message (str, optional): Additional information or error details
    
    Examples:
        get_current_weather("London") returns real weather data for London, UK
        get_current_weather("Miami, FL") returns weather for Miami, Florida
        get_current_weather("InvalidCity") returns error status with message
    
    Note:
        Requires OPENWEATHER_API_KEY environment variable for real data.
        Returns mock data if API key is not provided.
    """
    api_key = os.getenv("OPENWEATHER_API_KEY")
    
    if not api_key:
        # Return mock data if no API key
        return {
            "location": location,
            "temperature": "22°C",
            "condition": "sunny",
            "humidity": "65%",
            "wind_speed": "10 km/h",
            "pressure": "1013 hPa",
            "status": "mock_data",
            "message": f"Mock weather data for {location} (set OPENWEATHER_API_KEY for real data)"
        }
    
    try:
        url = f"http://api.openweathermap.org/data/2.5/weather"
        params = {
            "q": location,
            "appid": api_key,
            "units": "metric"
        }
        
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        
        return {
            "location": f"{data['name']}, {data['sys']['country']}",
            "temperature": f"{data['main']['temp']}°C",
            "condition": data["weather"][0]["description"],
            "humidity": f"{data['main']['humidity']}%",
            "wind_speed": f"{data['wind']['speed']} m/s",
            "pressure": f"{data['main']['pressure']} hPa",
            "feels_like": f"{data['main']['feels_like']}°C",
            "status": "real_data"
        }
    except requests.exceptions.RequestException as e:
        return {
            "error": f"Failed to fetch weather data: {str(e)}",
            "location": location,
            "status": "error",
            "message": "Check if location name is correct and API key is valid"
        }

@mcp.tool()
def get_forecast(location: str, days: int = 5) -> Dict:
    """Get weather forecast for a specific location over multiple days.
    
    Args:
        location (str): The city name, state, and/or country (same format as get_current_weather).
        days (int, optional): Number of forecast days to retrieve. Must be between 1-5. 
                              Defaults to 5. Values outside range are clamped to 1-5.
    
    Returns:
        Dict: Forecast information containing:
            - location (str): Confirmed location name
            - forecast_days (int): Actual number of forecast days returned
            - forecast (List[Dict]): List of daily forecasts, each containing:
                - day (int): Day number (1, 2, 3, etc.)
                - temperature (str): Predicted temperature (e.g., "24°C")
                - condition (str): Weather condition description
                - humidity (str): Predicted humidity percentage
            - status (str): "real_data", "mock_data", or "error"
            - message (str, optional): Additional information
    
    Examples:
        get_forecast("London", 3) returns 3-day forecast for London
        get_forecast("New York") returns 5-day forecast (default) for New York
        get_forecast("Paris", 10) returns 5-day forecast (clamped maximum)
    
    Note:
        Requires OPENWEATHER_API_KEY environment variable for real data.
        Returns mock forecast if API key is not provided.
    """
    if days < 1 or days > 5:
        days = max(1, min(5, days))  # Clamp to 1-5 range
        
    api_key = os.getenv("OPENWEATHER_API_KEY")
    
    if not api_key:
        # Return mock forecast
        mock_forecast = []
        conditions = ["sunny", "partly cloudy", "cloudy", "light rain", "clear sky"]
        base_temp = 20
        
        for day in range(days):
            mock_forecast.append({
                "day": day + 1,
                "temperature": f"{base_temp + day}°C",
                "condition": conditions[day % len(conditions)],
                "humidity": f"{60 + day * 2}%"
            })
        
        return {
            "location": location,
            "forecast_days": days,
            "forecast": mock_forecast,
            "status": "mock_data",
            "message": f"Mock {days}-day forecast for {location}"
        }
    
    try:
        url = f"http://api.openweathermap.org/data/2.5/forecast"
        params = {
            "q": location,
            "appid": api_key,
            "units": "metric",
            "cnt": days * 8  # 8 forecasts per day (3-hour intervals)
        }
        
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        
        # Process forecast data (take midday forecast for each day)
        forecast = []
        processed_days = set()
        
        for item in data["list"]:
            # Extract date from timestamp
            day_date = item["dt_txt"].split()[0]  # Get YYYY-MM-DD part
            
            if day_date not in processed_days and len(forecast) < days:
                processed_days.add(day_date)
                forecast.append({
                    "day": len(forecast) + 1,
                    "date": day_date,
                    "temperature": f"{item['main']['temp']}°C",
                    "condition": item["weather"][0]["description"],
                    "humidity": f"{item['main']['humidity']}%"
                })
        
        return {
            "location": f"{data['city']['name']}, {data['city']['country']}",
            "forecast_days": len(forecast),
            "forecast": forecast,
            "status": "real_data"
        }
    except requests.exceptions.RequestException as e:
        return {
            "error": f"Failed to fetch forecast data: {str(e)}",
            "location": location,
            "status": "error",
            "message": "Check if location name is correct and API key is valid"
        }

if __name__ == "__main__":
    mcp.run()