server.py•8.21 kB
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)