server.py

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
from starlette.applications import Starlette
from mcp.server.sse import SseServerTransport
from starlette.requests import Request
from starlette.routing import Mount, Route
from mcp.server import Server
import uvicorn

# Initialize FastMCP server for Weather tools (SSE)
mcp = FastMCP("weather")

# Constants for NWS (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 a request to the NWS API with proper error handling.
    
    This function handles the HTTP request to the NWS API, setting appropriate
    headers and handling potential errors during the request.
    
    Args:
        url: The complete URL for the NWS API endpoint
        
    Returns:
        A dictionary containing the JSON response if successful, None otherwise
    """
    # Set required headers for the NWS API
    headers = {
        "User-Agent": USER_AGENT,  # NWS API requires a user agent
        "Accept": "application/geo+json"  # Request GeoJSON format
    }
    # Create an async HTTP client
    async with httpx.AsyncClient() as client:
        try:
            # Make the GET request with timeout
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()  # Raise exception for 4XX/5XX responses
            return response.json()  # Parse and return JSON response
        except Exception:
            # Return None if any error occurs (connection, timeout, parsing, etc.)
            return None


def format_alert(feature: dict) -> str:
    """Format an alert feature into a readable string.
    
    Extracts relevant information from a weather alert feature and formats it
    into a human-readable string.
    
    Args:
        feature: A dictionary containing a single weather alert feature
        
    Returns:
        A formatted string with key alert information
    """
    # Extract properties from the feature
    props = feature["properties"]
    # Format the alert with important details
    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.
    
    Fetches active weather alerts from the NWS API for a specified US state.
    
    Args:
        state: Two-letter US state code (e.g. CA, NY)
        
    Returns:
        A formatted string containing all active alerts for the state,
        or a message indicating no alerts or an error
    """
    # Construct the URL for the state's active alerts
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    # Make the API request
    data = await make_nws_request(url)

    # Check if the response is valid
    if not data or "features" not in data:
        return "Unable to fetch alerts or no alerts found."

    # Check if there are any active alerts
    if not data["features"]:
        return "No active alerts for this state."

    # Format each alert and join them with separators
    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.
    
    Fetches the weather forecast from the NWS API for a specified location
    using latitude and longitude coordinates.
    
    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
        
    Returns:
        A formatted string containing the forecast for the next 5 periods,
        or an error message if the forecast couldn't be retrieved
    """
    # First get the forecast grid endpoint using the coordinates
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    # Check if we received valid point data
    if not points_data:
        return "Unable to fetch forecast data for this location."

    # Extract the forecast URL from the points response
    # NWS API requires this two-step process to get the forecast
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)

    # Check if we received valid forecast data
    if not forecast_data:
        return "Unable to fetch detailed forecast."

    # Extract and format the forecast periods
    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)

    # Join all forecast periods with separators
    return "\n---\n".join(forecasts)


def create_starlette_app(mcp_server: Server, *, debug: bool = False) -> Starlette:
    """Create a Starlette application that can serve the provided MCP server with SSE.
    
    Sets up a Starlette web application with routes for SSE (Server-Sent Events)
    communication with the MCP server.
    
    Args:
        mcp_server: The MCP server instance to connect
        debug: Whether to enable debug mode for the Starlette app
        
    Returns:
        A configured Starlette application
    """
    # Create an SSE transport with a base path for messages
    sse = SseServerTransport("/messages/")

    async def handle_sse(request: Request) -> None:
        """Handler for SSE connections.
        
        Establishes an SSE connection and connects it to the MCP server.
        
        Args:
            request: The incoming HTTP request
        """
        # Connect the SSE transport to the request
        async with sse.connect_sse(
                request.scope,
                request.receive,
                request._send,  # noqa: SLF001
        ) as (read_stream, write_stream):
            # Run the MCP server with the SSE streams
            await mcp_server.run(
                read_stream,
                write_stream,
                mcp_server.create_initialization_options(),
            )

    # Create and return the Starlette application with routes
    return Starlette(
        debug=debug,
        routes=[
            Route("/sse", endpoint=handle_sse),  # Endpoint for SSE connections
            Mount("/messages/", app=sse.handle_post_message),  # Endpoint for posting messages
        ],
    )


if __name__ == "__main__":
    # Get the underlying MCP server from the FastMCP instance
    mcp_server = mcp._mcp_server  # noqa: WPS437
    
    import argparse
    
    # Set up command-line argument parsing
    parser = argparse.ArgumentParser(description='Run MCP server with configurable transport')
    # Allow choosing between stdio and SSE transport modes
    parser.add_argument('--transport', choices=['stdio', 'sse'], default='stdio', 
                        help='Transport mode (stdio or sse)')
    # Host configuration for SSE mode
    parser.add_argument('--host', default='0.0.0.0', 
                        help='Host to bind to (for SSE mode)')
    # Port configuration for SSE mode
    parser.add_argument('--port', type=int, default=8080, 
                        help='Port to listen on (for SSE mode)')
    args = parser.parse_args()

    # Launch the server with the selected transport mode
    if args.transport == 'stdio':
        # Run with stdio transport (default)
        # This mode communicates through standard input/output
        mcp.run(transport='stdio')
    else:
        # Run with SSE transport (web-based)
        # Create a Starlette app to serve the MCP server
        starlette_app = create_starlette_app(mcp_server, debug=True)
        # Start the web server with the configured host and port
        uvicorn.run(starlette_app, host=args.host, port=args.port)