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()