#!/usr/bin/env python3
"""
Weather MCP Server
Provides tools to fetch weather data from a weather API.
"""
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
from typing import Optional
import httpx
import os
import json
# Initialize the MCP server
mcp = FastMCP("weather_mcp")
# Define input validation model
class WeatherQuery(BaseModel):
"""Input parameters for weather lookup."""
location: str = Field(
description="City name or location to get weather for (e.g., 'London', 'New York')",
min_length=1,
max_length=100
)
units: Optional[str] = Field(
default="metric",
description="Temperature units: 'metric' (Celsius), 'imperial' (Fahrenheit), or 'kelvin'",
pattern="^(metric|imperial|kelvin)$"
)
@mcp.tool(
name="weather_get_current",
annotations={
"title": "Get Current Weather",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def get_current_weather(params: WeatherQuery) -> str:
"""Get current weather conditions for a location.
Returns current temperature, conditions, humidity, and wind speed.
Requires WEATHER_API_KEY environment variable to be set.
Example response:
Location: London, UK
Temperature: 15°C
Conditions: Partly cloudy
Humidity: 72%
Wind: 12 km/h NW
Errors:
- Returns "Error: Missing API key" if WEATHER_API_KEY not set
- Returns "Error: Location not found" if location is invalid
- Returns "Error: API request failed" for network/API errors
"""
# Validate environment
api_key = os.getenv("WEATHER_API_KEY")
if not api_key:
return "Error: Missing API key. Please set WEATHER_API_KEY environment variable."
try:
# Make API request
async with httpx.AsyncClient() as client:
response = await client.get(
"https://api.weatherapi.com/v1/current.json",
params={
"key": api_key,
"q": params.location,
"aqi": "no"
},
timeout=10.0
)
# Handle different status codes
if response.status_code == 401:
return "Error: Invalid API key. Please check your WEATHER_API_KEY."
elif response.status_code == 400:
return f"Error: Location '{params.location}' not found. Please check the spelling."
elif response.status_code != 200:
return f"Error: API request failed with status {response.status_code}"
# Parse and format response
data = response.json()
location = data["location"]
current = data["current"]
# Convert temperature if needed
temp = current["temp_c"]
unit = "°C"
if params.units == "imperial":
temp = current["temp_f"]
unit = "°F"
elif params.units == "kelvin":
temp = current["temp_c"] + 273.15
unit = "K"
# Format output
result = f"""Location: {location['name']}, {location['country']}
Temperature: {temp}{unit}
Conditions: {current['condition']['text']}
Humidity: {current['humidity']}%
Wind: {current['wind_kph']} km/h {current['wind_dir']}"""
return result
except httpx.TimeoutException:
return "Error: Request timed out. The weather API is not responding."
except httpx.RequestError as e:
return f"Error: Network error occurred: {str(e)}"
except KeyError as e:
return f"Error: Unexpected API response format (missing field: {e})"
except Exception as e:
return f"Error: An unexpected error occurred: {str(e)}"
# Run the server
if __name__ == "__main__":
mcp.run()
