codewithdhaval/mcp-server

Weather MCP Server Improvements

Overview

This document outlines the comprehensive improvements made to the weather.py file to create a robust, production-ready HTTP streamable MCP server.

Key Improvements

1. Proper Stream Management

• Before: Simple dictionary-based stream storage with no cleanup
• After: Dedicated StreamManager class with:
  • Automatic cleanup of expired streams
  • Proper stream lifecycle management
  • Thread-safe operations
  • Background cleanup task

class StreamManager:
    def __init__(self):
        self.active_streams: Dict[str, Dict[str, Any]] = {}
        self._cleanup_task: Optional[asyncio.Task] = None
    
    async def cleanup_expired_streams(self) -> None:
        # Automatic cleanup of expired streams

2. Enhanced Error Handling

• Before: Basic try-catch with print statements
• After: Comprehensive error handling with:
  • Structured logging
  • Proper HTTP status codes
  • Graceful degradation
  • Detailed error messages

async def fetch_weather_data(url: str) -> Union[Dict[str, Any], None]:
    try:
        # ... implementation
    except httpx.HTTPStatusError as e:
        logger.error(f"HTTP error fetching data from NWS API: {e}")
        return None
    except Exception as e:
        logger.error(f"Unexpected error fetching weather data: {e}")
        return None

3. Complete Type Annotations

• Before: Minimal type hints
• After: Full type annotations for:
  • Function parameters and return types
  • Class attributes
  • Async generators
  • Complex data structures

async def stream_weather_data(stream_id: str) -> AsyncGenerator[str, None]:
    # Properly typed async generator

4. Missing Function Implementations

• Before: Referenced undefined functions
• After: Complete implementations of:
  • get_current_conditions(): Fetches real-time weather data
  • weather_stream_endpoint(): SSE streaming endpoint
  • stop_stream_endpoint(): Stream management endpoint
  • list_streams_endpoint(): Stream listing endpoint

5. Resource Management

• Before: No cleanup mechanism
• After: Proper resource management with:
  • Background cleanup task
  • Application lifecycle management
  • Graceful shutdown handling

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    cleanup_task_instance = asyncio.create_task(cleanup_task())
    yield
    # Shutdown
    cleanup_task_instance.cancel()

6. Enhanced MCP Tools

• Before: Basic forecast tool only
• After: Complete tool suite:
  • get_forecast(): Detailed weather forecasts
  • start_weather_stream(): Stream initialization
  • stop_weather_stream(): Stream termination
  • list_weather_streams(): Stream management

7. Improved HTTP Endpoints

• Before: Incomplete endpoint definitions
• After: Full REST API with:
  • Health check endpoint
  • Stream management endpoints
  • Proper HTTP status codes
  • CORS headers for SSE

@app.get("/health")
def health():
    return {
        "message": "MCP SSE server is running!",
        "active_streams": len(stream_manager.active_streams),
        "timestamp": datetime.now().isoformat()
    }

8. Better Logging

• Before: Print statements
• After: Structured logging with:
  • Configurable log levels
  • Contextual information
  • Error tracking
  • Performance monitoring

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("weather-mcp")

New Features

1. Real-time Weather Streaming

• SSE-based streaming of current weather conditions
• Configurable update intervals (30 seconds default)
• Automatic stream expiration
• Error recovery mechanisms

2. Stream Management API

# Start a stream
POST /weather/streams
{
    "latitude": 40.7128,
    "longitude": -74.0060,
    "duration_minutes": 60
}

# List active streams
GET /weather/streams

# Stop a stream
POST /weather/stream/{stream_id}/stop

# Stream data
GET /weather/stream/{stream_id}

3. Enhanced Weather Data

• Current conditions with detailed metrics
• Temperature, wind, humidity, and descriptions
• Proper unit handling
• Timestamp information

4. Production-Ready Features

• Health check endpoint
• Graceful error handling
• Resource cleanup
• Performance monitoring
• Scalable architecture

Usage Examples

Starting the Server

python weather.py

Testing the Server

python test_weather_stream.py

Using MCP Tools

# Get forecast
await get_forecast(latitude=40.7128, longitude=-74.0060)

# Start streaming
await start_weather_stream(latitude=40.7128, longitude=-74.0060, duration_minutes=60)

# List streams
await list_weather_streams()

# Stop stream
await stop_weather_stream(stream_id="uuid-here")

SSE Streaming

// Client-side SSE connection
const eventSource = new EventSource('/weather/stream/stream-id');
eventSource.onmessage = function(event) {
    const data = JSON.parse(event.data);
    console.log('Weather update:', data);
};

Architecture Benefits

1. Scalability

• Efficient stream management
• Background cleanup tasks
• Resource-aware operations

2. Reliability

• Comprehensive error handling
• Graceful degradation
• Automatic recovery

3. Maintainability

• Clear separation of concerns
• Well-documented code
• Type safety

4. Performance

• Async/await throughout
• Efficient HTTP client usage
• Minimal resource overhead

Testing

The improved server includes:

• Unit test examples
• Integration test scenarios
• Error condition testing
• Performance benchmarks

Future Enhancements

1. Caching Layer: Redis-based caching for weather data
2. Rate Limiting: API rate limiting and throttling
3. Authentication: JWT-based authentication
4. Metrics: Prometheus metrics collection
5. WebSocket Support: Real-time bidirectional communication
6. Multiple Weather Providers: Fallback weather data sources

Conclusion

The improved weather.py file provides a production-ready, scalable, and maintainable MCP server with comprehensive streaming capabilities, proper error handling, and a complete API surface for weather data access.