"""
Weather MCP Server (SSE Transport)
This module implements the same weather MCP server as weather.py but uses
SSE (Server-Sent Events) transport over HTTP instead of STDIO. The server
runs as an HTTP server on a configurable host/port.
Clients connect via HTTP GET to the /sse endpoint and POST to /messages/
for sending requests. This transport is useful when the server needs to run
as a standalone process (e.g., behind a reverse proxy) or when multiple
clients need to connect to the same server.
"""
from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
# Initialize FastMCP with explicit host and port for HTTP binding
# host="127.0.0.1" restricts to localhost; port=8000 is the default
mcp = FastMCP("weather", host="127.0.0.1", port=8000)
# Constants for the National Weather Service API
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
"""
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
"""
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."
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.
"""
return f"Resource echo: {message}"
if __name__ == "__main__":
# Entry point: start the server with SSE transport.
# This launches an HTTP server (uvicorn) listening on host:port.
# Endpoints: GET /sse (SSE stream), POST /messages/ (client messages)
mcp.run(transport="sse")
