# Directory Structure
```
├── .github
│ └── workflows
│ └── publish-to-pypi.yml
├── .gitignore
├── Dockerfile
├── glama.json
├── LICENSE
├── pyproject.toml
├── pytest.ini
├── README.md
├── requirements.txt
├── smithery.yaml
├── src
│ └── mcp_weather_server
│ ├── __init__.py
│ ├── __main__.py
│ ├── server.py
│ ├── tools
│ │ ├── __init__.py
│ │ ├── toolhandler.py
│ │ ├── tools_time.py
│ │ ├── tools_weather.py
│ │ └── weather_service.py
│ └── utils.py
├── tests
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_integration.py
│ ├── test_mcp_weather_server.py
│ ├── test_performance.py
│ ├── test_server.py
│ ├── test_time_tools.py
│ ├── test_utils.py
│ ├── test_weather_service.py
│ └── test_weather_tools.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
__pycache__/
*.pyc
venv/
.pytest_cache/
node_modules
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
[](https://smithery.ai/server/@isdaniel/mcp_weather_server)
[](https://pypi.org/project/mcp-weather-server/)
[](https://pypi.org/project/mcp-weather-server/)
<a href="https://glama.ai/mcp/servers/@isdaniel/mcp_weather_server">
<img width="380" height="200" src="https://glama.ai/mcp/servers/@isdaniel/mcp_weather_server/badge" />
</a>
# Weather MCP Server
A Model Context Protocol (MCP) server that provides weather information using the Open-Meteo API. This server supports both standard MCP communication and HTTP Server-Sent Events (SSE) for web-based integration.
## Features
* Get current weather information for a specified city
* Get weather data for a date range
* Get current date/time in any timezone
* Convert time between timezones
* Get timezone information
* HTTP SSE (Server-Sent Events) support for web applications
* RESTful API endpoints via Starlette/FastAPI integration
## Installation
### Standard Installation (for MCP clients like Claude Desktop)
This package can be installed using pip:
```bash
pip install mcp_weather_server
```
### Manual Configuration for MCP Clients
This server is designed to be installed manually by adding its configuration to the `cline_mcp_settings.json` file.
1. Add the following entry to the `mcpServers` object in your `cline_mcp_settings.json` file:
```json
{
"mcpServers": {
"weather": {
"command": "python",
"args": [
"-m",
"mcp_weather_server"
],
"disabled": false,
"autoApprove": []
}
}
}
```
2. Save the `cline_mcp_settings.json` file.
### SSE Server Installation (for web applications)
For HTTP SSE support, you'll need additional dependencies:
```bash
pip install mcp_weather_server starlette uvicorn
```
## Server Modes
This MCP server supports both **stdio** and **SSE** modes in a single unified server:
### 1. Standard MCP Mode (Default)
The standard mode communicates via stdio and is compatible with MCP clients like Claude Desktop.
```bash
# Default mode (stdio)
python -m mcp_weather_server
# Explicitly specify stdio mode
python -m mcp_weather_server.server --mode stdio
```
### 2. HTTP SSE Mode (Web Applications)
The SSE mode runs an HTTP server that provides MCP functionality via Server-Sent Events, making it accessible to web applications.
```bash
# Start SSE server on default host/port (0.0.0.0:8080)
python -m mcp_weather_server.server --mode sse
# Specify custom host and port
python -m mcp_weather_server.server --mode sse --host localhost --port 3000
# Enable debug mode
python -m mcp_weather_server.server --mode sse --debug
```
**Command Line Options:**
```
--mode {stdio,sse} Server mode: stdio (default) or sse
--host HOST Host to bind to (SSE mode only, default: 0.0.0.0)
--port PORT Port to listen on (SSE mode only, default: 8080)
--debug Enable debug mode
```
**SSE Endpoints:**
- `GET /sse` - SSE endpoint for MCP communication
- `POST /messages/` - Message endpoint for sending MCP requests
**Example SSE Usage:**
```javascript
// Connect to SSE endpoint
const eventSource = new EventSource('http://localhost:8080/sse');
// Send MCP tool request
fetch('http://localhost:8080/messages/', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
type: 'tool_call',
tool: 'get_weather',
arguments: { city: 'Tokyo' }
})
});
```
## Configuration
This server does not require an API key. It uses the Open-Meteo API, which is free and open-source.
## Usage
This server provides several tools for weather and time-related operations:
### Available Tools
1. **`get_current_weather`** - Get current weather for a city
2. **`get_weather_by_datetime_range`** - Get weather data for a date range
3. **`get_current_datetime`** - Get current time in any timezone
4. **`get_timezone_info`** - Get timezone information
5. **`convert_time`** - Convert time between timezones
### Tool Details
#### `get_current_weather`
Retrieves the current weather information for a given city.
**Parameters:**
- `city` (string, required): The name of the city (English names only)
**Returns:** JSON formatted weather data with current temperature and conditions
**Example:**
```json
{
"city": "Taipei",
"weather": "Partly cloudy",
"temperature_celsius": 25
}
```
#### `get_weather_by_datetime_range`
Retrieves weather information for a specified city between start and end dates.
**Parameters:**
- `city` (string, required): The name of the city (English names only)
- `start_date` (string, required): Start date in format YYYY-MM-DD (ISO 8601)
- `end_date` (string, required): End date in format YYYY-MM-DD (ISO 8601)
**Returns:** JSON array with daily weather summaries
**Example:**
```json
[
{
"date": "2024-01-01",
"day_of_week": "Monday",
"city": "London",
"weather": "Light rain",
"temperature_celsius": 8
}
]
```
#### `get_current_datetime`
Retrieves the current time in a specified timezone.
**Parameters:**
- `timezone_name` (string, required): IANA timezone name (e.g., 'America/New_York', 'Europe/London'). Use UTC if no timezone provided.
**Returns:** Current date and time in the specified timezone
**Example:**
```json
{
"timezone": "America/New_York",
"current_time": "2024-01-15T14:30:00-05:00",
"utc_time": "2024-01-15T19:30:00Z"
}
```
#### `get_timezone_info`
Get information about a specific timezone.
**Parameters:**
- `timezone_name` (string, required): IANA timezone name
**Returns:** Timezone details including offset and DST information
#### `convert_time`
Convert time from one timezone to another.
**Parameters:**
- `time_str` (string, required): Time to convert (ISO format)
- `from_timezone` (string, required): Source timezone
- `to_timezone` (string, required): Target timezone
**Returns:** Converted time in target timezone
## MCP Client Usage Examples
### Using with Claude Desktop or MCP Clients
```xml
<use_mcp_tool>
<server_name>weather</server_name>
<tool_name>get_current_weather</tool_name>
<arguments>
{
"city": "Tokyo"
}
</arguments>
</use_mcp_tool>
```
```xml
<use_mcp_tool>
<server_name>weather</server_name>
<tool_name>get_weather_by_datetime_range</tool_name>
<arguments>
{
"city": "Paris",
"start_date": "2024-01-01",
"end_date": "2024-01-07"
}
</arguments>
</use_mcp_tool>
```
```xml
<use_mcp_tool>
<server_name>weather</server_name>
<tool_name>get_current_datetime</tool_name>
<arguments>
{
"timezone_name": "Europe/Paris"
}
</arguments>
</use_mcp_tool>
```
## Web Integration (SSE Mode)
When running in SSE mode, you can integrate the weather server with web applications:
### HTML/JavaScript Example
```html
<!DOCTYPE html>
<html>
<head>
<title>Weather MCP Client</title>
</head>
<body>
<div id="weather-data"></div>
<script>
// Connect to SSE endpoint
const eventSource = new EventSource('http://localhost:8080/sse');
eventSource.onmessage = function(event) {
const data = JSON.parse(event.data);
document.getElementById('weather-data').innerHTML = JSON.stringify(data, null, 2);
};
// Function to get weather
async function getWeather(city) {
const response = await fetch('http://localhost:8080/messages/', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
jsonrpc: '2.0',
method: 'tools/call',
params: {
name: 'get_current_weather',
arguments: { city: city }
},
id: 1
})
});
}
// Example: Get weather for Tokyo
getWeather('Tokyo');
</script>
</body>
</html>
```
## Development
### Project Structure
```
mcp_weather_server/
├── src/
│ └── mcp_weather_server/
│ ├── __init__.py
│ ├── __main__.py # Main MCP server entry point
│ ├── server.py # Standard MCP server implementation
│ ├── server-see.py # SSE server implementation (NEW)
│ ├── utils.py # Utility functions
│ └── tools/ # Tool implementations
│ ├── __init__.py
│ ├── toolhandler.py # Base tool handler
│ ├── tools_weather.py # Weather-related tools
│ ├── tools_time.py # Time-related tools
│ └── weather_service.py # Weather API service
├── tests/
├── pyproject.toml
├── requirements.txt
└── README.md
```
### Running for Development
#### Standard MCP Mode
```bash
# From project root
python -m mcp_weather_server
# Or with PYTHONPATH
export PYTHONPATH="/path/to/mcp_weather_server/src"
python -m mcp_weather_server
```
#### SSE Server Mode
```bash
# From project root
python src/mcp_weather_server/server-see.py --host 0.0.0.0 --port 8080
# With custom host/port
python src/mcp_weather_server/server-see.py --host localhost --port 3000
```
### Adding New Tools
To add new weather or time-related tools:
1. Create a new tool handler in the appropriate file under `tools/`
2. Inherit from the `ToolHandler` base class
3. Implement the required methods (`get_name`, `get_description`, `call`)
4. Register the tool in `server.py`
## Dependencies
### Core Dependencies
- `mcp>=1.0.0` - Model Context Protocol implementation
- `httpx>=0.28.1` - HTTP client for API requests
- `python-dateutil>=2.8.2` - Date/time parsing utilities
### SSE Server Dependencies
- `starlette` - ASGI web framework
- `uvicorn` - ASGI server
### Development Dependencies
- `pytest` - Testing framework
## API Data Source
This server uses the [Open-Meteo API](https://open-meteo.com/), which is:
- Free and open-source
- No API key required
- Provides accurate weather forecasts
- Supports global locations
- Historical and current weather data
## Troubleshooting
### Common Issues
**1. City not found**
- Ensure city names are in English
- Try using the full city name or include country (e.g., "Paris, France")
- Check spelling of city names
**2. SSE Server not accessible**
- Verify the server is running: `python src/mcp_weather_server/server-see.py`
- Check firewall settings for the specified port
- Ensure all dependencies are installed: `pip install starlette uvicorn`
**3. MCP Client connection issues**
- Verify Python path in MCP client configuration
- Check that `mcp_weather_server` package is installed
- Ensure Python environment has required dependencies
**4. Date format errors**
- Use ISO 8601 format for dates: YYYY-MM-DD
- Ensure start_date is before end_date
- Check that dates are not too far in the future
### Error Responses
The server returns structured error messages:
```json
{
"error": "Could not retrieve coordinates for InvalidCity."
}
```
## Changelog
### v0.2.1 (Current)
- Added HTTP SSE (Server-Sent Events) support
- Added timezone conversion tools
- Enhanced weather data formatting
- Improved error handling
- Added comprehensive documentation
### Previous Versions
- v0.2.0: Added date range weather queries
- v0.1.0: Initial release with basic weather functionality
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
httpx
mcp
python-dateutil
```
--------------------------------------------------------------------------------
/src/mcp_weather_server/__main__.py:
--------------------------------------------------------------------------------
```python
import asyncio
from .server import main
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/glama.json:
--------------------------------------------------------------------------------
```json
{
"$schema": "https://glama.ai/mcp/schemas/server.json",
"maintainers": [
"isdaniel"
]
}
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
runtime: python
startCommand:
type: stdio
configSchema:
type: object
commandFunction: |
config => ({ command: "uv", args: ["run", "mcp_weather_server"] })
```
--------------------------------------------------------------------------------
/src/mcp_weather_server/tools/__init__.py:
--------------------------------------------------------------------------------
```python
"""Tools package for MCP Weather Server."""
from .toolhandler import *
from .tools_time import *
from .tools_weather import *
from .weather_service import *
__all__ = []
```
--------------------------------------------------------------------------------
/src/mcp_weather_server/__init__.py:
--------------------------------------------------------------------------------
```python
from .server import main as async_main
import asyncio
def main():
"""Synchronous entry point for the package."""
asyncio.run(async_main())
__all__ = ['main', 'async_main']
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim
# Set the working directory
WORKDIR /app
# Copy the current directory contents into the container at /app
COPY . /app
RUN (uv venv .venv) && (. .venv/bin/activate) && (uv pip install mcp_weather_server)
CMD ["uv","run","mcp_weather_server"]
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "mcp_weather_server"
version = "0.3.1"
authors = [
{ name = "danielshih", email = "[email protected]" },
]
description = "An MCP server for weather information"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"mcp>=1.0.0",
"argparse>=1.4.0",
"httpx>=0.28.1",
"mcp[cli]>=1.2.1",
"python-dotenv>=1.0.1",
"python-dateutil>=2.8.2"
]
[tool.hatch.build.targets.wheel]
packages = ["src/mcp_weather_server"]
[project.scripts]
mcp_weather_server = "mcp_weather_server:main"
[project.urls]
Source = "https://github.com/isdaniel/mcp_weather_server"
```
--------------------------------------------------------------------------------
/pytest.ini:
--------------------------------------------------------------------------------
```
[pytest]
minversion = 6.0
addopts = -ra --strict-markers --strict-config --tb=short -v
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
markers =
asyncio: marks tests as async
slow: marks tests as slow (deselect with '-m "not slow"')
integration: marks tests as integration tests
unit: marks tests as unit tests
performance: marks tests as performance tests
# Async test configuration
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function
# Filter warnings
filterwarnings =
ignore::DeprecationWarning
ignore::PendingDeprecationWarning
# Test output
console_output_style = progress
log_cli = true
log_cli_level = INFO
log_cli_format = %(asctime)s [%(levelname)8s] %(message)s
log_cli_date_format = %Y-%m-%d %H:%M:%S
```
--------------------------------------------------------------------------------
/.github/workflows/publish-to-pypi.yml:
--------------------------------------------------------------------------------
```yaml
name: Publish to PyPI
on:
push:
branches:
- main # Trigger on pushes to the main branch (adjust as needed)
tags:
- "v*" # or trigger when create new tag with "v" prefix
jobs:
build-and-publish:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.x' # Use a suitable Python version
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install build
- name: Build package
run: python -m build
- name: Extract version from pyproject.toml
run: |
VERSION=$(grep -oP '(?<=version = ").*(?=")' pyproject.toml)
echo "VERSION=$VERSION" >> $GITHUB_ENV
- name: Check if version exists on PyPI
id: check_version
run: |
PACKAGE_NAME="mcp_weather_server"
if curl -sSf "https://pypi.org/pypi/${PACKAGE_NAME}/${VERSION}/json" > /dev/null; then
echo "Version ${VERSION} already exists on PyPI. Skipping publish."
echo "skip_publish=true" >> $GITHUB_ENV
else
echo "skip_publish=false" >> $GITHUB_ENV
fi
- name: Publish to PyPI
if: env.skip_publish != 'true'
uses: pypa/gh-action-pypi-publish@release/v1
with:
password: ${{ secrets.PYPI_API_TOKEN }}
```
--------------------------------------------------------------------------------
/src/mcp_weather_server/utils.py:
--------------------------------------------------------------------------------
```python
from datetime import datetime, timezone
import json
from typing import List
from zoneinfo import ZoneInfo
from mcp.types import ErrorData
from mcp import McpError
from pydantic import BaseModel
from dateutil import parser
class TimeResult(BaseModel):
timezone: str
datetime: str
def get_zoneinfo(timezone_name: str) -> ZoneInfo:
try:
return ZoneInfo(timezone_name)
except Exception as e:
error_data = ErrorData(code=-1, message=f"Invalid timezone: {str(e)}")
raise McpError(error_data)
def format_get_weather_bytime(data_result) -> str:
return f"""
Please analyze the above JSON weather forecast information and generate a report for me. Please note that the content is provided
city: city name
start_date: search weather start time
end_date: search weather end time
weather_data: weather data.
{json.dumps(data_result)}
"""
def get_closest_utc_index(hourly_times: List[str]) -> int:
"""
Returns the index of the datetime in `hourly_times` closest to the current UTC time
or a provided datetime.
:param hourly_times: List of ISO 8601 time strings (UTC)
:return: Index of the closest datetime in the list
"""
current_time = datetime.now(timezone.utc)
parsed_times = [
parser.isoparse(t).replace(tzinfo=timezone.utc) if parser.isoparse(t).tzinfo is None
else parser.isoparse(t).astimezone(timezone.utc)
for t in hourly_times
]
return min(range(len(parsed_times)), key=lambda i: abs(parsed_times[i] - current_time))
# Weather code descriptions (from Open-Meteo documentation)
weather_descriptions = {
0: "Clear sky",
1: "Mainly clear",
2: "Partly cloudy",
3: "Overcast",
45: "Fog",
48: "Depositing rime fog",
51: "Light drizzle",
53: "Moderate drizzle",
55: "Dense drizzle",
56: "Light freezing drizzle",
57: "Dense freezing drizzle",
61: "Slight rain",
63: "Moderate rain",
65: "Heavy rain",
66: "Light freezing rain",
67: "Heavy freezing rain",
71: "Slight snow fall",
73: "Moderate snow fall",
75: "Heavy snow fall",
77: "Snow grains",
80: "Slight rain showers",
81: "Moderate rain showers",
82: "Violent rain showers",
85: "Slight snow showers",
86: "Heavy snow showers",
95: "Thunderstorm",
96: "Thunderstorm with slight hail",
99: "Thunderstorm with heavy hail",
}
```
--------------------------------------------------------------------------------
/src/mcp_weather_server/tools/toolhandler.py:
--------------------------------------------------------------------------------
```python
"""
Base ToolHandler class for extensible MCP tool management.
This follows the architecture pattern from mcp-gsuite for easy extension.
"""
from abc import ABC, abstractmethod
from collections.abc import Sequence
from mcp.types import (
Tool,
TextContent,
ImageContent,
EmbeddedResource,
)
class ToolHandler(ABC):
"""
Abstract base class for all MCP tool handlers.
This provides a consistent interface for tool registration,
description, and execution across the MCP weather server.
"""
def __init__(self, tool_name: str):
"""
Initialize a tool handler with a unique name.
Args:
tool_name: Unique identifier for this tool
"""
self.name = tool_name
@abstractmethod
def get_tool_description(self) -> Tool:
"""
Return the MCP Tool description for this handler.
This method must be implemented by each tool handler to define
the tool's schema, parameters, and documentation.
Returns:
Tool: MCP Tool object with schema and metadata
"""
raise NotImplementedError("Each tool handler must implement get_tool_description")
@abstractmethod
async def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
"""
Execute the tool with provided arguments.
This method contains the actual logic for the tool's functionality.
Note: This method should be async to support API calls.
Args:
args: Dictionary of arguments passed to the tool
Returns:
Sequence of MCP content objects (text, image, or embedded resources)
Raises:
RuntimeError: For missing required arguments or execution errors
"""
raise NotImplementedError("Each tool handler must implement run_tool")
def validate_required_args(self, args: dict, required_fields: list[str]) -> None:
"""
Validate that all required arguments are present.
Args:
args: Dictionary of provided arguments
required_fields: List of required field names
Raises:
RuntimeError: If any required field is missing
"""
missing_fields = [field for field in required_fields if field not in args]
if missing_fields:
raise RuntimeError(f"Missing required arguments: {', '.join(missing_fields)}")
```
--------------------------------------------------------------------------------
/tests/test_mcp_weather_server.py:
--------------------------------------------------------------------------------
```python
"""
Comprehensive test suite for MCP Weather Server.
This file imports and runs all test modules to ensure complete coverage.
"""
import pytest
from tests.test_weather_service import *
from tests.test_weather_tools import *
from tests.test_time_tools import *
from tests.test_utils import *
from tests.test_server import *
from tests.test_integration import *
# Additional legacy tests for backward compatibility
import httpx
from unittest.mock import AsyncMock, Mock, patch
@pytest.mark.asyncio
async def test_legacy_get_weather_functionality():
"""Legacy test to ensure backward compatibility."""
# This test ensures that any legacy get_weather function (if it exists) still works
try:
from src.mcp_weather_server.server import get_weather
# Mock HTTP client for legacy function
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
mock_response = Mock()
mock_response.status_code = 500
mock_client.get.return_value = mock_response
mock_client_class.return_value.__aenter__.return_value = mock_client
result = await get_weather("InvalidCity")
assert "Error" in str(result) or "error" in str(result).lower()
except ImportError:
# get_weather function doesn't exist, which is fine
# The new architecture uses tool handlers instead
pytest.skip("Legacy get_weather function not found - using new tool handler architecture")
# Smoke tests to verify all components work together
class TestSmokeTests:
"""Smoke tests to verify basic functionality."""
@pytest.mark.asyncio
async def test_can_import_all_modules(self):
"""Test that all modules can be imported without errors."""
from src.mcp_weather_server import server
from src.mcp_weather_server import utils
from src.mcp_weather_server.tools import toolhandler
from src.mcp_weather_server.tools import weather_service
from src.mcp_weather_server.tools import tools_weather
from src.mcp_weather_server.tools import tools_time
# Basic assertions to ensure imports worked
assert hasattr(server, 'register_all_tools')
assert hasattr(utils, 'get_zoneinfo')
assert hasattr(toolhandler, 'ToolHandler')
assert hasattr(weather_service, 'WeatherService')
@pytest.mark.asyncio
async def test_server_starts_without_errors(self):
"""Test that server initialization doesn't raise errors."""
from src.mcp_weather_server.server import register_all_tools, list_tools
# Clear any existing handlers
from src.mcp_weather_server.server import tool_handlers
tool_handlers.clear()
# Register tools and list them
register_all_tools()
tools = await list_tools()
# Verify we have the expected number of tools
assert len(tools) >= 6
tool_names = [tool.name for tool in tools]
expected_tools = [
"get_current_weather",
"get_weather_byDateTimeRange",
"get_weather_details",
"get_current_datetime",
"get_timezone_info",
"convert_time"
]
for expected_tool in expected_tools:
assert expected_tool in tool_names
```
--------------------------------------------------------------------------------
/src/mcp_weather_server/server.py:
--------------------------------------------------------------------------------
```python
"""
This server implements a modular, extensible design pattern similar to mcp-gsuite,
making it easy to add new weather-related tools and functionality.
Supports both stdio and SSE MCP server modes.
"""
import argparse
import asyncio
import logging
import sys
import traceback
from typing import Any, Dict, Optional
from collections.abc import Sequence
from mcp.server import Server
from mcp.types import (
Tool,
TextContent,
ImageContent,
EmbeddedResource,
)
# SSE-related imports (imported conditionally)
try:
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.routing import Mount, Route
import uvicorn
SSE_AVAILABLE = True
except ImportError:
SSE_AVAILABLE = False
# Import tool handlers
from .tools.toolhandler import ToolHandler
from .tools.tools_weather import (
GetCurrentWeatherToolHandler,
GetWeatherByDateRangeToolHandler,
GetWeatherDetailsToolHandler,
)
from .tools.tools_time import (
GetCurrentDateTimeToolHandler,
GetTimeZoneInfoToolHandler,
ConvertTimeToolHandler,
)
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-weather")
# Create the MCP server instance
app = Server("mcp-weather-server")
# Global tool handlers registry
tool_handlers: Dict[str, ToolHandler] = {}
def add_tool_handler(tool_handler: ToolHandler) -> None:
"""
Register a tool handler with the server.
Args:
tool_handler: The tool handler instance to register
"""
global tool_handlers
tool_handlers[tool_handler.name] = tool_handler
logger.info(f"Registered tool handler: {tool_handler.name}")
def get_tool_handler(name: str) -> ToolHandler | None:
"""
Retrieve a tool handler by name.
Args:
name: The name of the tool handler
Returns:
The tool handler instance or None if not found
"""
return tool_handlers.get(name)
def register_all_tools() -> None:
"""
Register all available tool handlers.
This function serves as the central registry for all tools.
New tool handlers should be added here for automatic registration.
"""
# Weather tools
add_tool_handler(GetCurrentWeatherToolHandler())
add_tool_handler(GetWeatherByDateRangeToolHandler())
add_tool_handler(GetWeatherDetailsToolHandler())
# Time tools
add_tool_handler(GetCurrentDateTimeToolHandler())
add_tool_handler(GetTimeZoneInfoToolHandler())
add_tool_handler(ConvertTimeToolHandler())
logger.info(f"Registered {len(tool_handlers)} tool handlers")
def create_starlette_app(mcp_server: Server, *, debug: bool = False) -> Starlette:
"""
Create a Starlette application that can serve the provided mcp server with SSE.
Args:
mcp_server: The MCP server instance
debug: Whether to enable debug mode
Returns:
Starlette application instance
"""
if not SSE_AVAILABLE:
raise RuntimeError("SSE dependencies not available. Install with: pip install starlette uvicorn")
sse = SseServerTransport("/messages/")
async def handle_sse(request: Request) -> None:
async with sse.connect_sse(
request.scope,
request.receive,
request._send,
) as (read_stream, write_stream):
await mcp_server.run(
read_stream,
write_stream,
mcp_server.create_initialization_options(),
)
return Starlette(
debug=debug,
routes=[
Route("/sse", endpoint=handle_sse),
Mount("/messages/", app=sse.handle_post_message),
],
)
@app.list_tools()
async def list_tools() -> list[Tool]:
"""
List all available tools.
Returns:
List of Tool objects describing all registered tools
"""
try:
tools = [handler.get_tool_description() for handler in tool_handlers.values()]
logger.info(f"Listed {len(tools)} available tools")
return tools
except Exception as e:
logger.exception(f"Error listing tools: {str(e)}")
raise
@app.call_tool()
async def call_tool(name: str, arguments: Any) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
"""
Execute a tool with the provided arguments.
Args:
name: The name of the tool to execute
arguments: The arguments to pass to the tool
Returns:
Sequence of MCP content objects
Raises:
RuntimeError: If the tool execution fails
"""
try:
# Validate arguments
if not isinstance(arguments, dict):
raise RuntimeError("Arguments must be a dictionary")
# Get the tool handler
tool_handler = get_tool_handler(name)
if not tool_handler:
raise ValueError(f"Unknown tool: {name}")
logger.info(f"Executing tool: {name} with arguments: {list(arguments.keys())}")
# Execute the tool
result = await tool_handler.run_tool(arguments)
logger.info(f"Tool {name} executed successfully")
return result
except Exception as e:
logger.exception(f"Error executing tool {name}: {str(e)}")
error_traceback = traceback.format_exc()
logger.error(f"Full traceback: {error_traceback}")
# Return error as text content
return [
TextContent(
type="text",
text=f"Error executing tool '{name}': {str(e)}"
)
]
async def main():
"""
Main entry point for the MCP weather server.
Supports both stdio and SSE modes based on command line arguments.
"""
# Parse command line arguments
parser = argparse.ArgumentParser(description='MCP Weather Server - supports stdio and SSE modes')
parser.add_argument('--mode', choices=['stdio', 'sse'], default='stdio',
help='Server mode: stdio (default) or sse')
parser.add_argument('--host', default='0.0.0.0',
help='Host to bind to (SSE mode only, default: 0.0.0.0)')
parser.add_argument('--port', type=int, default=8080,
help='Port to listen on (SSE mode only, default: 8080)')
parser.add_argument('--debug', action='store_true',
help='Enable debug mode')
args = parser.parse_args()
try:
# Register all tools
register_all_tools()
logger.info(f"Starting MCP Weather Server in {args.mode} mode...")
logger.info(f"Python version: {sys.version}")
logger.info(f"Registered tools: {list(tool_handlers.keys())}")
# Run the server in the specified mode
await run_server(args.mode, args.host, args.port, args.debug)
except Exception as e:
logger.exception(f"Failed to start server: {str(e)}")
raise
async def run_server(mode: str, host: str = "0.0.0.0", port: int = 8080, debug: bool = False):
"""
Unified server runner that supports both stdio and SSE modes.
Args:
mode: Server mode ("stdio" or "sse")
host: Host to bind to (SSE mode only)
port: Port to listen on (SSE mode only)
debug: Whether to enable debug mode
"""
if mode == "stdio":
logger.info("Starting stdio server...")
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
elif mode == "sse":
if not SSE_AVAILABLE:
raise RuntimeError(
"SSE mode requires additional dependencies. "
"Install with: pip install starlette uvicorn"
)
logger.info(f"Starting SSE server on {host}:{port}...")
# Create Starlette app with SSE transport
starlette_app = create_starlette_app(app, debug=debug)
# Configure uvicorn
config = uvicorn.Config(
app=starlette_app,
host=host,
port=port,
log_level="debug" if debug else "info"
)
# Run the server
server = uvicorn.Server(config)
await server.serve()
else:
raise ValueError(f"Unknown mode: {mode}")
if __name__ == "__main__":
import asyncio
asyncio.run(main())
```
--------------------------------------------------------------------------------
/tests/conftest.py:
--------------------------------------------------------------------------------
```python
"""
Pytest configuration and fixtures for MCP weather server tests.
"""
import pytest
import asyncio
import subprocess
import json
import time
import sys
import os
from pathlib import Path
import httpx
from typing import Any, Dict, List
class MCPServerProcess:
"""Helper class to manage MCP server process for testing."""
def __init__(self):
self.process = None
self.port = 8000
self.base_url = f"http://localhost:{self.port}"
async def start(self):
"""Start the MCP server process."""
# Get the project root directory
project_root = Path(__file__).parent.parent
server_script = project_root / "test_server.py"
# Start the server process
self.process = subprocess.Popen([
sys.executable, str(server_script)
], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
# Wait for server to start up
max_retries = 30
for _ in range(max_retries):
try:
async with httpx.AsyncClient() as client:
response = await client.get(f"{self.base_url}/health")
if response.status_code == 200:
break
except (httpx.RequestError, httpx.HTTPStatusError):
pass
await asyncio.sleep(0.1)
else:
raise RuntimeError("MCP server failed to start")
async def stop(self):
"""Stop the MCP server process."""
if self.process:
self.process.terminate()
self.process.wait()
self.process = None
async def call_tool(self, name: str, arguments: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Call a tool on the MCP server."""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/call_tool",
json={"name": name, "arguments": arguments}
)
response.raise_for_status()
return response.json()["result"]
async def list_tools(self) -> List[Dict[str, Any]]:
"""List available tools on the MCP server."""
async with httpx.AsyncClient() as client:
response = await client.get(f"{self.base_url}/list_tools")
response.raise_for_status()
return response.json()["tools"]
@pytest.fixture(scope="session")
async def mcp_server():
"""Fixture that provides a running MCP server process for testing."""
server = MCPServerProcess()
await server.start()
yield server
await server.stop()
@pytest.fixture
def mock_datetime():
"""Fixture to provide controlled datetime mocking."""
from unittest.mock import patch
from datetime import datetime
from zoneinfo import ZoneInfo
class MockDateTime:
def __init__(self):
self.fixed_time = None
self.patches = []
def set_fixed_time(self, dt: datetime):
"""Set a fixed time to return from datetime.now()."""
self.fixed_time = dt
def start_mocking(self):
"""Start datetime mocking."""
if self.patches:
self.stop_mocking()
def mock_now(tz=None):
if self.fixed_time and tz:
return self.fixed_time.astimezone(tz)
elif self.fixed_time:
return self.fixed_time
else:
return datetime.now(tz)
patch_obj = patch('src.mcp_weather_server.tools.tools_time.datetime')
mock_datetime = patch_obj.start()
mock_datetime.now = mock_now
self.patches.append(patch_obj)
def stop_mocking(self):
"""Stop datetime mocking."""
for patch_obj in self.patches:
patch_obj.stop()
self.patches.clear()
mock_dt = MockDateTime()
yield mock_dt
mock_dt.stop_mocking()
@pytest.fixture
def mock_timezone():
"""Fixture to provide controlled timezone mocking."""
from unittest.mock import patch
from zoneinfo import ZoneInfo
class MockTimezone:
def __init__(self):
self.patches = []
self.zone_mapping = {}
def add_zone(self, name: str, zone: ZoneInfo):
"""Add a timezone mapping."""
self.zone_mapping[name] = zone
def start_mocking(self):
"""Start timezone mocking."""
if self.patches:
self.stop_mocking()
def mock_get_zoneinfo(name: str):
if name in self.zone_mapping:
return self.zone_mapping[name]
return ZoneInfo(name)
patch_obj = patch('src.mcp_weather_server.utils.get_zoneinfo', side_effect=mock_get_zoneinfo)
patch_obj.start()
self.patches.append(patch_obj)
def stop_mocking(self):
"""Stop timezone mocking."""
for patch_obj in self.patches:
patch_obj.stop()
self.patches.clear()
mock_tz = MockTimezone()
yield mock_tz
mock_tz.stop_mocking()
# Weather service fixtures
@pytest.fixture
def weather_service():
"""Create a WeatherService instance for testing."""
from src.mcp_weather_server.tools.weather_service import WeatherService
return WeatherService()
@pytest.fixture
def mock_geo_response():
"""Mock geocoding API response."""
return {
"results": [
{
"latitude": 40.7128,
"longitude": -74.0060
}
]
}
@pytest.fixture
def mock_empty_geo_response():
"""Mock empty geocoding API response."""
return {"results": []}
@pytest.fixture
def mock_weather_response():
"""Mock weather API response."""
return {
"hourly": {
"time": [
"2024-01-01T12:00",
"2024-01-01T13:00"
],
"temperature_2m": [20.0, 21.0],
"relative_humidity_2m": [65, 66],
"dew_point_2m": [13.0, 14.0],
"weather_code": [0, 1]
}
}
@pytest.fixture
def mock_weather_range_response():
"""Mock weather range API response."""
return {
"hourly": {
"time": [
"2024-01-01T12:00",
"2024-01-01T13:00",
"2024-01-02T12:00",
"2024-01-02T13:00"
],
"temperature_2m": [20.0, 21.0, 22.0, 23.0],
"relative_humidity_2m": [65, 66, 67, 68],
"dew_point_2m": [13.0, 14.0, 15.0, 16.0],
"weather_code": [0, 1, 0, 1]
}
}
@pytest.fixture
def sample_current_weather_data():
"""Sample current weather data for testing."""
return {
"city": "New York",
"latitude": 40.7128,
"longitude": -74.0060,
"temperature_c": 25.0,
"relative_humidity_percent": 70,
"dew_point_c": 16.0,
"weather_code": 1,
"weather_description": "Mainly clear"
}
@pytest.fixture
def sample_weather_range_data():
"""Sample weather range data for testing."""
return {
"city": "New York",
"start_date": "2024-01-01",
"end_date": "2024-01-02",
"weather_data": [
{
"datetime": "2024-01-01T12:00:00",
"temperature_c": 20.0,
"relative_humidity_percent": 65,
"dew_point_c": 13.0,
"weather_code": 0,
"weather_description": "Clear sky"
},
{
"datetime": "2024-01-01T13:00:00",
"temperature_c": 21.0,
"relative_humidity_percent": 66,
"dew_point_c": 14.0,
"weather_code": 1,
"weather_description": "Mainly clear"
}
]
}
# HTTP client mock fixtures
@pytest.fixture
def mock_successful_geo_client():
"""Mock successful geocoding client."""
from unittest.mock import AsyncMock, Mock
client = AsyncMock()
response = Mock()
response.status_code = 200
response.json.return_value = {
"results": [
{
"latitude": 40.7128,
"longitude": -74.0060
}
]
}
client.get.return_value = response
return client
@pytest.fixture
def mock_failed_client():
"""Mock failed HTTP client."""
from unittest.mock import AsyncMock, Mock
client = AsyncMock()
response = Mock()
response.status_code = 500
client.get.return_value = response
return client
@pytest.fixture
def mock_network_error_client():
"""Mock network error HTTP client."""
from unittest.mock import AsyncMock
import httpx
client = AsyncMock()
client.get.side_effect = httpx.RequestError("Network error")
return client
```
--------------------------------------------------------------------------------
/src/mcp_weather_server/tools/tools_time.py:
--------------------------------------------------------------------------------
```python
"""
Time-related tool handlers for the MCP weather server.
This module contains time and timezone-related tool implementations.
"""
import json
import logging
from collections.abc import Sequence
from datetime import datetime
from mcp.types import Tool, TextContent, ImageContent, EmbeddedResource
from .toolhandler import ToolHandler
from .. import utils
logger = logging.getLogger("mcp-weather")
class GetCurrentDateTimeToolHandler(ToolHandler):
"""
Tool handler for getting current date and time in a specified timezone.
"""
def __init__(self):
super().__init__("get_current_datetime")
def get_tool_description(self) -> Tool:
"""
Return the tool description for current datetime lookup.
"""
return Tool(
name=self.name,
description="""Get current time in specified timezone.""",
inputSchema={
"type": "object",
"properties": {
"timezone_name": {
"type": "string",
"description": "IANA timezone name (e.g., 'America/New_York', 'Europe/London'). Use UTC timezone if no timezone provided by the user."
}
},
"required": ["timezone_name"]
}
)
async def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
"""
Execute the current datetime tool.
"""
try:
self.validate_required_args(args, ["timezone_name"])
timezone_name = args["timezone_name"]
logger.info(f"Getting current time for timezone: {timezone_name}")
# Get timezone info
timezone = utils.get_zoneinfo(timezone_name)
current_time = datetime.now(timezone)
# Create time result
time_result = utils.TimeResult(
timezone=timezone_name,
datetime=current_time.isoformat(timespec="seconds"),
)
return [
TextContent(
type="text",
text=json.dumps(time_result.model_dump(), indent=2)
)
]
except Exception as e:
logger.exception(f"Error in get_current_datetime: {str(e)}")
return [
TextContent(
type="text",
text=f"Error getting current time: {str(e)}"
)
]
class GetTimeZoneInfoToolHandler(ToolHandler):
"""
Tool handler for getting information about timezones.
"""
def __init__(self):
super().__init__("get_timezone_info")
def get_tool_description(self) -> Tool:
"""
Return the tool description for timezone information lookup.
"""
return Tool(
name=self.name,
description="""Get information about a specific timezone including current time and UTC offset.""",
inputSchema={
"type": "object",
"properties": {
"timezone_name": {
"type": "string",
"description": "IANA timezone name (e.g., 'America/New_York', 'Europe/London')"
}
},
"required": ["timezone_name"]
}
)
async def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
"""
Execute the timezone info tool.
"""
try:
self.validate_required_args(args, ["timezone_name"])
timezone_name = args["timezone_name"]
logger.info(f"Getting timezone info for: {timezone_name}")
# Get timezone info
timezone = utils.get_zoneinfo(timezone_name)
current_time = datetime.now(timezone)
utc_time = datetime.utcnow()
# Calculate UTC offset
offset = current_time.utcoffset()
offset_hours = offset.total_seconds() / 3600 if offset else 0
timezone_info = {
"timezone_name": timezone_name,
"current_local_time": current_time.isoformat(timespec="seconds"),
"current_utc_time": utc_time.isoformat(timespec="seconds"),
"utc_offset_hours": offset_hours,
"is_dst": current_time.dst() is not None and current_time.dst().total_seconds() > 0,
"timezone_abbreviation": current_time.strftime("%Z"),
}
return [
TextContent(
type="text",
text=json.dumps(timezone_info, indent=2)
)
]
except Exception as e:
logger.exception(f"Error in get_timezone_info: {str(e)}")
return [
TextContent(
type="text",
text=f"Error getting timezone info: {str(e)}"
)
]
class ConvertTimeToolHandler(ToolHandler):
"""
Tool handler for converting time between different timezones.
"""
def __init__(self):
super().__init__("convert_time")
def get_tool_description(self) -> Tool:
"""
Return the tool description for time conversion.
"""
return Tool(
name=self.name,
description="""Convert time from one timezone to another.""",
inputSchema={
"type": "object",
"properties": {
"datetime_str": {
"type": "string",
"description": "DateTime string in ISO format (e.g., '2024-01-15T14:30:00') or 'now' for current time"
},
"from_timezone": {
"type": "string",
"description": "Source timezone (IANA timezone name)"
},
"to_timezone": {
"type": "string",
"description": "Target timezone (IANA timezone name)"
}
},
"required": ["datetime_str", "from_timezone", "to_timezone"]
}
)
async def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
"""
Execute the time conversion tool.
"""
try:
self.validate_required_args(args, ["datetime_str", "from_timezone", "to_timezone"])
datetime_str = args["datetime_str"]
from_timezone_name = args["from_timezone"]
to_timezone_name = args["to_timezone"]
logger.info(f"Converting time '{datetime_str}' from {from_timezone_name} to {to_timezone_name}")
# Get timezone objects
from_timezone = utils.get_zoneinfo(from_timezone_name)
to_timezone = utils.get_zoneinfo(to_timezone_name)
# Parse the datetime
if datetime_str.lower() == "now":
source_time = datetime.now(from_timezone)
else:
# Parse the datetime string and localize it
if datetime_str.endswith('Z'):
# UTC time
naive_time = datetime.fromisoformat(datetime_str[:-1])
source_time = naive_time.replace(tzinfo=from_timezone)
else:
# Assume local time in from_timezone
naive_time = datetime.fromisoformat(datetime_str)
source_time = naive_time.replace(tzinfo=from_timezone)
# Convert to target timezone
target_time = source_time.astimezone(to_timezone)
conversion_result = {
"original_datetime": source_time.isoformat(timespec="seconds"),
"original_timezone": from_timezone_name,
"converted_datetime": target_time.isoformat(timespec="seconds"),
"converted_timezone": to_timezone_name,
"time_difference_hours": (target_time.utcoffset().total_seconds() - source_time.utcoffset().total_seconds()) / 3600
}
return [
TextContent(
type="text",
text=json.dumps(conversion_result, indent=2)
)
]
except Exception as e:
logger.exception(f"Error in convert_time: {str(e)}")
return [
TextContent(
type="text",
text=f"Error converting time: {str(e)}"
)
]
```
--------------------------------------------------------------------------------
/src/mcp_weather_server/tools/weather_service.py:
--------------------------------------------------------------------------------
```python
"""
Weather service for handling all weather API interactions.
This separates the business logic from the tool handlers.
"""
import httpx
import logging
from typing import Dict, List, Tuple, Any
from datetime import datetime, timezone
from . import utils
logger = logging.getLogger("mcp-weather")
class WeatherService:
"""
Service class for weather-related API interactions.
This class encapsulates all weather API logic, making it reusable
across different tool handlers and easier to test and maintain.
"""
BASE_GEO_URL = "https://geocoding-api.open-meteo.com/v1/search"
BASE_WEATHER_URL = "https://api.open-meteo.com/v1/forecast"
def __init__(self):
"""Initialize the weather service."""
pass
async def get_coordinates(self, city: str) -> Tuple[float, float]:
"""
Fetch the latitude and longitude for a given city using the Open-Meteo Geocoding API.
Args:
city: The name of the city to fetch coordinates for
Returns:
Tuple of (latitude, longitude)
Raises:
ValueError: If the coordinates cannot be retrieved
"""
async with httpx.AsyncClient() as client:
try:
geo_response = await client.get(f"{self.BASE_GEO_URL}?name={city}")
if geo_response.status_code != 200:
raise ValueError(f"Geocoding API returned status {geo_response.status_code}")
geo_data = geo_response.json()
if "results" not in geo_data or not geo_data["results"]:
raise ValueError(f"No coordinates found for city: {city}")
result = geo_data["results"][0]
return result["latitude"], result["longitude"]
except httpx.RequestError as e:
raise ValueError(f"Network error while fetching coordinates for {city}: {str(e)}")
except (KeyError, IndexError) as e:
raise ValueError(f"Invalid response format from geocoding API: {str(e)}")
async def get_current_weather(self, city: str) -> Dict[str, Any]:
"""
Get current weather information for a specified city.
Args:
city: The name of the city
Returns:
Dictionary containing current weather data
Raises:
ValueError: If weather data cannot be retrieved
"""
try:
latitude, longitude = await self.get_coordinates(city)
# Build the weather API URL for current conditions
url = (
f"{self.BASE_WEATHER_URL}"
f"?latitude={latitude}&longitude={longitude}"
f"&hourly=temperature_2m,relative_humidity_2m,dew_point_2m,weather_code"
f"&timezone=GMT&forecast_days=1"
)
logger.info(f"Fetching current weather from: {url}")
async with httpx.AsyncClient() as client:
weather_response = await client.get(url)
if weather_response.status_code != 200:
raise ValueError(f"Weather API returned status {weather_response.status_code}")
weather_data = weather_response.json()
# Find the current hour index
current_index = utils.get_closest_utc_index(weather_data["hourly"]["time"])
# Extract current weather data
current_weather = {
"city": city,
"latitude": latitude,
"longitude": longitude,
"time": weather_data["hourly"]["time"][current_index],
"temperature_c": weather_data["hourly"]["temperature_2m"][current_index],
"relative_humidity_percent": weather_data["hourly"]["relative_humidity_2m"][current_index],
"dew_point_c": weather_data["hourly"]["dew_point_2m"][current_index],
"weather_code": weather_data["hourly"]["weather_code"][current_index],
"weather_description": utils.weather_descriptions.get(
weather_data["hourly"]["weather_code"][current_index],
"Unknown weather condition"
)
}
return current_weather
except httpx.RequestError as e:
raise ValueError(f"Network error while fetching weather for {city}: {str(e)}")
except (KeyError, IndexError) as e:
raise ValueError(f"Invalid response format from weather API: {str(e)}")
async def get_weather_by_date_range(
self,
city: str,
start_date: str,
end_date: str
) -> Dict[str, Any]:
"""
Get weather information for a specified city between start and end dates.
Args:
city: The name of the city
start_date: Start date in YYYY-MM-DD format
end_date: End date in YYYY-MM-DD format
Returns:
Dictionary containing weather data for the date range
Raises:
ValueError: If weather data cannot be retrieved
"""
try:
latitude, longitude = await self.get_coordinates(city)
# Build the weather API URL for date range
url = (
f"{self.BASE_WEATHER_URL}"
f"?latitude={latitude}&longitude={longitude}"
f"&hourly=temperature_2m,relative_humidity_2m,dew_point_2m,weather_code"
f"&timezone=GMT&start_date={start_date}&end_date={end_date}"
)
logger.info(f"Fetching weather history from: {url}")
async with httpx.AsyncClient() as client:
response = await client.get(url)
if response.status_code != 200:
raise ValueError(f"Weather API returned status {response.status_code}")
data = response.json()
# Process the hourly data
times = data["hourly"]["time"]
temperatures = data["hourly"]["temperature_2m"]
humidities = data["hourly"]["relative_humidity_2m"]
dew_points = data["hourly"]["dew_point_2m"]
weather_codes = data["hourly"]["weather_code"]
weather_data = []
for time, temp, humidity, dew_point, weather_code in zip(
times, temperatures, humidities, dew_points, weather_codes
):
weather_data.append({
"time": time,
"temperature_c": temp,
"humidity_percent": humidity,
"dew_point_c": dew_point,
"weather_code": weather_code,
"weather_description": utils.weather_descriptions.get(
weather_code, "Unknown weather condition"
)
})
return {
"city": city,
"latitude": latitude,
"longitude": longitude,
"start_date": start_date,
"end_date": end_date,
"weather_data": weather_data
}
except httpx.RequestError as e:
raise ValueError(f"Network error while fetching weather for {city}: {str(e)}")
except (KeyError, IndexError) as e:
raise ValueError(f"Invalid response format from weather API: {str(e)}")
def format_current_weather_response(self, weather_data: Dict[str, Any]) -> str:
"""
Format current weather data into a human-readable string.
Args:
weather_data: Weather data dictionary from get_current_weather
Returns:
Formatted weather description string
"""
return (
f"The weather in {weather_data['city']} is {weather_data['weather_description']} "
f"with a temperature of {weather_data['temperature_c']}°C, "
f"relative humidity at {weather_data['relative_humidity_percent']}%, "
f"and dew point at {weather_data['dew_point_c']}°C."
)
def format_weather_range_response(self, weather_data: Dict[str, Any]) -> str:
"""
Format weather range data for analysis.
Args:
weather_data: Weather data dictionary from get_weather_by_date_range
Returns:
Formatted string ready for AI analysis
"""
return utils.format_get_weather_bytime(weather_data)
```
--------------------------------------------------------------------------------
/src/mcp_weather_server/tools/tools_weather.py:
--------------------------------------------------------------------------------
```python
"""
Weather-related tool handlers for the MCP weather server.
This module contains all weather-specific tool implementations.
"""
import json
import logging
from collections.abc import Sequence
from mcp.types import Tool, TextContent, ImageContent, EmbeddedResource
from .toolhandler import ToolHandler
from .weather_service import WeatherService
logger = logging.getLogger("mcp-weather")
class GetCurrentWeatherToolHandler(ToolHandler):
"""
Tool handler for getting current weather information for a city.
"""
def __init__(self):
super().__init__("get_current_weather")
self.weather_service = WeatherService()
def get_tool_description(self) -> Tool:
"""
Return the tool description for current weather lookup.
"""
return Tool(
name=self.name,
description="""Get current weather information for a specified city.
It extracts the current hour's temperature and weather code, maps
the weather code to a human-readable description, and returns a formatted summary.""",
inputSchema={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city to fetch weather information for, PLEASE NOTE English name only, if the parameter city isn't English please translate to English before invoking this function."
}
},
"required": ["city"]
}
)
async def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
"""
Execute the current weather tool.
"""
try:
self.validate_required_args(args, ["city"])
city = args["city"]
logger.info(f"Getting current weather for: {city}")
# Get weather data from service
weather_data = await self.weather_service.get_current_weather(city)
# Format the response
formatted_response = self.weather_service.format_current_weather_response(weather_data)
return [
TextContent(
type="text",
text=formatted_response
)
]
except ValueError as e:
logger.error(f"Weather service error: {str(e)}")
return [
TextContent(
type="text",
text=f"Error: {str(e)}"
)
]
except Exception as e:
logger.exception(f"Unexpected error in get_current_weather: {str(e)}")
return [
TextContent(
type="text",
text=f"Unexpected error occurred: {str(e)}"
)
]
class GetWeatherByDateRangeToolHandler(ToolHandler):
"""
Tool handler for getting weather information for a date range.
"""
def __init__(self):
super().__init__("get_weather_byDateTimeRange")
self.weather_service = WeatherService()
def get_tool_description(self) -> Tool:
"""
Return the tool description for weather date range lookup.
"""
return Tool(
name=self.name,
description="""Get weather information for a specified city between start and end dates.""",
inputSchema={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city to fetch weather information for, PLEASE NOTE English name only, if the parameter city isn't English please translate to English before invoking this function."
},
"start_date": {
"type": "string",
"description": "Start date in format YYYY-MM-DD, please follow ISO 8601 format"
},
"end_date": {
"type": "string",
"description": "End date in format YYYY-MM-DD , please follow ISO 8601 format"
}
},
"required": ["city", "start_date", "end_date"]
}
)
async def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
"""
Execute the weather date range tool.
"""
try:
self.validate_required_args(args, ["city", "start_date", "end_date"])
city = args["city"]
start_date = args["start_date"]
end_date = args["end_date"]
logger.info(f"Getting weather for {city} from {start_date} to {end_date}")
# Get weather data from service
weather_data = await self.weather_service.get_weather_by_date_range(
city, start_date, end_date
)
# Format the response for analysis
formatted_response = self.weather_service.format_weather_range_response(weather_data)
return [
TextContent(
type="text",
text=formatted_response
)
]
except ValueError as e:
logger.error(f"Weather service error: {str(e)}")
return [
TextContent(
type="text",
text=f"Error: {str(e)}"
)
]
except Exception as e:
logger.exception(f"Unexpected error in get_weather_by_date_range: {str(e)}")
return [
TextContent(
type="text",
text=f"Unexpected error occurred: {str(e)}"
)
]
class GetWeatherDetailsToolHandler(ToolHandler):
"""
Tool handler for getting detailed weather information with raw data.
This tool provides structured JSON output for programmatic use.
"""
def __init__(self):
super().__init__("get_weather_details")
self.weather_service = WeatherService()
def get_tool_description(self) -> Tool:
"""
Return the tool description for detailed weather lookup.
"""
return Tool(
name=self.name,
description="""Get detailed weather information for a specified city as structured JSON data.
This tool provides raw weather data for programmatic analysis and processing.""",
inputSchema={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city to fetch weather information for, PLEASE NOTE English name only, if the parameter city isn't English please translate to English before invoking this function."
},
"include_forecast": {
"type": "boolean",
"description": "Whether to include forecast data (next 24 hours)",
"default": False
}
},
"required": ["city"]
}
)
async def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
"""
Execute the detailed weather tool.
"""
try:
self.validate_required_args(args, ["city"])
city = args["city"]
include_forecast = args.get("include_forecast", False)
logger.info(f"Getting detailed weather for: {city} (forecast: {include_forecast})")
# Get current weather data
weather_data = await self.weather_service.get_current_weather(city)
# If forecast is requested, get the next 24 hours
if include_forecast:
from datetime import datetime, timedelta
today = datetime.now().strftime("%Y-%m-%d")
tomorrow = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d")
forecast_data = await self.weather_service.get_weather_by_date_range(
city, today, tomorrow
)
weather_data["forecast"] = forecast_data["weather_data"]
return [
TextContent(
type="text",
text=json.dumps(weather_data, indent=2)
)
]
except ValueError as e:
logger.error(f"Weather service error: {str(e)}")
return [
TextContent(
type="text",
text=json.dumps({"error": str(e)}, indent=2)
)
]
except Exception as e:
logger.exception(f"Unexpected error in get_weather_details: {str(e)}")
return [
TextContent(
type="text",
text=json.dumps({"error": f"Unexpected error occurred: {str(e)}"}, indent=2)
)
]
```
--------------------------------------------------------------------------------
/tests/test_weather_service.py:
--------------------------------------------------------------------------------
```python
"""
Unit tests for WeatherService class.
"""
import pytest
import httpx
from unittest.mock import AsyncMock, Mock, patch
from src.mcp_weather_server.tools.weather_service import WeatherService
class TestWeatherService:
"""Test cases for WeatherService class."""
@pytest.fixture
def weather_service(self):
"""Create a WeatherService instance for testing."""
return WeatherService()
@pytest.mark.asyncio
async def test_get_coordinates_success(
self,
weather_service,
mock_successful_geo_client,
mock_geo_response
):
"""Test successful coordinate retrieval."""
with patch('httpx.AsyncClient') as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_successful_geo_client
lat, lon = await weather_service.get_coordinates("New York")
assert lat == 40.7128
assert lon == -74.0060
mock_successful_geo_client.get.assert_called_once()
@pytest.mark.asyncio
async def test_get_coordinates_api_error(
self,
weather_service,
mock_failed_client
):
"""Test coordinate retrieval with API error."""
with patch('httpx.AsyncClient') as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_failed_client
with pytest.raises(ValueError, match="Geocoding API returned status 500"):
await weather_service.get_coordinates("Invalid City")
@pytest.mark.asyncio
async def test_get_coordinates_no_results(
self,
weather_service,
mock_empty_geo_response
):
"""Test coordinate retrieval with no results."""
mock_client = AsyncMock()
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = mock_empty_geo_response
mock_client.get.return_value = mock_response
with patch('httpx.AsyncClient') as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
with pytest.raises(ValueError, match="No coordinates found for city: Unknown City"):
await weather_service.get_coordinates("Unknown City")
@pytest.mark.asyncio
async def test_get_coordinates_network_error(
self,
weather_service,
mock_network_error_client
):
"""Test coordinate retrieval with network error."""
with patch('httpx.AsyncClient') as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_network_error_client
with pytest.raises(ValueError, match="Network error while fetching coordinates"):
await weather_service.get_coordinates("Test City")
@pytest.mark.asyncio
async def test_get_current_weather_success(
self,
weather_service,
mock_geo_response,
mock_weather_response
):
"""Test successful current weather retrieval."""
# Mock the coordinates call
with patch.object(weather_service, 'get_coordinates', return_value=(40.7128, -74.0060)):
# Mock the weather API call
mock_client = AsyncMock()
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = mock_weather_response
mock_client.get.return_value = mock_response
with patch('httpx.AsyncClient') as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
# Mock the utility function to return a predictable index
with patch('src.mcp_weather_server.utils.get_closest_utc_index', return_value=1):
result = await weather_service.get_current_weather("New York")
assert result["city"] == "New York"
assert result["latitude"] == 40.7128
assert result["longitude"] == -74.0060
assert result["temperature_c"] == 21.0
assert result["relative_humidity_percent"] == 66
assert result["dew_point_c"] == 14.0
assert result["weather_code"] == 1
assert "Mainly clear" in result["weather_description"]
@pytest.mark.asyncio
async def test_get_current_weather_api_error(self, weather_service):
"""Test current weather retrieval with API error."""
with patch.object(weather_service, 'get_coordinates', return_value=(40.7128, -74.0060)):
mock_client = AsyncMock()
mock_response = Mock()
mock_response.status_code = 500
mock_client.get.return_value = mock_response
with patch('httpx.AsyncClient') as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
with pytest.raises(ValueError, match="Weather API returned status 500"):
await weather_service.get_current_weather("New York")
@pytest.mark.asyncio
async def test_get_weather_by_date_range_success(
self,
weather_service,
mock_weather_range_response
):
"""Test successful weather range retrieval."""
with patch.object(weather_service, 'get_coordinates', return_value=(40.7128, -74.0060)):
mock_client = AsyncMock()
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = mock_weather_range_response
mock_client.get.return_value = mock_response
with patch('httpx.AsyncClient') as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
result = await weather_service.get_weather_by_date_range(
"New York", "2024-01-01", "2024-01-02"
)
assert result["city"] == "New York"
assert result["start_date"] == "2024-01-01"
assert result["end_date"] == "2024-01-02"
assert len(result["weather_data"]) == 4
assert result["weather_data"][0]["temperature_c"] == 20.0
assert result["weather_data"][1]["weather_description"] == "Mainly clear"
@pytest.mark.asyncio
async def test_get_weather_by_date_range_invalid_response(self, weather_service):
"""Test weather range retrieval with invalid response."""
with patch.object(weather_service, 'get_coordinates', return_value=(40.7128, -74.0060)):
mock_client = AsyncMock()
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {"invalid": "response"}
mock_client.get.return_value = mock_response
with patch('httpx.AsyncClient') as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
with pytest.raises(ValueError, match="Invalid response format from weather API"):
await weather_service.get_weather_by_date_range(
"New York", "2024-01-01", "2024-01-02"
)
def test_format_current_weather_response(
self,
weather_service,
sample_current_weather_data
):
"""Test formatting of current weather response."""
result = weather_service.format_current_weather_response(sample_current_weather_data)
expected = (
"The weather in New York is Mainly clear with a temperature of 25.0°C, "
"relative humidity at 70%, and dew point at 16.0°C."
)
assert result == expected
def test_format_weather_range_response(
self,
weather_service,
sample_weather_range_data
):
"""Test formatting of weather range response."""
with patch('src.mcp_weather_server.utils.format_get_weather_bytime') as mock_format:
mock_format.return_value = "formatted response"
result = weather_service.format_weather_range_response(sample_weather_range_data)
assert result == "formatted response"
mock_format.assert_called_once_with(sample_weather_range_data)
@pytest.mark.asyncio
async def test_get_coordinates_malformed_response(self, weather_service):
"""Test coordinate retrieval with malformed JSON response."""
mock_client = AsyncMock()
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {"results": [{"invalid": "data"}]}
mock_client.get.return_value = mock_response
with patch('httpx.AsyncClient') as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
with pytest.raises(ValueError, match="Invalid response format from geocoding API"):
await weather_service.get_coordinates("Test City")
@pytest.mark.asyncio
async def test_get_current_weather_network_error(self, weather_service):
"""Test current weather retrieval with network error."""
with patch.object(weather_service, 'get_coordinates', return_value=(40.7128, -74.0060)):
mock_client = AsyncMock()
mock_client.get.side_effect = httpx.RequestError("Network error")
with patch('httpx.AsyncClient') as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
with pytest.raises(ValueError, match="Network error while fetching weather"):
await weather_service.get_current_weather("New York")
@pytest.mark.asyncio
async def test_get_weather_by_date_range_network_error(self, weather_service):
"""Test weather range retrieval with network error."""
with patch.object(weather_service, 'get_coordinates', return_value=(40.7128, -74.0060)):
mock_client = AsyncMock()
mock_client.get.side_effect = httpx.RequestError("Network error")
with patch('httpx.AsyncClient') as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
with pytest.raises(ValueError, match="Network error while fetching weather"):
await weather_service.get_weather_by_date_range(
"New York", "2024-01-01", "2024-01-02"
)
```
--------------------------------------------------------------------------------
/tests/test_server.py:
--------------------------------------------------------------------------------
```python
"""
Unit tests for server functionality.
"""
import pytest
from unittest.mock import AsyncMock, Mock, patch
from src.mcp_weather_server.server import (
add_tool_handler,
get_tool_handler,
register_all_tools,
list_tools,
call_tool,
tool_handlers
)
from src.mcp_weather_server.tools.toolhandler import ToolHandler
from mcp.types import Tool, TextContent
class MockToolHandler(ToolHandler):
"""Mock tool handler for testing."""
def __init__(self, name="mock_tool"):
super().__init__(name)
self.get_tool_description_called = False
self.run_tool_called = False
self.run_tool_args = None
def get_tool_description(self) -> Tool:
self.get_tool_description_called = True
return Tool(
name=self.name,
description="Mock tool for testing",
inputSchema={
"type": "object",
"properties": {
"test_param": {"type": "string"}
},
"required": ["test_param"]
}
)
async def run_tool(self, args: dict):
self.run_tool_called = True
self.run_tool_args = args
return [TextContent(type="text", text=f"Mock result for {args}")]
class TestServerFunctions:
"""Test cases for server functions."""
def setup_method(self):
"""Clear tool handlers before each test."""
global tool_handlers
tool_handlers.clear()
def test_add_tool_handler(self):
"""Test adding a tool handler."""
handler = MockToolHandler("test_tool")
add_tool_handler(handler)
assert "test_tool" in tool_handlers
assert tool_handlers["test_tool"] == handler
def test_get_tool_handler_existing(self):
"""Test getting an existing tool handler."""
handler = MockToolHandler("existing_tool")
add_tool_handler(handler)
retrieved = get_tool_handler("existing_tool")
assert retrieved == handler
def test_get_tool_handler_nonexistent(self):
"""Test getting a non-existent tool handler."""
retrieved = get_tool_handler("nonexistent_tool")
assert retrieved is None
def test_register_all_tools(self):
"""Test registering all tools."""
register_all_tools()
# Check that expected tools are registered
expected_tools = [
"get_current_weather",
"get_weather_byDateTimeRange",
"get_weather_details",
"get_current_datetime",
"get_timezone_info",
"convert_time"
]
for tool_name in expected_tools:
assert tool_name in tool_handlers
assert tool_handlers[tool_name] is not None
@pytest.mark.asyncio
async def test_list_tools_empty(self):
"""Test listing tools when no tools are registered."""
result = await list_tools()
assert result == []
@pytest.mark.asyncio
async def test_list_tools_with_handlers(self):
"""Test listing tools with registered handlers."""
handler1 = MockToolHandler("tool1")
handler2 = MockToolHandler("tool2")
add_tool_handler(handler1)
add_tool_handler(handler2)
result = await list_tools()
assert len(result) == 2
assert handler1.get_tool_description_called
assert handler2.get_tool_description_called
tool_names = [tool.name for tool in result]
assert "tool1" in tool_names
assert "tool2" in tool_names
@pytest.mark.asyncio
async def test_list_tools_exception_handling(self):
"""Test list_tools exception handling."""
# Create a mock handler that raises an exception
with patch.dict(tool_handlers, {"bad_tool": Mock()}):
tool_handlers["bad_tool"].get_tool_description.side_effect = Exception("Test error")
with pytest.raises(Exception):
await list_tools()
@pytest.mark.asyncio
async def test_call_tool_success(self):
"""Test successful tool execution."""
handler = MockToolHandler("test_tool")
add_tool_handler(handler)
args = {"test_param": "test_value"}
result = await call_tool("test_tool", args)
assert handler.run_tool_called
assert handler.run_tool_args == args
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "test_value" in result[0].text
@pytest.mark.asyncio
async def test_call_tool_nonexistent(self):
"""Test calling a non-existent tool."""
args = {"test_param": "test_value"}
result = await call_tool("nonexistent_tool", args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Error executing tool 'nonexistent_tool'" in result[0].text
@pytest.mark.asyncio
async def test_call_tool_invalid_arguments(self):
"""Test calling a tool with invalid arguments."""
handler = MockToolHandler("test_tool")
add_tool_handler(handler)
# Pass non-dict arguments
result = await call_tool("test_tool", "invalid_args")
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Error executing tool 'test_tool'" in result[0].text
assert "Arguments must be a dictionary" in result[0].text
@pytest.mark.asyncio
async def test_call_tool_handler_exception(self):
"""Test calling a tool when handler raises an exception."""
handler = MockToolHandler("test_tool")
handler.run_tool = AsyncMock(side_effect=Exception("Handler error"))
add_tool_handler(handler)
args = {"test_param": "test_value"}
result = await call_tool("test_tool", args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Error executing tool 'test_tool'" in result[0].text
assert "Handler error" in result[0].text
@pytest.mark.asyncio
async def test_call_tool_with_complex_args(self):
"""Test calling a tool with complex arguments."""
handler = MockToolHandler("complex_tool")
add_tool_handler(handler)
complex_args = {
"string_param": "test_string",
"number_param": 42,
"list_param": [1, 2, 3],
"dict_param": {"nested": "value"}
}
result = await call_tool("complex_tool", complex_args)
assert handler.run_tool_called
assert handler.run_tool_args == complex_args
assert len(result) == 1
class TestToolHandlerIntegration:
"""Integration tests for tool handler registration and execution."""
def setup_method(self):
"""Clear tool handlers before each test."""
global tool_handlers
tool_handlers.clear()
@pytest.mark.asyncio
async def test_full_workflow(self):
"""Test complete workflow from registration to execution."""
# Register a tool
handler = MockToolHandler("workflow_tool")
add_tool_handler(handler)
# List tools
tools = await list_tools()
assert len(tools) == 1
assert tools[0].name == "workflow_tool"
# Call the tool
args = {"test_param": "workflow_test"}
result = await call_tool("workflow_tool", args)
assert len(result) == 1
assert "workflow_test" in result[0].text
@pytest.mark.asyncio
async def test_multiple_tools_workflow(self):
"""Test workflow with multiple tools."""
# Register multiple tools
handler1 = MockToolHandler("tool_one")
handler2 = MockToolHandler("tool_two")
handler3 = MockToolHandler("tool_three")
add_tool_handler(handler1)
add_tool_handler(handler2)
add_tool_handler(handler3)
# List all tools
tools = await list_tools()
assert len(tools) == 3
tool_names = [tool.name for tool in tools]
assert "tool_one" in tool_names
assert "tool_two" in tool_names
assert "tool_three" in tool_names
# Call each tool
for tool_name in ["tool_one", "tool_two", "tool_three"]:
args = {"test_param": f"test_{tool_name}"}
result = await call_tool(tool_name, args)
assert len(result) == 1
assert f"test_{tool_name}" in result[0].text
@pytest.mark.asyncio
async def test_real_tool_registration(self):
"""Test registration of real tool handlers."""
register_all_tools()
# Verify specific tools are registered
weather_tool = get_tool_handler("get_current_weather")
assert weather_tool is not None
time_tool = get_tool_handler("get_current_datetime")
assert time_tool is not None
# Test tool descriptions
tools = await list_tools()
assert len(tools) >= 6 # Should have at least 6 tools
# Verify tool names
tool_names = [tool.name for tool in tools]
expected_tools = [
"get_current_weather",
"get_weather_byDateTimeRange",
"get_weather_details",
"get_current_datetime",
"get_timezone_info",
"convert_time"
]
for expected_tool in expected_tools:
assert expected_tool in tool_names
def test_tool_handler_replacement(self):
"""Test replacing an existing tool handler."""
# Add initial handler
handler1 = MockToolHandler("replaceable_tool")
add_tool_handler(handler1)
# Verify it's registered
assert get_tool_handler("replaceable_tool") == handler1
# Replace with new handler
handler2 = MockToolHandler("replaceable_tool")
add_tool_handler(handler2)
# Verify replacement
assert get_tool_handler("replaceable_tool") == handler2
assert get_tool_handler("replaceable_tool") != handler1
def test_tool_handlers_persistence(self):
"""Test that tool handlers persist across multiple operations."""
handler = MockToolHandler("persistent_tool")
add_tool_handler(handler)
# Perform multiple operations
retrieved1 = get_tool_handler("persistent_tool")
retrieved2 = get_tool_handler("persistent_tool")
retrieved3 = get_tool_handler("persistent_tool")
# All should return the same handler instance
assert retrieved1 == handler
assert retrieved2 == handler
assert retrieved3 == handler
assert retrieved1 == retrieved2 == retrieved3
```
--------------------------------------------------------------------------------
/tests/test_utils.py:
--------------------------------------------------------------------------------
```python
"""
Unit tests for utility functions.
"""
import pytest
from datetime import datetime, timezone
from unittest.mock import patch, Mock
from zoneinfo import ZoneInfo
from mcp import McpError
from src.mcp_weather_server.utils import (
TimeResult,
get_zoneinfo,
format_get_weather_bytime,
get_closest_utc_index,
weather_descriptions
)
class TestTimeResult:
"""Test cases for TimeResult model."""
def test_time_result_creation(self):
"""Test TimeResult model creation."""
time_result = TimeResult(
timezone="America/New_York",
datetime="2024-01-01T15:30:45"
)
assert time_result.timezone == "America/New_York"
assert time_result.datetime == "2024-01-01T15:30:45"
def test_time_result_dict(self):
"""Test TimeResult conversion to dictionary."""
time_result = TimeResult(
timezone="UTC",
datetime="2024-01-01T12:00:00"
)
result_dict = time_result.model_dump()
assert result_dict["timezone"] == "UTC"
assert result_dict["datetime"] == "2024-01-01T12:00:00"
class TestGetZoneinfo:
"""Test cases for get_zoneinfo function."""
def test_get_zoneinfo_valid_timezone(self):
"""Test get_zoneinfo with valid timezone."""
tz = get_zoneinfo("America/New_York")
assert isinstance(tz, ZoneInfo)
assert str(tz) == "America/New_York"
def test_get_zoneinfo_utc(self):
"""Test get_zoneinfo with UTC timezone."""
tz = get_zoneinfo("UTC")
assert isinstance(tz, ZoneInfo)
assert str(tz) == "UTC"
def test_get_zoneinfo_invalid_timezone(self):
"""Test get_zoneinfo with invalid timezone."""
with pytest.raises(McpError, match="Invalid timezone"):
get_zoneinfo("Invalid/Timezone")
def test_get_zoneinfo_empty_string(self):
"""Test get_zoneinfo with empty string."""
with pytest.raises(McpError, match="Invalid timezone"):
get_zoneinfo("")
def test_get_zoneinfo_none(self):
"""Test get_zoneinfo with None."""
with pytest.raises(McpError, match="Invalid timezone"):
get_zoneinfo(None)
class TestFormatGetWeatherBytime:
"""Test cases for format_get_weather_bytime function."""
def test_format_get_weather_bytime_basic(self):
"""Test basic formatting of weather data."""
weather_data = {
"city": "New York",
"start_date": "2024-01-01",
"end_date": "2024-01-02",
"weather_data": [
{
"time": "2024-01-01T12:00",
"temperature_c": 25.0,
"humidity_percent": 70,
"weather_description": "Clear sky"
}
]
}
result = format_get_weather_bytime(weather_data)
assert "analyze" in result.lower()
assert "New York" in result
assert "2024-01-01" in result
assert "2024-01-02" in result
assert "weather_data" in result
def test_format_get_weather_bytime_empty_data(self):
"""Test formatting with empty weather data."""
weather_data = {
"city": "Test City",
"start_date": "2024-01-01",
"end_date": "2024-01-01",
"weather_data": []
}
result = format_get_weather_bytime(weather_data)
assert "Test City" in result
assert "weather_data" in result
assert "[]" in result
def test_format_get_weather_bytime_complex_data(self):
"""Test formatting with complex weather data."""
weather_data = {
"city": "London",
"latitude": 51.5074,
"longitude": -0.1278,
"start_date": "2024-01-01",
"end_date": "2024-01-03",
"weather_data": [
{
"time": "2024-01-01T00:00",
"temperature_c": 15.0,
"humidity_percent": 80,
"dew_point_c": 11.5,
"weather_code": 61,
"weather_description": "Slight rain"
},
{
"time": "2024-01-01T12:00",
"temperature_c": 18.0,
"humidity_percent": 75,
"dew_point_c": 13.0,
"weather_code": 1,
"weather_description": "Mainly clear"
}
]
}
result = format_get_weather_bytime(weather_data)
assert "London" in result
assert "51.5074" in result
assert "-0.1278" in result
assert "Slight rain" in result
assert "Mainly clear" in result
class TestGetClosestUtcIndex:
"""Test cases for get_closest_utc_index function."""
def test_get_closest_utc_index_exact_match(self):
"""Test finding index when current time matches exactly."""
current_time = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
hourly_times = [
"2024-01-01T10:00:00Z",
"2024-01-01T11:00:00Z",
"2024-01-01T12:00:00Z", # This should match
"2024-01-01T13:00:00Z",
"2024-01-01T14:00:00Z"
]
with patch('src.mcp_weather_server.utils.datetime') as mock_datetime:
mock_datetime.now.return_value = current_time
mock_datetime.side_effect = lambda *args, **kwargs: datetime(*args, **kwargs)
index = get_closest_utc_index(hourly_times)
assert index == 2
def test_get_closest_utc_index_closest_before(self):
"""Test finding closest index when current time is between hours."""
current_time = datetime(2024, 1, 1, 12, 30, 0, tzinfo=timezone.utc)
hourly_times = [
"2024-01-01T10:00:00Z",
"2024-01-01T11:00:00Z",
"2024-01-01T12:00:00Z", # This should be closest (30 min away)
"2024-01-01T13:00:00Z", # This is also 30 min away, but 12:00 comes first
"2024-01-01T14:00:00Z"
]
with patch('src.mcp_weather_server.utils.datetime') as mock_datetime:
mock_datetime.now.return_value = current_time
mock_datetime.side_effect = lambda *args, **kwargs: datetime(*args, **kwargs)
index = get_closest_utc_index(hourly_times)
assert index == 2
def test_get_closest_utc_index_closest_after(self):
"""Test finding closest index when current time is closer to next hour."""
current_time = datetime(2024, 1, 1, 12, 35, 0, tzinfo=timezone.utc)
hourly_times = [
"2024-01-01T10:00:00Z",
"2024-01-01T11:00:00Z",
"2024-01-01T12:00:00Z", # 35 min away
"2024-01-01T13:00:00Z", # 25 min away - closer
"2024-01-01T14:00:00Z"
]
with patch('src.mcp_weather_server.utils.datetime') as mock_datetime:
mock_datetime.now.return_value = current_time
mock_datetime.side_effect = lambda *args, **kwargs: datetime(*args, **kwargs)
index = get_closest_utc_index(hourly_times)
assert index == 3
def test_get_closest_utc_index_single_time(self):
"""Test with single time entry."""
current_time = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
hourly_times = ["2024-01-01T15:00:00Z"]
with patch('src.mcp_weather_server.utils.datetime') as mock_datetime:
mock_datetime.now.return_value = current_time
mock_datetime.side_effect = lambda *args, **kwargs: datetime(*args, **kwargs)
index = get_closest_utc_index(hourly_times)
assert index == 0
def test_get_closest_utc_index_empty_list(self):
"""Test with empty time list."""
current_time = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
hourly_times = []
with patch('src.mcp_weather_server.utils.datetime') as mock_datetime:
mock_datetime.now.return_value = current_time
mock_datetime.side_effect = lambda *args, **kwargs: datetime(*args, **kwargs)
with pytest.raises(ValueError):
get_closest_utc_index(hourly_times)
def test_get_closest_utc_index_different_timezones(self):
"""Test with times in different timezone formats."""
current_time = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
hourly_times = [
"2024-01-01T10:00:00", # No timezone, should be treated as UTC
"2024-01-01T11:00:00Z", # UTC
"2024-01-01T12:00:00+00:00", # UTC with offset
"2024-01-01T08:00:00-04:00", # EST (12:00 UTC)
"2024-01-01T14:00:00Z"
]
with patch('src.mcp_weather_server.utils.datetime') as mock_datetime:
mock_datetime.now.return_value = current_time
mock_datetime.side_effect = lambda *args, **kwargs: datetime(*args, **kwargs)
# Should find one of the 12:00 UTC equivalent times
index = get_closest_utc_index(hourly_times)
assert index in [2, 3] # Either the +00:00 or -04:00 versions
class TestWeatherDescriptions:
"""Test cases for weather descriptions dictionary."""
def test_weather_descriptions_coverage(self):
"""Test that weather_descriptions contains expected codes."""
# Test some common weather codes
assert weather_descriptions[0] == "Clear sky"
assert weather_descriptions[1] == "Mainly clear"
assert weather_descriptions[2] == "Partly cloudy"
assert weather_descriptions[3] == "Overcast"
# Test precipitation codes
assert weather_descriptions[61] == "Slight rain"
assert weather_descriptions[63] == "Moderate rain"
assert weather_descriptions[65] == "Heavy rain"
# Test snow codes
assert weather_descriptions[71] == "Slight snow fall"
assert weather_descriptions[73] == "Moderate snow fall"
assert weather_descriptions[75] == "Heavy snow fall"
# Test severe weather
assert weather_descriptions[95] == "Thunderstorm"
assert weather_descriptions[99] == "Thunderstorm with heavy hail"
def test_weather_descriptions_unknown_code(self):
"""Test behavior with unknown weather codes."""
# Unknown codes should not be in the dictionary
assert 999 not in weather_descriptions
assert -1 not in weather_descriptions
def test_weather_descriptions_fog_codes(self):
"""Test fog-related weather codes."""
assert weather_descriptions[45] == "Fog"
assert weather_descriptions[48] == "Depositing rime fog"
def test_weather_descriptions_drizzle_codes(self):
"""Test drizzle-related weather codes."""
assert weather_descriptions[51] == "Light drizzle"
assert weather_descriptions[53] == "Moderate drizzle"
assert weather_descriptions[55] == "Dense drizzle"
assert weather_descriptions[56] == "Light freezing drizzle"
assert weather_descriptions[57] == "Dense freezing drizzle"
def test_weather_descriptions_shower_codes(self):
"""Test shower-related weather codes."""
assert weather_descriptions[80] == "Slight rain showers"
assert weather_descriptions[81] == "Moderate rain showers"
assert weather_descriptions[82] == "Violent rain showers"
assert weather_descriptions[85] == "Slight snow showers"
assert weather_descriptions[86] == "Heavy snow showers"
def test_weather_descriptions_hail_codes(self):
"""Test hail-related weather codes."""
assert weather_descriptions[96] == "Thunderstorm with slight hail"
assert weather_descriptions[99] == "Thunderstorm with heavy hail"
def test_weather_descriptions_completeness(self):
"""Test that all expected weather codes are present."""
expected_codes = [
0, 1, 2, 3, 45, 48,
51, 53, 55, 56, 57,
61, 63, 65, 66, 67,
71, 73, 75, 77,
80, 81, 82, 85, 86,
95, 96, 99
]
for code in expected_codes:
assert code in weather_descriptions, f"Weather code {code} is missing"
assert isinstance(weather_descriptions[code], str)
assert len(weather_descriptions[code]) > 0
```
--------------------------------------------------------------------------------
/tests/test_weather_tools.py:
--------------------------------------------------------------------------------
```python
"""
Unit tests for weather tool handlers.
"""
import pytest
import json
from unittest.mock import AsyncMock, Mock, patch
from mcp.types import TextContent
from src.mcp_weather_server.tools.tools_weather import (
GetCurrentWeatherToolHandler,
GetWeatherByDateRangeToolHandler,
GetWeatherDetailsToolHandler
)
class TestGetCurrentWeatherToolHandler:
"""Test cases for GetCurrentWeatherToolHandler."""
@pytest.fixture
def handler(self):
"""Create a GetCurrentWeatherToolHandler instance."""
return GetCurrentWeatherToolHandler()
def test_tool_description(self, handler):
"""Test the tool description is properly formatted."""
description = handler.get_tool_description()
assert description.name == "get_current_weather"
assert "current weather" in description.description.lower()
assert description.inputSchema["type"] == "object"
assert "city" in description.inputSchema["properties"]
assert description.inputSchema["required"] == ["city"]
@pytest.mark.asyncio
async def test_run_tool_success(self, handler, sample_current_weather_data):
"""Test successful tool execution."""
# Mock the weather service
from unittest.mock import Mock
mock_service = Mock()
mock_service.get_current_weather = AsyncMock(return_value=sample_current_weather_data)
mock_service.format_current_weather_response.return_value = "Formatted weather response"
handler.weather_service = mock_service
args = {"city": "New York"}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "Formatted weather response"
mock_service.get_current_weather.assert_called_once_with("New York")
mock_service.format_current_weather_response.assert_called_once_with(sample_current_weather_data)
@pytest.mark.asyncio
async def test_run_tool_missing_city(self, handler):
"""Test tool execution with missing city argument."""
args = {}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Missing required arguments: city" in result[0].text
@pytest.mark.asyncio
async def test_run_tool_weather_service_error(self, handler):
"""Test tool execution when weather service raises ValueError."""
mock_service = AsyncMock()
mock_service.get_current_weather.side_effect = ValueError("API error")
handler.weather_service = mock_service
args = {"city": "Invalid City"}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Error: API error" in result[0].text
@pytest.mark.asyncio
async def test_run_tool_unexpected_error(self, handler):
"""Test tool execution with unexpected error."""
mock_service = AsyncMock()
mock_service.get_current_weather.side_effect = Exception("Unexpected error")
handler.weather_service = mock_service
args = {"city": "Test City"}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Unexpected error occurred: Unexpected error" in result[0].text
class TestGetWeatherByDateRangeToolHandler:
"""Test cases for GetWeatherByDateRangeToolHandler."""
@pytest.fixture
def handler(self):
"""Create a GetWeatherByDateRangeToolHandler instance."""
return GetWeatherByDateRangeToolHandler()
def test_tool_description(self, handler):
"""Test the tool description is properly formatted."""
description = handler.get_tool_description()
assert description.name == "get_weather_byDateTimeRange"
assert "weather information" in description.description.lower()
assert description.inputSchema["type"] == "object"
required_fields = ["city", "start_date", "end_date"]
for field in required_fields:
assert field in description.inputSchema["properties"]
assert set(description.inputSchema["required"]) == set(required_fields)
@pytest.mark.asyncio
async def test_run_tool_success(self, handler, sample_weather_range_data):
"""Test successful tool execution."""
from unittest.mock import Mock
mock_service = Mock()
mock_service.get_weather_by_date_range = AsyncMock(return_value=sample_weather_range_data)
mock_service.format_weather_range_response.return_value = "Formatted range response"
handler.weather_service = mock_service
args = {
"city": "New York",
"start_date": "2024-01-01",
"end_date": "2024-01-02"
}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "Formatted range response"
mock_service.get_weather_by_date_range.assert_called_once_with(
"New York", "2024-01-01", "2024-01-02"
)
@pytest.mark.asyncio
async def test_run_tool_missing_required_args(self, handler):
"""Test tool execution with missing required arguments."""
args = {"city": "New York"} # Missing dates
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Missing required arguments" in result[0].text
assert "start_date" in result[0].text
assert "end_date" in result[0].text
@pytest.mark.asyncio
async def test_run_tool_service_error(self, handler):
"""Test tool execution when weather service raises error."""
mock_service = AsyncMock()
mock_service.get_weather_by_date_range.side_effect = ValueError("Date range error")
handler.weather_service = mock_service
args = {
"city": "Test City",
"start_date": "invalid-date",
"end_date": "2024-01-02"
}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Error: Date range error" in result[0].text
class TestGetWeatherDetailsToolHandler:
"""Test cases for GetWeatherDetailsToolHandler."""
@pytest.fixture
def handler(self):
"""Create a GetWeatherDetailsToolHandler instance."""
return GetWeatherDetailsToolHandler()
def test_tool_description(self, handler):
"""Test the tool description is properly formatted."""
description = handler.get_tool_description()
assert description.name == "get_weather_details"
assert "detailed weather information" in description.description.lower()
assert description.inputSchema["type"] == "object"
assert "city" in description.inputSchema["properties"]
assert description.inputSchema["required"] == ["city"]
@pytest.mark.asyncio
async def test_run_tool_success(self, handler, sample_current_weather_data):
"""Test successful tool execution."""
mock_service = AsyncMock()
mock_service.get_current_weather.return_value = sample_current_weather_data
handler.weather_service = mock_service
args = {"city": "New York"}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
# Parse the JSON response to verify structure
response_data = json.loads(result[0].text)
assert response_data["city"] == "New York"
assert response_data["temperature_c"] == 25.0
assert response_data["weather_description"] == "Mainly clear"
mock_service.get_current_weather.assert_called_once_with("New York")
@pytest.mark.asyncio
async def test_run_tool_missing_city(self, handler):
"""Test tool execution with missing city argument."""
args = {}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Missing required arguments: city" in result[0].text
@pytest.mark.asyncio
async def test_run_tool_service_error(self, handler):
"""Test tool execution when weather service raises error."""
mock_service = AsyncMock()
mock_service.get_current_weather.side_effect = ValueError("Service error")
handler.weather_service = mock_service
args = {"city": "Invalid City"}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
# Check that the error is properly formatted as JSON
response_data = json.loads(result[0].text)
assert "error" in response_data
assert "Service error" in response_data["error"]
@pytest.mark.asyncio
async def test_run_tool_json_serialization_error(self, handler):
"""Test tool execution when JSON serialization fails."""
mock_service = AsyncMock()
# Create an object that can't be JSON serialized
invalid_data = {"city": "Test", "invalid": object()}
mock_service.get_current_weather.return_value = invalid_data
handler.weather_service = mock_service
args = {"city": "Test City"}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Unexpected error occurred" in result[0].text
# Additional integration-style tests for tool handlers
class TestToolHandlerIntegration:
"""Integration tests for tool handlers with realistic scenarios."""
@pytest.mark.asyncio
async def test_current_weather_end_to_end(self):
"""Test current weather tool with mocked HTTP responses."""
handler = GetCurrentWeatherToolHandler()
# Mock the entire HTTP call chain
mock_geo_data = {
"results": [{"latitude": 51.5074, "longitude": -0.1278}]
}
mock_weather_data = {
"hourly": {
"time": ["2024-01-01T12:00"],
"temperature_2m": [15.5],
"relative_humidity_2m": [80],
"dew_point_2m": [12.0],
"weather_code": [61]
}
}
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
# Setup mock responses for geo and weather API calls
geo_response = Mock()
geo_response.status_code = 200
geo_response.json.return_value = mock_geo_data
weather_response = Mock()
weather_response.status_code = 200
weather_response.json.return_value = mock_weather_data
# Return different responses for different URLs
def mock_get(url):
if "geocoding-api" in url:
return geo_response
elif "api.open-meteo.com" in url:
return weather_response
else:
raise ValueError(f"Unexpected URL: {url}")
mock_client.get.side_effect = mock_get
mock_client_class.return_value.__aenter__.return_value = mock_client
with patch('src.mcp_weather_server.utils.get_closest_utc_index', return_value=0):
result = await handler.run_tool({"city": "London"})
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "London" in result[0].text
assert "15.5°C" in result[0].text
assert "Slight rain" in result[0].text
@pytest.mark.asyncio
async def test_weather_range_end_to_end(self):
"""Test weather range tool with mocked HTTP responses."""
handler = GetWeatherByDateRangeToolHandler()
mock_geo_data = {
"results": [{"latitude": 48.8566, "longitude": 2.3522}]
}
mock_weather_data = {
"hourly": {
"time": ["2024-01-01T00:00", "2024-01-01T12:00"],
"temperature_2m": [10.0, 15.0],
"relative_humidity_2m": [90, 75],
"dew_point_2m": [8.5, 11.0],
"weather_code": [3, 1]
}
}
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
geo_response = Mock()
geo_response.status_code = 200
geo_response.json.return_value = mock_geo_data
weather_response = Mock()
weather_response.status_code = 200
weather_response.json.return_value = mock_weather_data
def mock_get(url):
if "geocoding-api" in url:
return geo_response
else:
return weather_response
mock_client.get.side_effect = mock_get
mock_client_class.return_value.__aenter__.return_value = mock_client
args = {
"city": "Paris",
"start_date": "2024-01-01",
"end_date": "2024-01-01"
}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
# The response should contain analysis prompt with JSON data
assert "analyze" in result[0].text.lower()
assert "Paris" in result[0].text
```
--------------------------------------------------------------------------------
/tests/test_time_tools.py:
--------------------------------------------------------------------------------
```python
"""
Unit tests for time-related tool handlers.
"""
import pytest
import json
from unittest.mock import Mock, patch
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
from mcp.types import TextContent, ErrorData
from mcp import McpError
from src.mcp_weather_server.tools.tools_time import (
GetCurrentDateTimeToolHandler,
GetTimeZoneInfoToolHandler,
ConvertTimeToolHandler
)
class TestGetCurrentDateTimeToolHandler:
"""Test cases for GetCurrentDateTimeToolHandler."""
@pytest.fixture
def handler(self):
"""Create a GetCurrentDateTimeToolHandler instance."""
return GetCurrentDateTimeToolHandler()
def test_tool_description(self, handler):
"""Test the tool description is properly formatted."""
description = handler.get_tool_description()
assert description.name == "get_current_datetime"
assert "current time" in description.description.lower()
assert description.inputSchema["type"] == "object"
assert "timezone_name" in description.inputSchema["properties"]
assert description.inputSchema["required"] == ["timezone_name"]
@pytest.mark.asyncio
async def test_run_tool_success(self, handler):
"""Test successful tool execution."""
fixed_time = datetime(2024, 1, 1, 15, 30, 45, tzinfo=ZoneInfo("America/New_York"))
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
mock_get_tz.return_value = ZoneInfo("America/New_York")
with patch('src.mcp_weather_server.tools.tools_time.datetime') as mock_datetime:
mock_datetime.now.return_value = fixed_time
args = {"timezone_name": "America/New_York"}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
response_data = json.loads(result[0].text)
assert response_data["timezone"] == "America/New_York"
assert "2024-01-01T15:30:45" in response_data["datetime"]
@pytest.mark.asyncio
async def test_run_tool_missing_timezone(self, handler):
"""Test tool execution with missing timezone argument."""
args = {}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Missing required arguments: timezone_name" in result[0].text
@pytest.mark.asyncio
async def test_run_tool_invalid_timezone(self, handler):
"""Test tool execution with invalid timezone."""
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
mock_get_tz.side_effect = McpError(ErrorData(code=-1, message="Invalid timezone: Invalid/Timezone"))
args = {"timezone_name": "Invalid/Timezone"}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Error getting current time: Invalid timezone" in result[0].text
@pytest.mark.asyncio
async def test_run_tool_utc_timezone(self, handler):
"""Test tool execution with UTC timezone."""
fixed_time = datetime(2024, 1, 1, 12, 0, 0, tzinfo=ZoneInfo("UTC"))
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
mock_get_tz.return_value = ZoneInfo("UTC")
with patch('src.mcp_weather_server.tools.tools_time.datetime') as mock_datetime:
mock_datetime.now.return_value = fixed_time
args = {"timezone_name": "UTC"}
result = await handler.run_tool(args)
assert len(result) == 1
response_data = json.loads(result[0].text)
assert response_data["timezone"] == "UTC"
assert "2024-01-01T12:00:00" in response_data["datetime"]
class TestGetTimeZoneInfoToolHandler:
"""Test cases for GetTimeZoneInfoToolHandler."""
@pytest.fixture
def handler(self):
"""Create a GetTimeZoneInfoToolHandler instance."""
return GetTimeZoneInfoToolHandler()
def test_tool_description(self, handler):
"""Test the tool description is properly formatted."""
description = handler.get_tool_description()
assert description.name == "get_timezone_info"
assert "timezone" in description.description.lower()
assert description.inputSchema["type"] == "object"
assert "timezone_name" in description.inputSchema["properties"]
assert description.inputSchema["required"] == ["timezone_name"]
@pytest.mark.asyncio
async def test_run_tool_success(self, handler):
"""Test successful tool execution."""
fixed_time = datetime(2024, 6, 15, 14, 30, 0, tzinfo=ZoneInfo("Europe/London"))
fixed_utc_time = datetime(2024, 6, 15, 13, 0, 0) # UTC time without timezone
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
mock_get_tz.return_value = ZoneInfo("Europe/London")
with patch('src.mcp_weather_server.tools.tools_time.datetime') as mock_datetime:
mock_datetime.now.return_value = fixed_time
mock_datetime.utcnow.return_value = fixed_utc_time
args = {"timezone_name": "Europe/London"}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
response_data = json.loads(result[0].text)
assert response_data["timezone_name"] == "Europe/London"
assert "current_local_time" in response_data
assert "utc_offset_hours" in response_data
@pytest.mark.asyncio
async def test_run_tool_missing_timezone(self, handler):
"""Test tool execution with missing timezone argument."""
args = {}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Missing required arguments: timezone_name" in result[0].text
@pytest.mark.asyncio
async def test_run_tool_invalid_timezone(self, handler):
"""Test tool execution with invalid timezone."""
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
mock_get_tz.side_effect = McpError(ErrorData(code=-1, message="Invalid timezone"))
args = {"timezone_name": "Invalid/Timezone"}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Error getting timezone info: Invalid timezone" in result[0].text
class TestConvertTimeToolHandler:
"""Test cases for ConvertTimeToolHandler."""
@pytest.fixture
def handler(self):
"""Create a ConvertTimeToolHandler instance."""
return ConvertTimeToolHandler()
def test_tool_description(self, handler):
"""Test the tool description is properly formatted."""
description = handler.get_tool_description()
assert description.name == "convert_time"
assert "convert time" in description.description.lower()
assert description.inputSchema["type"] == "object"
# Check that required parameters are supported
assert "datetime_str" in description.inputSchema["properties"]
assert "from_timezone" in description.inputSchema["properties"]
assert "to_timezone" in description.inputSchema["properties"]
assert "to_timezone" in description.inputSchema["properties"]
# Check that required fields include the timezone fields
assert "from_timezone" in description.inputSchema["required"]
assert "to_timezone" in description.inputSchema["required"]
@pytest.mark.asyncio
async def test_run_tool_success(self, handler):
"""Test successful time conversion."""
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
# Mock timezone creation
mock_get_tz.side_effect = lambda tz: ZoneInfo(tz)
with patch('dateutil.parser.parse') as mock_parse:
# Mock parsing the datetime string
source_time = datetime(2024, 1, 1, 12, 0, 0, tzinfo=ZoneInfo("UTC"))
mock_parse.return_value = source_time
args = {
"datetime_str": "2024-01-01T12:00:00",
"from_timezone": "UTC",
"to_timezone": "America/New_York"
}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
response_data = json.loads(result[0].text)
assert response_data["original_timezone"] == "UTC"
assert response_data["converted_timezone"] == "America/New_York"
# Check that the datetime includes timezone info
assert "2024-01-01T12:00:00" in response_data["original_datetime"]
assert "converted_datetime" in response_data
@pytest.mark.asyncio
async def test_run_tool_missing_required_args(self, handler):
"""Test tool execution with missing required arguments."""
args = {"datetime": "2024-01-01T12:00:00"} # Missing timezones
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Missing required arguments" in result[0].text
@pytest.mark.asyncio
async def test_run_tool_invalid_datetime_format(self, handler):
"""Test tool execution with invalid datetime format."""
with patch('dateutil.parser.parse') as mock_parse:
mock_parse.side_effect = ValueError("Invalid datetime format")
args = {
"datetime": "invalid-datetime",
"from_timezone": "UTC",
"to_timezone": "America/New_York"
}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Error converting time" in result[0].text
@pytest.mark.asyncio
async def test_run_tool_invalid_from_timezone(self, handler):
"""Test tool execution with invalid from_timezone."""
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
def side_effect(tz):
if tz == "Invalid/Timezone":
raise McpError(ErrorData(code=-1, message="Invalid timezone"))
return ZoneInfo(tz)
mock_get_tz.side_effect = side_effect
args = {
"datetime": "2024-01-01T12:00:00",
"from_timezone": "Invalid/Timezone",
"to_timezone": "UTC"
}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Error converting time" in result[0].text
@pytest.mark.asyncio
async def test_run_tool_invalid_to_timezone(self, handler):
"""Test tool execution with invalid to_timezone."""
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
def side_effect(tz):
if tz == "Invalid/Timezone":
raise McpError(ErrorData(code=-1, message="Invalid timezone"))
return ZoneInfo(tz)
mock_get_tz.side_effect = side_effect
with patch('dateutil.parser.parse') as mock_parse:
mock_parse.return_value = datetime(2024, 1, 1, 12, 0, 0, tzinfo=ZoneInfo("UTC"))
args = {
"datetime": "2024-01-01T12:00:00",
"from_timezone": "UTC",
"to_timezone": "Invalid/Timezone"
}
result = await handler.run_tool(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Error converting time" in result[0].text
@pytest.mark.asyncio
async def test_run_tool_same_timezone_conversion(self, handler):
"""Test time conversion between same timezones."""
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
mock_get_tz.return_value = ZoneInfo("UTC")
with patch('dateutil.parser.parse') as mock_parse:
source_time = datetime(2024, 1, 1, 12, 0, 0, tzinfo=ZoneInfo("UTC"))
mock_parse.return_value = source_time
args = {
"datetime_str": "2024-01-01T12:00:00",
"from_timezone": "UTC",
"to_timezone": "UTC"
}
result = await handler.run_tool(args)
assert len(result) == 1
response_data = json.loads(result[0].text)
# When converting to same timezone, time should remain the same
assert "2024-01-01T12:00:00" in response_data["original_datetime"]
assert response_data["original_timezone"] == "UTC"
assert response_data["converted_timezone"] == "UTC"
@pytest.mark.asyncio
async def test_run_tool_across_date_line(self, handler):
"""Test time conversion across international date line."""
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
mock_get_tz.side_effect = lambda tz: ZoneInfo(tz)
with patch('dateutil.parser.parse') as mock_parse:
# Late evening in one timezone
source_time = datetime(2024, 1, 1, 23, 0, 0, tzinfo=ZoneInfo("Pacific/Kiritimati"))
mock_parse.return_value = source_time
args = {
"datetime_str": "2024-01-01T23:00:00",
"from_timezone": "Pacific/Kiritimati",
"to_timezone": "Pacific/Honolulu"
}
result = await handler.run_tool(args)
assert len(result) == 1
response_data = json.loads(result[0].text)
assert response_data["original_timezone"] == "Pacific/Kiritimati"
assert response_data["converted_timezone"] == "Pacific/Honolulu"
# Should handle date line crossing correctly
assert "converted_datetime" in response_data
```
--------------------------------------------------------------------------------
/tests/test_performance.py:
--------------------------------------------------------------------------------
```python
"""
Performance and load tests for MCP Weather Server.
"""
import pytest
import asyncio
import time
from unittest.mock import AsyncMock, Mock, patch
from src.mcp_weather_server.server import (
register_all_tools,
call_tool,
tool_handlers
)
class TestPerformance:
"""Performance tests for the weather server."""
def setup_method(self):
"""Setup for each test."""
global tool_handlers
tool_handlers.clear()
register_all_tools()
@pytest.mark.asyncio
async def test_weather_request_performance(self):
"""Test that weather requests complete within reasonable time."""
mock_geo_response = {
"results": [{"latitude": 40.7128, "longitude": -74.0060}]
}
mock_weather_response = {
"hourly": {
"time": ["2024-01-01T12:00"],
"temperature_2m": [20.0],
"relative_humidity_2m": [60],
"dew_point_2m": [12.0],
"weather_code": [0]
}
}
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
def mock_get(url):
response = Mock()
response.status_code = 200
if "geocoding-api" in url:
response.json.return_value = mock_geo_response
else:
response.json.return_value = mock_weather_response
return response
mock_client.get.side_effect = mock_get
mock_client_class.return_value.__aenter__.return_value = mock_client
with patch('src.mcp_weather_server.utils.get_closest_utc_index', return_value=0):
start_time = time.time()
result = await call_tool("get_current_weather", {"city": "New York"})
end_time = time.time()
# Request should complete in less than 1 second (mocked)
assert end_time - start_time < 1.0
assert len(result) == 1
assert not result[0].text.startswith("Error:")
@pytest.mark.asyncio
async def test_concurrent_request_performance(self):
"""Test performance with multiple concurrent requests."""
mock_geo_response = {
"results": [{"latitude": 40.7128, "longitude": -74.0060}]
}
mock_weather_response = {
"hourly": {
"time": ["2024-01-01T12:00"],
"temperature_2m": [20.0],
"relative_humidity_2m": [60],
"dew_point_2m": [12.0],
"weather_code": [0]
}
}
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
def mock_get(url):
response = Mock()
response.status_code = 200
if "geocoding-api" in url:
response.json.return_value = mock_geo_response
else:
response.json.return_value = mock_weather_response
return response
mock_client.get.side_effect = mock_get
mock_client_class.return_value.__aenter__.return_value = mock_client
with patch('src.mcp_weather_server.utils.get_closest_utc_index', return_value=0):
# Create 10 concurrent requests
tasks = [
call_tool("get_current_weather", {"city": f"City{i}"})
for i in range(10)
]
start_time = time.time()
results = await asyncio.gather(*tasks)
end_time = time.time()
# All requests should complete in reasonable time
assert end_time - start_time < 2.0
assert len(results) == 10
# All requests should succeed
for result in results:
assert len(result) == 1
assert not result[0].text.startswith("Error:")
@pytest.mark.asyncio
async def test_time_tool_performance(self):
"""Test performance of time-related tools."""
from datetime import datetime
from zoneinfo import ZoneInfo
fixed_time = datetime(2024, 1, 1, 12, 0, 0, tzinfo=ZoneInfo("UTC"))
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
mock_get_tz.return_value = ZoneInfo("UTC")
with patch('src.mcp_weather_server.tools.tools_time.datetime') as mock_datetime:
mock_datetime.now.return_value = fixed_time
start_time = time.time()
result = await call_tool("get_current_datetime", {"timezone_name": "UTC"})
end_time = time.time()
# Time tools should be very fast
assert end_time - start_time < 0.1
assert len(result) == 1
assert not result[0].text.startswith("Error:")
@pytest.mark.asyncio
async def test_memory_usage_with_many_requests(self):
"""Test that memory usage doesn't grow excessively with many requests."""
import gc
import sys
# Get initial memory usage (approximation)
gc.collect()
initial_objects = len(gc.get_objects())
mock_geo_response = {
"results": [{"latitude": 40.7128, "longitude": -74.0060}]
}
mock_weather_response = {
"hourly": {
"time": ["2024-01-01T12:00"],
"temperature_2m": [20.0],
"relative_humidity_2m": [60],
"dew_point_2m": [12.0],
"weather_code": [0]
}
}
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
def mock_get(url):
response = Mock()
response.status_code = 200
if "geocoding-api" in url:
response.json.return_value = mock_geo_response
else:
response.json.return_value = mock_weather_response
return response
mock_client.get.side_effect = mock_get
mock_client_class.return_value.__aenter__.return_value = mock_client
with patch('src.mcp_weather_server.utils.get_closest_utc_index', return_value=0):
# Make many sequential requests
for i in range(50):
result = await call_tool("get_current_weather", {"city": f"City{i}"})
assert len(result) == 1
assert not result[0].text.startswith("Error:")
# Force garbage collection and check memory
gc.collect()
final_objects = len(gc.get_objects())
# Object count shouldn't grow dramatically
# Allow some growth but not excessive (factor of 2)
assert final_objects < initial_objects * 2
class TestLoadTesting:
"""Load testing scenarios."""
def setup_method(self):
"""Setup for each test."""
global tool_handlers
tool_handlers.clear()
register_all_tools()
@pytest.mark.asyncio
async def test_burst_load_handling(self):
"""Test handling of burst load (many requests at once)."""
mock_geo_response = {
"results": [{"latitude": 40.7128, "longitude": -74.0060}]
}
mock_weather_response = {
"hourly": {
"time": ["2024-01-01T12:00"],
"temperature_2m": [20.0],
"relative_humidity_2m": [60],
"dew_point_2m": [12.0],
"weather_code": [0]
}
}
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
def mock_get(url):
response = Mock()
response.status_code = 200
if "geocoding-api" in url:
response.json.return_value = mock_geo_response
else:
response.json.return_value = mock_weather_response
return response
mock_client.get.side_effect = mock_get
mock_client_class.return_value.__aenter__.return_value = mock_client
with patch('src.mcp_weather_server.utils.get_closest_utc_index', return_value=0):
# Create a large burst of requests (simulate high load)
burst_size = 25
tasks = [
call_tool("get_current_weather", {"city": f"BurstCity{i}"})
for i in range(burst_size)
]
start_time = time.time()
results = await asyncio.gather(*tasks, return_exceptions=True)
end_time = time.time()
# Check that most requests succeeded
successful_results = [r for r in results if not isinstance(r, Exception)]
error_results = [r for r in results if isinstance(r, Exception)]
# At least 80% should succeed under load
success_rate = len(successful_results) / len(results)
assert success_rate >= 0.8, f"Success rate {success_rate} too low"
# Total time should be reasonable even under load
assert end_time - start_time < 5.0
@pytest.mark.asyncio
async def test_mixed_tool_load(self):
"""Test load with mixed tool types."""
from datetime import datetime
from zoneinfo import ZoneInfo
# Setup weather mocks
mock_geo_response = {
"results": [{"latitude": 40.7128, "longitude": -74.0060}]
}
mock_weather_response = {
"hourly": {
"time": ["2024-01-01T12:00"],
"temperature_2m": [20.0],
"relative_humidity_2m": [60],
"dew_point_2m": [12.0],
"weather_code": [0]
}
}
# Setup time mocks
fixed_time = datetime(2024, 1, 1, 12, 0, 0, tzinfo=ZoneInfo("UTC"))
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
def mock_get(url):
response = Mock()
response.status_code = 200
if "geocoding-api" in url:
response.json.return_value = mock_geo_response
else:
response.json.return_value = mock_weather_response
return response
mock_client.get.side_effect = mock_get
mock_client_class.return_value.__aenter__.return_value = mock_client
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
mock_get_tz.return_value = ZoneInfo("UTC")
with patch('src.mcp_weather_server.tools.tools_time.datetime') as mock_datetime:
mock_datetime.now.return_value = fixed_time
with patch('src.mcp_weather_server.utils.get_closest_utc_index', return_value=0):
# Create mixed requests
tasks = []
for i in range(20):
if i % 3 == 0:
tasks.append(call_tool("get_current_weather", {"city": f"City{i}"}))
elif i % 3 == 1:
tasks.append(call_tool("get_current_datetime", {"timezone_name": "UTC"}))
else:
tasks.append(call_tool("get_timezone_info", {"timezone_name": "UTC"}))
start_time = time.time()
results = await asyncio.gather(*tasks, return_exceptions=True)
end_time = time.time()
# Check results
successful_results = [r for r in results if not isinstance(r, Exception)]
success_rate = len(successful_results) / len(results)
assert success_rate >= 0.9, f"Mixed load success rate {success_rate} too low"
assert end_time - start_time < 3.0
assert len(results) == 20
class TestStressConditions:
"""Stress testing under adverse conditions."""
def setup_method(self):
"""Setup for each test."""
global tool_handlers
tool_handlers.clear()
register_all_tools()
@pytest.mark.asyncio
async def test_high_error_rate_resilience(self):
"""Test resilience when external APIs have high error rates."""
error_count = 0
success_count = 0
def mock_get_with_errors(url):
nonlocal error_count, success_count
response = Mock()
# Simulate 30% error rate
if (error_count + success_count) % 10 < 3:
error_count += 1
response.status_code = 500
else:
success_count += 1
response.status_code = 200
if "geocoding-api" in url:
response.json.return_value = {
"results": [{"latitude": 40.7128, "longitude": -74.0060}]
}
else:
response.json.return_value = {
"hourly": {
"time": ["2024-01-01T12:00"],
"temperature_2m": [20.0],
"relative_humidity_2m": [60],
"dew_point_2m": [12.0],
"weather_code": [0]
}
}
return response
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
mock_client.get.side_effect = mock_get_with_errors
mock_client_class.return_value.__aenter__.return_value = mock_client
with patch('src.mcp_weather_server.utils.get_closest_utc_index', return_value=0):
# Make requests despite high error rate
tasks = [
call_tool("get_current_weather", {"city": f"ErrorTestCity{i}"})
for i in range(20)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Count successes and errors
successful_results = []
error_results = []
for result in results:
if isinstance(result, Exception):
error_results.append(result)
else:
if result[0].text.startswith("Error:"):
error_results.append(result)
else:
successful_results.append(result)
# Should handle errors gracefully without crashing
assert len(results) == 20
# Some should succeed despite errors
assert len(successful_results) > 0
# Error handling should be proper
for error_result in error_results:
if not isinstance(error_result, Exception):
assert "Error:" in error_result[0].text
```
--------------------------------------------------------------------------------
/tests/test_integration.py:
--------------------------------------------------------------------------------
```python
"""
Integration tests for the MCP Weather Server.
These tests verify end-to-end functionality with real-world scenarios.
"""
import pytest
import asyncio
import json
from unittest.mock import AsyncMock, Mock, patch
from mcp.types import ErrorData
from mcp import McpError
from src.mcp_weather_server.server import (
register_all_tools,
list_tools,
call_tool,
tool_handlers
)
class TestWeatherIntegration:
"""Integration tests for weather functionality."""
def setup_method(self):
"""Setup for each test."""
global tool_handlers
tool_handlers.clear()
register_all_tools()
@pytest.mark.asyncio
async def test_current_weather_integration(self):
"""Test complete current weather workflow."""
# Mock HTTP responses
mock_geo_response = {
"results": [{"latitude": 40.7128, "longitude": -74.0060}]
}
mock_weather_response = {
"hourly": {
"time": ["2024-01-01T12:00"],
"temperature_2m": [22.5],
"relative_humidity_2m": [65],
"dew_point_2m": [15.2],
"weather_code": [1]
}
}
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
def mock_get(url):
response = Mock()
response.status_code = 200
if "geocoding-api" in url:
response.json.return_value = mock_geo_response
else:
response.json.return_value = mock_weather_response
return response
mock_client.get.side_effect = mock_get
mock_client_class.return_value.__aenter__.return_value = mock_client
with patch('src.mcp_weather_server.utils.get_closest_utc_index', return_value=0):
# Test the complete workflow
tools = await list_tools()
assert any(tool.name == "get_current_weather" for tool in tools)
result = await call_tool("get_current_weather", {"city": "New York"})
assert len(result) == 1
response_text = result[0].text
assert "New York" in response_text
assert "22.5°C" in response_text
assert "Mainly clear" in response_text
@pytest.mark.asyncio
async def test_weather_range_integration(self):
"""Test complete weather range workflow."""
mock_geo_response = {
"results": [{"latitude": 51.5074, "longitude": -0.1278}]
}
mock_weather_response = {
"hourly": {
"time": ["2024-01-01T00:00", "2024-01-01T12:00", "2024-01-02T00:00"],
"temperature_2m": [8.0, 12.0, 6.0],
"relative_humidity_2m": [85, 70, 90],
"dew_point_2m": [6.0, 7.0, 4.5],
"weather_code": [3, 1, 61]
}
}
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
def mock_get(url):
response = Mock()
response.status_code = 200
if "geocoding-api" in url:
response.json.return_value = mock_geo_response
else:
response.json.return_value = mock_weather_response
return response
mock_client.get.side_effect = mock_get
mock_client_class.return_value.__aenter__.return_value = mock_client
result = await call_tool("get_weather_byDateTimeRange", {
"city": "London",
"start_date": "2024-01-01",
"end_date": "2024-01-02"
})
assert len(result) == 1
response_text = result[0].text
assert "London" in response_text
assert "analyze" in response_text.lower()
assert "2024-01-01" in response_text
@pytest.mark.asyncio
async def test_weather_details_integration(self):
"""Test weather details JSON output."""
mock_geo_response = {
"results": [{"latitude": 48.8566, "longitude": 2.3522}]
}
mock_weather_response = {
"hourly": {
"time": ["2024-01-01T14:00"],
"temperature_2m": [18.5],
"relative_humidity_2m": [72],
"dew_point_2m": [13.8],
"weather_code": [2]
}
}
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
def mock_get(url):
response = Mock()
response.status_code = 200
if "geocoding-api" in url:
response.json.return_value = mock_geo_response
else:
response.json.return_value = mock_weather_response
return response
mock_client.get.side_effect = mock_get
mock_client_class.return_value.__aenter__.return_value = mock_client
with patch('src.mcp_weather_server.utils.get_closest_utc_index', return_value=0):
result = await call_tool("get_weather_details", {"city": "Paris"})
assert len(result) == 1
# Parse the JSON response
weather_data = json.loads(result[0].text)
assert weather_data["city"] == "Paris"
assert weather_data["latitude"] == 48.8566
assert weather_data["longitude"] == 2.3522
assert weather_data["temperature_c"] == 18.5
assert weather_data["weather_description"] == "Partly cloudy"
class TestTimeIntegration:
"""Integration tests for time functionality."""
def setup_method(self):
"""Setup for each test."""
global tool_handlers
tool_handlers.clear()
register_all_tools()
@pytest.mark.asyncio
async def test_current_datetime_integration(self):
"""Test complete current datetime workflow."""
from datetime import datetime
from zoneinfo import ZoneInfo
fixed_time = datetime(2024, 6, 15, 10, 30, 45, tzinfo=ZoneInfo("America/New_York"))
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
mock_get_tz.return_value = ZoneInfo("America/New_York")
with patch('src.mcp_weather_server.tools.tools_time.datetime') as mock_datetime:
mock_datetime.now.return_value = fixed_time
result = await call_tool("get_current_datetime", {
"timezone_name": "America/New_York"
})
assert len(result) == 1
time_data = json.loads(result[0].text)
assert time_data["timezone"] == "America/New_York"
assert "2024-06-15T10:30:45" in time_data["datetime"]
@pytest.mark.asyncio
async def test_timezone_info_integration(self):
"""Test complete timezone info workflow."""
from datetime import datetime
from zoneinfo import ZoneInfo
fixed_time = datetime(2024, 12, 25, 15, 0, 0, tzinfo=ZoneInfo("Europe/London"))
fixed_utc_time = datetime(2024, 12, 25, 15, 0, 0) # UTC time without timezone
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
mock_get_tz.return_value = ZoneInfo("Europe/London")
with patch('src.mcp_weather_server.tools.tools_time.datetime') as mock_datetime:
mock_datetime.now.return_value = fixed_time
mock_datetime.utcnow.return_value = fixed_utc_time
result = await call_tool("get_timezone_info", {
"timezone_name": "Europe/London"
})
assert len(result) == 1
tz_data = json.loads(result[0].text)
assert tz_data["timezone_name"] == "Europe/London"
assert "current_local_time" in tz_data
assert "utc_offset_hours" in tz_data
@pytest.mark.asyncio
async def test_time_conversion_integration(self):
"""Test complete time conversion workflow."""
from datetime import datetime
from zoneinfo import ZoneInfo
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
mock_get_tz.side_effect = lambda tz: ZoneInfo(tz)
with patch('dateutil.parser.parse') as mock_parse:
source_time = datetime(2024, 7, 4, 12, 0, 0, tzinfo=ZoneInfo("UTC"))
mock_parse.return_value = source_time
result = await call_tool("convert_time", {
"datetime_str": "2024-07-04T12:00:00",
"from_timezone": "UTC",
"to_timezone": "America/Los_Angeles"
})
assert len(result) == 1
conversion_data = json.loads(result[0].text)
assert "2024-07-04T12:00:00" in conversion_data["original_datetime"]
assert conversion_data["original_timezone"] == "UTC"
assert conversion_data["converted_timezone"] == "America/Los_Angeles"
assert "converted_datetime" in conversion_data
class TestErrorHandlingIntegration:
"""Integration tests for error handling scenarios."""
def setup_method(self):
"""Setup for each test."""
global tool_handlers
tool_handlers.clear()
register_all_tools()
@pytest.mark.asyncio
async def test_invalid_city_error_handling(self):
"""Test error handling for invalid city names."""
# Mock empty geocoding response
mock_geo_response = {"results": []}
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = mock_geo_response
mock_client.get.return_value = mock_response
mock_client_class.return_value.__aenter__.return_value = mock_client
result = await call_tool("get_current_weather", {"city": "NonexistentCity"})
assert len(result) == 1
assert "Error:" in result[0].text
assert "coordinates" in result[0].text.lower()
@pytest.mark.asyncio
async def test_api_error_handling(self):
"""Test handling of API errors."""
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
mock_response = Mock()
mock_response.status_code = 500 # Server error
mock_client.get.return_value = mock_response
mock_client_class.return_value.__aenter__.return_value = mock_client
result = await call_tool("get_current_weather", {"city": "TestCity"})
assert len(result) == 1
assert "Error:" in result[0].text
assert "500" in result[0].text
@pytest.mark.asyncio
async def test_network_error_handling(self):
"""Test handling of network errors."""
import httpx
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
mock_client.get.side_effect = httpx.RequestError("Network unavailable")
mock_client_class.return_value.__aenter__.return_value = mock_client
result = await call_tool("get_current_weather", {"city": "TestCity"})
assert len(result) == 1
assert "Error:" in result[0].text
assert "Network" in result[0].text or "network" in result[0].text
@pytest.mark.asyncio
async def test_invalid_timezone_error_handling(self):
"""Test error handling for invalid timezones."""
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
mock_get_tz.side_effect = McpError(ErrorData(code=-1, message="Invalid timezone: BadTimezone"))
result = await call_tool("get_current_datetime", {
"timezone_name": "BadTimezone"
})
assert len(result) == 1
assert "Error getting current time" in result[0].text
@pytest.mark.asyncio
async def test_missing_arguments_error_handling(self):
"""Test error handling for missing required arguments."""
# Test weather tool without city
result = await call_tool("get_current_weather", {})
assert len(result) == 1
assert "Missing required arguments: city" in result[0].text
# Test time conversion without required fields
result = await call_tool("convert_time", {"datetime": "2024-01-01T12:00:00"})
assert len(result) == 1
assert "Missing required arguments" in result[0].text
class TestConcurrentOperations:
"""Integration tests for concurrent operations."""
def setup_method(self):
"""Setup for each test."""
global tool_handlers
tool_handlers.clear()
register_all_tools()
@pytest.mark.asyncio
async def test_concurrent_weather_requests(self):
"""Test concurrent weather requests."""
mock_responses = {
"New York": {
"geo": {"results": [{"latitude": 40.7128, "longitude": -74.0060}]},
"weather": {
"hourly": {
"time": ["2024-01-01T12:00"],
"temperature_2m": [20.0],
"relative_humidity_2m": [60],
"dew_point_2m": [12.0],
"weather_code": [0]
}
}
},
"London": {
"geo": {"results": [{"latitude": 51.5074, "longitude": -0.1278}]},
"weather": {
"hourly": {
"time": ["2024-01-01T12:00"],
"temperature_2m": [15.0],
"relative_humidity_2m": [80],
"dew_point_2m": [11.5],
"weather_code": [3]
}
}
}
}
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
def mock_get(url):
response = Mock()
response.status_code = 200
# Default to New York for any unmapped requests
default_geo = mock_responses["New York"]["geo"]
default_weather = mock_responses["New York"]["weather"]
# Determine which city based on the URL parameters
if "name=New%20York" in url or "name=New+York" in url:
if "geocoding-api" in url:
response.json.return_value = mock_responses["New York"]["geo"]
else:
response.json.return_value = mock_responses["New York"]["weather"]
elif "name=London" in url:
if "geocoding-api" in url:
response.json.return_value = mock_responses["London"]["geo"]
else:
response.json.return_value = mock_responses["London"]["weather"]
elif "geocoding-api" in url:
# Default geocoding response
response.json.return_value = default_geo
else:
# Default weather response based on coordinates
if "latitude=40.7128" in url:
response.json.return_value = mock_responses["New York"]["weather"]
elif "latitude=51.5074" in url:
response.json.return_value = mock_responses["London"]["weather"]
else:
response.json.return_value = default_weather
return response
mock_client.get.side_effect = mock_get
mock_client_class.return_value.__aenter__.return_value = mock_client
with patch('src.mcp_weather_server.utils.get_closest_utc_index', return_value=0):
# Run concurrent requests
tasks = [
call_tool("get_current_weather", {"city": "New York"}),
call_tool("get_current_weather", {"city": "London"}),
]
results = await asyncio.gather(*tasks)
assert len(results) == 2
# Check that both requests completed successfully
for result in results:
assert len(result) == 1
assert not result[0].text.startswith("Error:")
# Verify different cities returned different results
result_texts = [result[0].text for result in results]
assert any("New York" in text for text in result_texts)
assert any("London" in text for text in result_texts)
@pytest.mark.asyncio
async def test_mixed_tool_concurrent_requests(self):
"""Test concurrent requests to different types of tools."""
from datetime import datetime
from zoneinfo import ZoneInfo
# Setup mocks for weather
mock_geo_response = {
"results": [{"latitude": 35.6762, "longitude": 139.6503}]
}
mock_weather_response = {
"hourly": {
"time": ["2024-01-01T12:00"],
"temperature_2m": [25.0],
"relative_humidity_2m": [70],
"dew_point_2m": [18.0],
"weather_code": [1]
}
}
# Setup mocks for time
fixed_time = datetime(2024, 1, 1, 9, 0, 0, tzinfo=ZoneInfo("Asia/Tokyo"))
with patch('httpx.AsyncClient') as mock_client_class:
mock_client = AsyncMock()
def mock_get(url):
response = Mock()
response.status_code = 200
if "geocoding-api" in url:
response.json.return_value = mock_geo_response
else:
response.json.return_value = mock_weather_response
return response
mock_client.get.side_effect = mock_get
mock_client_class.return_value.__aenter__.return_value = mock_client
with patch('src.mcp_weather_server.utils.get_zoneinfo') as mock_get_tz:
mock_get_tz.return_value = ZoneInfo("Asia/Tokyo")
with patch('src.mcp_weather_server.tools.tools_time.datetime') as mock_datetime:
mock_datetime.now.return_value = fixed_time
with patch('src.mcp_weather_server.utils.get_closest_utc_index', return_value=0):
# Run concurrent requests to different tools
tasks = [
call_tool("get_current_weather", {"city": "Tokyo"}),
call_tool("get_current_datetime", {"timezone_name": "Asia/Tokyo"}),
]
results = await asyncio.gather(*tasks)
assert len(results) == 2
# Verify weather result
weather_result = results[0]
assert len(weather_result) == 1
assert "Tokyo" in weather_result[0].text
assert "25.0°C" in weather_result[0].text
# Verify time result
time_result = results[1]
assert len(time_result) == 1
time_data = json.loads(time_result[0].text)
assert time_data["timezone"] == "Asia/Tokyo"
```