server.py

import json

from fastmcp import FastMCP
from weather_helper import WeatherManager

# Initialize MCP server and weather manager
mcp = FastMCP("Weather Forecast MCP Server", debug=True)
weather_manager = WeatherManager()

@mcp.tool(description="Get 3-day weather forecast for a location using postal/zip code or city name")
async def get_weather_forecast(location: str) -> dict:
    """
    Fetch 3-day weather forecast for the given location.

    Args:
        location: Postal code, zip code, or city name (e.g., "10001", "London", "New York")

    Returns:
        Dictionary containing location info, current weather, and 3-day forecast
    """
    weather_data = weather_manager.get_data(location)

    if 'error' in weather_data:
        return weather_data

    return {
        "success": True,
        "data": weather_data
    }

@mcp.tool(description="Get saved weather data from files in the data directory")
async def get_saved_weather() -> dict:
    """
    Retrieve all previously saved weather forecasts from the data directory.

    Returns:
        Dictionary containing all saved weather data
    """
    weathers = weather_manager.load_weather()
    return {
        "success": True,
        "count": len(weathers),
        "data": weathers
    }

@mcp.prompt(name="Weather Forecast Workflow")
def weather_prompt(location: str) -> str:
    """Get Weather Forecast Workflow"""
    return f"""Weather Forecast Request: {location}

WORKFLOW:
1. get_weather_forecast(location="{location}") - Fetch 3-day weather forecast
2. The data will be automatically saved to the data directory

AVAILABLE TOOLS:
- get_weather_forecast: Get 3-day forecast for any location
- get_saved_weather: View all previously saved weather data
- weather_status: Check status of saved weather locations

GOAL: Provide comprehensive 3-day weather forecast with current conditions"""

@mcp.resource("weather://status")
def weather_status() -> str:
    """Current weather data status and saved locations"""
    summary = weather_manager.get_status()
    return json.dumps(summary, indent=2)

if __name__ == "__main__":
    mcp.run()