"""
Weather MCP Server (STDIO Transport)
This module implements an MCP (Model Context Protocol) server that exposes
weather alerts from the National Weather Service API. It uses STDIO transport,
meaning it communicates via standard input/output when spawned as a subprocess.
When run directly, the server starts and waits for JSON-RPC messages on stdin,
responding with results on stdout. This is the standard transport for
local MCP server execution.
"""
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
# Initialize the FastMCP server instance with name "weather"
# FastMCP provides the MCP protocol framework and tool registration
mcp = FastMCP("weather")
# Constants for the National Weather Service API
# NWS API requires a User-Agent header and returns GeoJSON format
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 an HTTP GET request to the NWS API with proper headers and error handling.
Args:
url: The full URL to request (e.g., alerts endpoint for a state)
Returns:
Parsed JSON response as dict, or None if request fails
"""
# NWS API requires User-Agent and Accept headers per their terms of use
headers = {
"User-Agent": USER_AGENT,
"Accept": "application/geo+json"
}
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, headers=headers, timeout=30.0)
response.raise_for_status()
return response.json()
except Exception:
return None
def format_alert(feature: dict) -> str:
"""
Extract and format alert properties from an NWS GeoJSON feature into
a human-readable string.
Args:
feature: A GeoJSON feature object from the NWS alerts API
Returns:
Formatted string with event, area, severity, description, and instructions
"""
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:
"""
MCP Tool: Get active weather alerts for a US state.
This tool is exposed to MCP clients and can be invoked by AI agents.
It fetches from the NWS API and returns formatted alert information.
Args:
state: Two-letter US state code (e.g. CA, NY, TX)
Returns:
Formatted string of all active alerts, or a message if none found
"""
# Build the NWS API URL for active alerts in the given state
url = f"{NWS_API_BASE}/alerts/active/area/{state}"
data = await make_nws_request(url)
# Handle API errors or missing data
if not data or "features" not in data:
return "Unable to fetch alerts or no alerts found."
# Handle case when no alerts are active
if not data["features"]:
return "No active alerts for this state."
# Format each alert feature and join with separator
alerts = [format_alert(feature) for feature in data["features"]]
return "\n---\n".join(alerts)
@mcp.resource("echo://{message}")
def echo_resource(message: str) -> str:
"""
MCP Resource: Echo a message back (for testing resource URIs).
Resources are URI-addressable content that clients can read.
This URI template echo://{message} allows reading any echo://... URI.
"""
return f"Resource echo: {message}"
if __name__ == "__main__":
# Entry point: start the server with STDIO transport (default)
# This blocks and waits for JSON-RPC messages on stdin
mcp.run()
