from typing import Any
import httpx
import sys
from datetime import datetime
from mcp.server.fastmcp import FastMCP
# Initialize FastMCP server with the name "weather"
mcp = FastMCP("weather")
# Constants for API base URL and user agent header
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"
async def make_nws_request(url: str) -> dict[str, Any] | None:
"""Make a request to the NWS API with proper error handling.
Args:
url (str): The URL of the API endpoint to fetch data from.
Returns:
dict[str, Any] | None: The JSON response if the request is successful,
or None if an exception occurs during the request.
"""
headers = {
"User-Agent": USER_AGENT,
"Accept": "application/geo+json"
}
try:
async with httpx.AsyncClient() as client:
# Send a GET request to the NWS API
response = await client.get(url, headers=headers, timeout=30.0)
response.raise_for_status()
# Return the JSON response if successful
return response.json()
except Exception:
# Return None if an exception occurs during the request
return None
def format_alert(feature: dict) -> str:
"""Format an alert feature into a readable string.
Args:
feature (dict): A dictionary containing alert information from the NWS API.
Returns:
str: A formatted string representing the alert.
"""
props = feature["properties"]
return f"""
Event: {props.get('event', 'Unknown')}
Area: {props.get('areaDesc', 'Unknown')}
Severity: {props.get('severity', 'Unknown')}
Description: {props.get('description', 'No description available')}
Instructions: {props.get('instruction', 'No specific instructions provided')}
"""
@mcp.tool()
async def get_alerts(state: str) -> str:
"""Get weather alerts for a US state using the NWS API.
Args:
state (str): Two-letter US state code (e.g. CA, NY).
Returns:
str: A string containing formatted alert information or an error message if no alerts are found.
"""
url = f"{NWS_API_BASE}/alerts/active/area/{state}"
data = await make_nws_request(url)
if not data or "features" not in data:
return "Unable to fetch alerts or no alerts found."
if not data["features"]:
return "No active alerts for this state."
# Format each alert into a readable string
alerts = [format_alert(feature) for feature in data["features"]]
return "\n---\n".join(alerts)
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
"""Get weather forecast for a location using the NWS API.
Args:
latitude (float): Latitude of the location in decimal degrees (north positive).
longitude (float): Longitude of the location in decimal degrees (east positive; west negative).
Returns:
str: A formatted forecast string containing up to the next five forecast periods. If the NWS requests fail or
the expected data is not present, a short human-readable error message is returned instead.
Notes:
- The function performs network I/O and must be awaited.
- Under the hood it relies on helper code to perform HTTP requests; network errors or unexpected response shapes
may result in the returned error messages rather than raised exceptions.
- Temperature unit, wind speed/direction, and a detailed forecast description are included for each period.
"""
# Convert to float in case strings are passed
try:
lat = float(latitude)
lon = float(longitude)
except (ValueError, TypeError):
return "Invalid latitude or longitude values provided."
# Get the forecast grid endpoint for the given latitude and longitude
points_url = f"{NWS_API_BASE}/points/{lat},{lon}"
points_data = await make_nws_request(points_url)
if not points_data:
return "Unable to fetch forecast data for this location."
# Get the forecast URL from the points response
forecast_url = points_data["properties"]["forecast"]
forecast_data = await make_nws_request(forecast_url)
if not forecast_data:
return "Unable to fetch detailed forecast."
# Format the periods into a readable forecast
periods = forecast_data["properties"]["periods"]
forecasts = []
for period in periods[:5]: # Only show next 5 periods
forecast = f"""
{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""
forecasts.append(forecast)
return "\n---\n".join(forecasts)
def main():
# Initialize and run the FastMCP server
mcp.run(transport='stdio')
if __name__ == "__main__":
main()
