Weather MCP Server

"""MCP Weather Server - Provides weather tools, resources, and prompts.""" import asyncio import json import sys from typing import Any, Sequence from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import ( Tool, Resource, Prompt, TextContent, ImageContent, EmbeddedResource, PromptMessage, ) from weather_api import WeatherAPI # Initialize the weather API client weather_api = WeatherAPI() # Initialize the MCP server app = Server("weather-mcp-server") @app.list_tools() async def list_tools() -> list[Tool]: """List available weather tools.""" return [ Tool( name="get_current_weather", description="Get current weather conditions for a specific location", inputSchema={ "type": "object", "properties": { "location": { "type": "string", "description": "City name or location (e.g., 'New York', 'London')" } }, "required": ["location"] } ), Tool( name="get_weather_forecast", description="Get weather forecast for a specific location for multiple days", inputSchema={ "type": "object", "properties": { "location": { "type": "string", "description": "City name or location (e.g., 'New York', 'London')" }, "days": { "type": "integer", "description": "Number of days to forecast (1-5)", "minimum": 1, "maximum": 5, "default": 5 } }, "required": ["location"] } ), Tool( name="search_locations", description="Search for location names matching a query string", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "Search query for location name" } }, "required": ["query"] } ), ] @app.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent | ImageContent | EmbeddedResource]: """Handle tool calls.""" if name == "get_current_weather": location = arguments.get("location", "") if not location: return [TextContent( type="text", text="Error: location parameter is required" )] weather_data = await weather_api.get_current_weather(location) return [TextContent( type="text", text=json.dumps(weather_data, indent=2) )] elif name == "get_weather_forecast": location = arguments.get("location", "") days = arguments.get("days", 5) if not location: return [TextContent( type="text", text="Error: location parameter is required" )] forecast_data = await weather_api.get_forecast(location, days) return [TextContent( type="text", text=json.dumps(forecast_data, indent=2) )] elif name == "search_locations": query = arguments.get("query", "") if not query: return [TextContent( type="text", text="Error: query parameter is required" )] locations = await weather_api.search_locations(query) return [TextContent( type="text", text=json.dumps(locations, indent=2) )] else: return [TextContent( type="text", text=f"Error: Unknown tool '{name}'" )] @app.list_resources() async def list_resources() -> list[Resource]: """List available weather resources.""" return [ Resource( uri="weather://current", name="Current Weather", description="Current weather data for any location", mimeType="application/json" ), Resource( uri="weather://forecast", name="Weather Forecast", description="Weather forecast data for any location", mimeType="application/json" ), ] @app.read_resource() async def read_resource(uri: str) -> str: """Read weather resource data.""" if uri.startswith("weather://current/"): location = uri.replace("weather://current/", "") if not location: return json.dumps({"error": "Location required in URI: weather://current/{location}"}) weather_data = await weather_api.get_current_weather(location) return json.dumps(weather_data, indent=2) elif uri.startswith("weather://forecast/"): location = uri.replace("weather://forecast/", "") if not location: return json.dumps({"error": "Location required in URI: weather://forecast/{location}"}) forecast_data = await weather_api.get_forecast(location) return json.dumps(forecast_data, indent=2) elif uri == "weather://current" or uri == "weather://forecast": return json.dumps({ "error": "Location required", "usage": "Use weather://current/{location} or weather://forecast/{location}" }) else: return json.dumps({"error": f"Unknown resource URI: {uri}"}) @app.list_prompts() async def list_prompts() -> list[Prompt]: """List available weather prompts.""" return [ Prompt( name="analyze_weather", description="Analyze weather patterns and provide insights about current conditions", arguments=[ { "name": "location", "description": "Location to analyze weather for", "required": True } ] ), Prompt( name="compare_locations", description="Compare weather conditions between multiple locations", arguments=[ { "name": "locations", "description": "Comma-separated list of locations to compare", "required": True } ] ), ] @app.get_prompt() async def get_prompt(name: str, arguments: dict) -> Prompt: """Get prompt template with injected arguments.""" if name == "analyze_weather": location = arguments.get("location", "the specified location") prompt_text = f"""Analyze the weather conditions for {location}. Please provide: 1. Current weather summary 2. Temperature analysis (is it hot, cold, or moderate?) 3. Weather conditions (clear, cloudy, rainy, etc.) 4. Wind conditions 5. Overall comfort level and recommendations 6. Any notable weather patterns or warnings Use the get_current_weather tool to fetch the latest data for {location}.""" return Prompt( name="analyze_weather", description="Analyze weather patterns and provide insights", arguments=[], messages=[ PromptMessage( role="user", content=[ TextContent( type="text", text=prompt_text ) ] ) ] ) elif name == "compare_locations": locations_str = arguments.get("locations", "") locations = [loc.strip() for loc in locations_str.split(",") if loc.strip()] if not locations: locations = ["Location 1", "Location 2"] locations_list = ", ".join(locations) prompt_text = f"""Compare weather conditions between the following locations: {locations_list} Please provide: 1. Side-by-side comparison of current temperatures 2. Weather condition differences 3. Humidity and wind comparisons 4. Which location has more favorable weather 5. Recommendations based on weather preferences Use the get_current_weather tool for each location to fetch the latest data.""" return Prompt( name="compare_locations", description="Compare weather between locations", arguments=[], messages=[ PromptMessage( role="user", content=[ TextContent( type="text", text=prompt_text ) ] ) ] ) else: raise ValueError(f"Unknown prompt: {name}") async def main(): """Run the MCP server.""" async with stdio_server() as (read_stream, write_stream): await app.run( read_stream, write_stream, app.create_initialization_options() ) if __name__ == "__main__": asyncio.run(main())