CHAPTER 1: THE WORLD HAS CHANGED, AND SO HAS THE WAY WE BUILD BACKENDS
There is a moment in the life of every software engineer when a technology arrives that does not just add a new tool to the toolbox but instead fundamentally reshapes what the toolbox is for. The Model Context Protocol, MCP for short, is one of those moments. If you have spent years building REST APIs, GraphQL endpoints, or gRPC services, you already know how to make systems talk to each other. What MCP does is something subtler and far more powerful: it defines how an intelligent agent, driven by a large language model, talks to those systems. It gives the LLM a structured, standardized, and safe grammar for reaching out and touching the world.
Before MCP, connecting an LLM to an external service was an artisanal, hand-crafted affair. Every team invented its own function-calling schemas, its own tool-dispatch loops, its own error-handling conventions. The result was a landscape of incompatible integrations that each had to be maintained, updated, and debugged in isolation. MCP changes that. It is the USB standard for AI tool connectivity: once you build an MCP server, any MCP-compatible client, whether it is Claude Fable 5.1, GPT-6 Astra, Gemini 3.8 Flash, or a local Llama 4 model running on your own hardware, can discover and use your server's capabilities without any additional glue code.
This tutorial will take you all the way from first principles to a fully production-ready MCP server. The example we will use throughout is the OpenWeatherMap API, one of the most widely used REST APIs on the planet, with millions of active developers and a rich set of endpoints covering current conditions, multi-day forecasts, historical data, and air quality. It is the perfect vehicle for our journey because it is familiar enough to be immediately understandable, yet rich enough to let us explore every interesting corner of MCP server design.
By the time you reach the final chapter, you will have built a server that exposes weather data as MCP tools and resources, handles errors gracefully, supports both remote cloud LLMs and local models running through Ollama, and is structured according to clean architecture principles that will make it a pleasure to extend and maintain. Let us begin.
CHAPTER 2: UNDERSTANDING MCP FROM THE GROUND UP
The Model Context Protocol was originally introduced by Anthropic in late 2024. In the time since, it has been adopted by virtually every major AI lab and has become the de facto standard for LLM-to-tool connectivity. The specification that matters for us today is version 2026-07-28, and it represents a major architectural shift from earlier versions that is worth understanding deeply before we write a single line of code.
The most important thing to understand about the 2026-07-28 specification is that MCP is now a fully stateless, request-and-response protocol. Earlier versions of MCP were stateful: a client would establish a session with a server through an initialization handshake, maintain that session across multiple interactions, and the server would remember context between calls. This was powerful in theory but created significant operational headaches. Stateful servers are hard to scale horizontally. They require sticky load balancing. They accumulate memory. They fail in subtle ways when sessions are dropped.
The 2026 specification throws all of that away. Every single request to an MCP server is now completely self-describing. The protocol version, the client's identity, and its declared capabilities all travel inside a "_meta" field within the request body. There are no session IDs to track. There are no handshakes to perform. The server processes each request in complete isolation from every other request, which means you can put as many server instances behind a load balancer as you like and route requests in round-robin fashion without a care in the world.
To make this stateless routing even more efficient, the 2026 spec introduces standardized HTTP headers that allow load balancers and proxies to route requests without needing to parse the JSON body. The "Mcp-Method" header tells infrastructure what kind of MCP operation is being performed, and the "Mcp-Name" header carries the name of the specific tool, resource, or prompt being invoked. This is a thoughtful, operations-friendly design that shows the protocol has matured from a research prototype into something built for real production environments.
The three fundamental primitives that an MCP server can expose are tools, resources, and prompts. Understanding the difference between these three is crucial, because choosing the wrong one for a given capability leads to confusion for the LLM and poor user experiences.
A tool is an executable action. It is something the LLM can call to make something happen or to retrieve information that requires computation. Tools can have side effects. They accept typed input parameters and return structured output. In our weather server, "get the current temperature in Berlin" is a tool call, because it requires going out to the OpenWeatherMap API, fetching live data, and returning a result.
A resource is a piece of data that can be read. Resources are more like files or documents than functions. They have a URI that identifies them, and they can be fetched by the client. In our weather server, a resource might be the complete weather report for a city, expressed as a structured document that the LLM can read and reason about. Resources are ideal for providing the LLM with context: background information, configuration data, or reference material.
A prompt is a reusable template that helps the LLM formulate better requests. Prompts are less about doing things and more about guiding the LLM toward asking the right questions in the right way. In our server, a prompt might be a template for asking "what should I wear today given the forecast?" that structures the LLM's reasoning about clothing choices based on weather data.
The flow of a typical MCP interaction looks like this:
LLM Client (e.g., Claude Fable 5.1)
|
| 1. POST /mcp (tools/list)
| Mcp-Method: tools/list
v
MCP Server
|
| 2. Returns JSON list of all available tools
| with their names, descriptions, and input schemas
v
LLM Client
|
| 3. LLM reasons about user's request and picks a tool
|
| 4. POST /mcp (tools/call)
| Mcp-Method: tools/call
| Mcp-Name: get_current_weather
| Body: { "name": "get_current_weather",
| "arguments": { "city": "Berlin",
| "units": "metric" } }
v
MCP Server
|
| 5. Validates input, calls OpenWeatherMap REST API
|
| 6. Returns structured result to LLM client
v
LLM Client
|
| 7. LLM incorporates result into its response to user
v
End User receives a natural language answer backed by real data
This flow is elegant in its simplicity. The LLM does not need to know anything about HTTP, authentication, or the OpenWeatherMap API's quirks. It just knows about tools, and the MCP server handles all the messy details of actually talking to the underlying REST API.
CHAPTER 3: THE OPENWEATHERMAP API - OUR CANVAS
Before we start building our MCP server, let us spend some quality time understanding the API we are wrapping. The OpenWeatherMap API is a REST API that provides weather data for locations all over the world. It has been around since 2012, it has tens of millions of registered users, and it offers a generous free tier that makes it perfect for learning and experimentation. You can get your own API key by registering at openweathermap.org, and the free tier gives you 1,000 API calls per day, which is more than enough for everything we will do in this tutorial.
The endpoints we will wrap in our MCP server cover three major use cases. The first is current weather data, available at the path /data/2.5/weather. This endpoint accepts a city name, a geographic coordinate pair, or a zip code, and returns a rich JSON document describing the current atmospheric conditions. The second is the five-day forecast, available at /data/2.5/forecast, which returns weather predictions at three-hour intervals for the next five days. The third is the One Call API 3.0 at /data/3.0/onecall, which is the most powerful endpoint of all: it returns current conditions, minute-by-minute precipitation for the next hour, hourly forecasts for 48 hours, daily forecasts for eight days, and weather alerts, all in a single API call.
The base URL for all API calls is https://api.openweathermap.org. Authentication is handled by appending an "appid" query parameter to every request, or alternatively by sending the key in an "X-Api-Key" HTTP header. The API returns JSON by default, and the response structure is consistent and well-documented.
A typical response from the /data/2.5/weather endpoint looks like this when you request weather for London:
{
"coord": { "lon": -0.1257, "lat": 51.5085 },
"weather": [
{
"id": 800,
"main": "Clear",
"description": "clear sky",
"icon": "01d"
}
],
"main": {
"temp": 18.5,
"feels_like": 17.9,
"temp_min": 16.2,
"temp_max": 20.1,
"pressure": 1018,
"humidity": 62
},
"visibility": 10000,
"wind": { "speed": 3.6, "deg": 220 },
"dt": 1725696000,
"name": "London",
"cod": 200
}
The "main" object contains the core temperature and pressure readings. The "weather" array describes the sky conditions. The "wind" object gives speed in meters per second and direction in degrees. All of this rich, structured data is exactly what we want to expose through our MCP server, translated into a form that an LLM can reason about naturally.
One important operational detail: the OpenWeatherMap API uses rate limiting. The free tier allows 60 calls per minute. Our MCP server will need to handle the case where we exceed that limit gracefully, returning a meaningful error to the LLM rather than crashing or returning garbage. This is one of those real-world details that separates a tutorial project from production software, and we will handle it properly.
CHAPTER 4: ARCHITECTURE - THINKING BEFORE TYPING
The single most important thing you can do before writing any code is to think carefully about the architecture of your system. A badly architected MCP server will work fine for a demo but will become a maintenance nightmare the moment you need to add a new tool, change an API endpoint, or swap out one LLM for another. We are going to build something that is genuinely well-structured.
Our architecture follows the hexagonal architecture pattern, also known as ports and adapters. The core idea is that your business logic sits in the center, completely isolated from external concerns. External systems, whether they are REST APIs, databases, or LLM clients, connect to the core through well-defined interfaces called ports. The concrete implementations of those interfaces are called adapters.
In our weather MCP server, the architecture looks like this:
```
+-----------------------------------------------------------------+
| MCP SERVER PROCESS |
| |
| +------------------+ +------------------------------+ |
| | MCP LAYER | | WEATHER SERVICE LAYER | |
| | | | | |
| | Tools | ----> | WeatherService | |
| | Resources | | (business logic) | |
| | Prompts | | | |
| +------------------+ +------------------------------+ |
| | |
| +-----------v------------------+ |
| | HTTP ADAPTER LAYER | |
| | | |
| | OpenWeatherMapClient | |
| | (httpx async HTTP client) | |
| | TokenBucketRateLimiter | |
| | Response mapper | |
| +-----------+------------------+ |
| | |
+-----------------------------------------|---------------------+
|
INTERNET / NETWORK
|
+-------------v-----------+
| OpenWeatherMap REST API|
| api.openweathermap.org |
+-------------------------+
The project directory structure on disk mirrors this architecture precisely. Every directory has a single, clear responsibility, and no module reaches across boundaries it should not cross.
weather-mcp-server/
|
+-- src/
| +-- weather_mcp/
| +-- __init__.py
| +-- server.py (MCP server entry point)
| +-- config.py (configuration and settings)
| |
| +-- domain/
| | +-- __init__.py
| | +-- models.py (Pydantic domain models)
| | +-- exceptions.py (domain-specific exceptions)
| |
| +-- service/
| | +-- __init__.py
| | +-- weather_service.py (business logic)
| |
| +-- adapters/
| | +-- __init__.py
| | +-- owm_client.py (OpenWeatherMap HTTP adapter)
| | +-- rate_limiter.py
| |
| +-- tools/
| +-- __init__.py
| +-- weather_tools.py (MCP tool definitions)
| +-- weather_resources.py (MCP resource definitions)
| +-- weather_prompts.py (MCP prompt definitions)
|
+-- tests/
| +-- __init__.py
| +-- conftest.py
| +-- test_tools.py
| +-- test_client_integration.py
|
+-- clients/
| +-- cloud_llm_client.py (client using GPT-6/Claude/Gemini)
| +-- local_llm_client.py (client using Ollama + Llama 4)
|
+-- pyproject.toml
+-- .env.example
+-- README.md
This structure is not arbitrary. The "domain" layer contains pure Python data classes and exceptions that have zero external dependencies. They do not know about HTTP, MCP, or OpenWeatherMap. The "service" layer contains business logic that depends only on the domain layer. The "adapters" layer contains the concrete implementations that know how to talk to external systems. The "tools" layer is where MCP-specific concerns live: it translates between the MCP protocol and the service layer. This separation means that if OpenWeatherMap changes their API tomorrow, you only need to change the adapter. If Anthropic changes the MCP specification, you only need to change the tools layer. The business logic in the service layer never changes for either reason.
CHAPTER 5: SETTING UP THE PROJECT
Let us get our hands dirty. We will start by setting up the project with a modern Python toolchain. We are using Python 3.12 or higher, which gives us the latest type system improvements that make MCP schema generation work beautifully. The package manager we will use is uv, the Rust-based Python package manager that has become the industry standard by 2026 because of its speed and reliability.
Open a terminal and run these commands in a bash-script to create the project scaffold:
# Create the project directory and navigate into it
mkdir weather-mcp-server
cd weather-mcp-server
# Initialize a new Python project with uv.
# This creates pyproject.toml and a virtual environment automatically.
uv init --name weather-mcp --python 3.12
# Install the core runtime dependencies.
# mcp[cli] is the official MCP Python SDK, version 2.0.0 or higher.
# fastapi and uvicorn power the HTTP server that hosts the MCP endpoint.
# httpx is the async HTTP client we use to call OpenWeatherMap.
# pydantic and pydantic-settings handle data validation and configuration.
# python-dotenv loads our API keys from a .env file during development.
uv add "mcp[cli]>=2.0.0" fastapi uvicorn httpx pydantic \
pydantic-settings python-dotenv
# Install the LLM provider client libraries.
# These are used by the client scripts in the clients/ directory.
# The ollama library talks to a locally running Ollama inference server.
# openai is used for GPT-6 Astra.
# anthropic is used for Claude Fable 5.1.
# google-genai is the 2026 package name for Google's Generative AI SDK.
uv add ollama openai anthropic google-genai
# Install development and testing dependencies.
# pytest-asyncio enables async test functions.
# respx provides HTTP mocking for httpx in tests.
uv add --dev pytest pytest-asyncio pytest-httpx respx
```
Now let us create the directory structure. The following commands create every directory and touch every file we will need:
# Create the source package directories
mkdir -p src/weather_mcp/domain
mkdir -p src/weather_mcp/service
mkdir -p src/weather_mcp/adapters
mkdir -p src/weather_mcp/tools
mkdir -p tests
mkdir -p clients
# Create the __init__.py files that make Python recognize these as packages.
# All of these files are intentionally empty; they exist purely as markers.
touch src/weather_mcp/__init__.py
touch src/weather_mcp/domain/__init__.py
touch src/weather_mcp/service/__init__.py
touch src/weather_mcp/adapters/__init__.py
touch src/weather_mcp/tools/__init__.py
touch tests/__init__.py
The pyproject.toml file is the heart of our project configuration. It declares our dependencies, our entry points, and our tooling settings. Notice that fastapi and uvicorn are listed explicitly: the server module imports both, and without them the application will fail to start.
# pyproject.toml
# Project metadata and dependency configuration for the Weather MCP Server.
# This file is the single source of truth for everything the project needs.
[project]
name = "weather-mcp"
version = "1.0.0"
description = "A production-ready MCP server wrapping the OpenWeatherMap REST API"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
# MCP Python SDK (stateless HTTP transport, 2026-07-28 spec)
"mcp[cli]>=2.0.0",
# Web framework and ASGI server for hosting the MCP endpoint
"fastapi>=0.115.0",
"uvicorn[standard]>=0.30.0",
# Async HTTP client for calling the OpenWeatherMap REST API
"httpx>=0.27.0",
# Data validation, settings management, and schema generation
"pydantic>=2.7.0",
"pydantic-settings>=2.3.0",
# Loads .env files during development
"python-dotenv>=1.0.0",
# LLM provider client libraries (used by clients/ scripts)
"ollama>=0.3.0",
"openai>=1.40.0",
"anthropic>=0.34.0",
"google-genai>=1.0.0",
]
[project.scripts]
# This creates a 'weather-mcp' command-line entry point.
# After installing with 'uv sync', run 'weather-mcp' to start the server.
weather-mcp = "weather_mcp.server:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["src/weather_mcp"]
[tool.pytest.ini_options]
# asyncio_mode = "auto" means every async test function is automatically
# treated as an async test without needing the @pytest.mark.asyncio decorator.
asyncio_mode = "auto"
testpaths = ["tests"]
[tool.ruff]
line-length = 88
target-version = "py312"
Before we write any Python, we need to set up our environment variables. The .env.example file serves as the template that every developer on the team copies to create their own private .env file:
# .env.example
# Copy this file to .env and fill in your actual values.
# The .env file must NEVER be committed to version control.
# Add '.env' to your .gitignore file immediately.
# Your OpenWeatherMap API key.
# Get one for free at https://openweathermap.org/api
OWM_API_KEY=your_openweathermap_api_key_here
# The base URL for the OpenWeatherMap API.
# Override this in tests to point to a mock server.
OWM_BASE_URL=https://api.openweathermap.org
# The default unit system for temperature measurements.
# Accepted values: metric (Celsius), imperial (Fahrenheit), standard (Kelvin)
OWM_DEFAULT_UNITS=metric
# The host and port on which the MCP server will listen.
# Use 0.0.0.0 to accept connections from all network interfaces.
MCP_HOST=0.0.0.0
MCP_PORT=8000
# The full URL of the MCP server's endpoint.
# Used by client scripts to connect to the server.
MCP_SERVER_URL=http://localhost:8000/mcp
# API keys for cloud LLM providers (used by clients/cloud_llm_client.py)
OPENAI_API_KEY=your_openai_api_key_here
ANTHROPIC_API_KEY=your_anthropic_api_key_here
GOOGLE_API_KEY=your_google_api_key_here
# The URL of a locally running Ollama instance.
# Default is localhost:11434 which is Ollama's standard port.
OLLAMA_BASE_URL=http://localhost:11434
# The local Llama 4 model to use with Ollama.
# Run 'ollama list' to see which models are available locally.
OLLAMA_MODEL=llama4:maverick
The README.md file provides the quick-start guide that every new contributor needs:
# Weather MCP Server
A production-ready MCP server (spec 2026-07-28) that wraps the
OpenWeatherMap REST API and exposes weather data to any MCP-compatible
LLM client.
## Prerequisites
- Python 3.12 or higher
- uv package manager (https://docs.astral.sh/uv/)
- An OpenWeatherMap API key (free at https://openweathermap.org/api)
- Ollama (optional, for local LLM support): https://ollama.com
## Quick Start
# 1. Clone and enter the project
git clone <repo-url>
cd weather-mcp-server
# 2. Install dependencies
uv sync
# 3. Configure environment
cp .env.example .env
# Edit .env and add your OWM_API_KEY
# 4. Start the MCP server
uv run weather-mcp
# 5. Verify it is running
curl http://localhost:8000/health
## Running Tests
uv run pytest tests/ -v
## Running the Cloud LLM Client
# Ensure the server is running first, then:
uv run python clients/cloud_llm_client.py
## Running the Local LLM Client (requires Ollama)
# Terminal 1: Start Ollama
ollama serve
# Terminal 2: Pull the model (first time only)
ollama pull llama4:maverick
# Terminal 3: Start the MCP server
uv run weather-mcp
# Terminal 4: Run the local client
uv run python clients/local_llm_client.py
## MCP Endpoint
POST http://localhost:8000/mcp
Headers: Mcp-Method: tools/list | tools/call | resources/read | prompts/get
## Available Tools
- get_current_weather - Current conditions by city name
- get_weather_by_coordinates - Current conditions by lat/lon
- get_weather_forecast - Multi-step forecast by city name
- get_comprehensive_weather_report - Combined current + 48h forecast
## Available Resources
- weather://current/{city} - Current weather as JSON document
- weather://forecast/{city} - 48-hour forecast as JSON document
## Available Prompts
- clothing_recommendation - Guide LLM to recommend clothing
- travel_weather_briefing - Guide LLM to compare two cities
CHAPTER 6: THE DOMAIN LAYER - PURE PYTHON, ZERO DEPENDENCIES
The domain layer is the soul of our application. It contains the data structures that represent weather concepts in our system, and it contains the exceptions that describe everything that can go wrong. Nothing in this layer imports from httpx, from the MCP SDK, or from any external library except Pydantic, which we use purely for data validation and serialization. This purity is what makes the domain layer so valuable: it can be tested in complete isolation, it can be understood without knowing anything about external systems, and it never needs to change because of infrastructure concerns.
Let us start with the configuration module, which technically sits outside the domain but needs to be established first because everything else depends on it. Notice that we import only what we actually use: Field from pydantic and BaseSettings from pydantic-settings. Unused imports are removed entirely.
# src/weather_mcp/config.py
#
# Configuration management using Pydantic Settings.
#
# Pydantic Settings automatically reads values from environment variables
# and .env files, providing type validation and clear error messages
# when required configuration is missing. Every field is type-annotated,
# which means Pydantic validates each value at startup rather than
# failing silently at the moment the value is first used.
from __future__ import annotations
from functools import lru_cache
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""
Application-wide settings loaded from environment variables.
All fields have type annotations. If OWM_API_KEY is missing from
the environment, the application will raise a clear ValidationError
at startup rather than failing with a cryptic error later.
"""
model_config = SettingsConfigDict(
# Read from a .env file in the current working directory.
# The env_file is optional; if it does not exist, settings are
# read purely from the process environment.
env_file=".env",
env_file_encoding="utf-8",
# Allow extra fields in the .env file without raising an error.
# This is useful when the .env file contains variables for other
# tools in the same repository.
extra="ignore",
)
# OpenWeatherMap API configuration
owm_api_key: str = Field(
..., # Ellipsis means this field is required; no default value.
description=(
"OpenWeatherMap API key. Required. "
"Get one for free at https://openweathermap.org/api."
),
)
owm_base_url: str = Field(
default="https://api.openweathermap.org",
description="Base URL for the OpenWeatherMap API.",
)
owm_default_units: str = Field(
default="metric",
description="Default unit system: metric, imperial, or standard.",
)
# MCP server network configuration
mcp_host: str = Field(
default="0.0.0.0",
description="Host address for the MCP server to bind to.",
)
mcp_port: int = Field(
default=8000,
description="TCP port for the MCP server.",
)
# The full URL of the MCP endpoint, used by client scripts.
mcp_server_url: str = Field(
default="http://localhost:8000/mcp",
description="Full URL of the MCP server endpoint, used by clients.",
)
# Cloud LLM provider API keys (optional; only needed for cloud clients)
openai_api_key: str = Field(
default="",
description="OpenAI API key for GPT-6 Astra access.",
)
anthropic_api_key: str = Field(
default="",
description="Anthropic API key for Claude Fable 5.1 access.",
)
google_api_key: str = Field(
default="",
description="Google API key for Gemini 3.8 Flash access.",
)
# Local LLM configuration via Ollama
ollama_base_url: str = Field(
default="http://localhost:11434",
description="Base URL of the local Ollama inference server.",
)
ollama_model: str = Field(
default="llama4:maverick",
description="The Ollama model to use for local inference.",
)
@lru_cache(maxsize=1)
def get_settings() -> Settings:
"""
Return the application settings, loading them on first call.
The @lru_cache decorator ensures that the settings are only loaded
once per process, regardless of how many times this function is called.
This is both a performance optimization and a correctness guarantee:
all parts of the application see the same configuration values.
To override settings in tests, use:
from weather_mcp.config import get_settings
get_settings.cache_clear()
monkeypatch.setenv("OWM_API_KEY", "test_key")
"""
return Settings()
Now let us build the domain models. These are the Pydantic data classes that represent weather concepts in our system. Every piece of data that flows through our application will be validated against one of these models. We import only what we actually use: Optional from typing is kept because it is used in field type annotations, and field_validator is removed because it was imported but never applied anywhere in the original.
# src/weather_mcp/domain/models.py
#
# Domain models for the Weather MCP Server.
#
# These are pure data containers with validation. They know nothing about
# HTTP, MCP, or OpenWeatherMap's specific JSON format. The adapter layer
# is responsible for mapping OpenWeatherMap's raw JSON into these models.
# This separation means we can change the underlying API without touching
# any business logic.
from __future__ import annotations
from datetime import datetime
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field
class UnitSystem(str, Enum):
"""
The unit system to use for temperature and wind speed measurements.
Using an Enum instead of raw strings prevents typos and makes the
valid values self-documenting across the entire codebase.
"""
METRIC = "metric" # Temperature in Celsius, wind in m/s
IMPERIAL = "imperial" # Temperature in Fahrenheit, wind in mph
STANDARD = "standard" # Temperature in Kelvin, wind in m/s
class WindInfo(BaseModel):
"""Represents wind conditions at a location."""
speed: float = Field(
description="Wind speed in the units of the current unit system.",
)
direction_degrees: int = Field(
ge=0,
le=360,
description="Wind direction in meteorological degrees (0-360).",
)
gust: Optional[float] = Field(
default=None,
description="Wind gust speed, if reported by the station.",
)
@property
def direction_label(self) -> str:
"""
Convert degrees to a human-readable compass direction label.
This is a computed property derived from direction_degrees.
The 16-point compass divides 360 degrees into 16 sectors of
22.5 degrees each. Adding half a sector width (11.25 degrees)
before dividing centers each label on its sector correctly.
Examples:
0 degrees -> "N"
90 degrees -> "E"
180 degrees -> "S"
225 degrees -> "SW"
359 degrees -> "N"
"""
directions = [
"N", "NNE", "NE", "ENE", "E", "ESE", "SE", "SSE",
"S", "SSW", "SW", "WSW", "W", "WNW", "NW", "NNW",
]
index = int((self.direction_degrees + 11.25) / 22.5) % 16
return directions[index]
class WeatherCondition(BaseModel):
"""Describes the sky and atmospheric conditions at a location."""
main: str = Field(
description=(
"Short category label for the condition, e.g., 'Rain', 'Clear'."
),
)
description: str = Field(
description=(
"Longer human-readable description, e.g., 'light rain'."
),
)
icon_code: str = Field(
description="OpenWeatherMap icon code for visual representation.",
)
class CurrentWeather(BaseModel):
"""
A complete snapshot of current weather conditions at a location.
This is the primary domain object for current weather data.
"""
city_name: str = Field(
description="The name of the city or location.",
)
country_code: str = Field(
description="ISO 3166-1 alpha-2 country code, e.g., 'GB', 'DE'.",
)
latitude: float = Field(
description="Geographic latitude of the location.",
)
longitude: float = Field(
description="Geographic longitude of the location.",
)
temperature: float = Field(
description="Current temperature in the current unit system.",
)
feels_like: float = Field(
description="Apparent (feels-like) temperature.",
)
temp_min: float = Field(
description="Minimum temperature currently observed in the area.",
)
temp_max: float = Field(
description="Maximum temperature currently observed in the area.",
)
humidity: int = Field(
ge=0,
le=100,
description="Relative humidity as a percentage (0-100).",
)
pressure: int = Field(
description="Atmospheric pressure in hPa at sea level.",
)
visibility_meters: Optional[int] = Field(
default=None,
description="Visibility in meters. Maximum reported value is 10,000 m.",
)
condition: WeatherCondition = Field(
description="The primary weather condition at this location.",
)
wind: WindInfo = Field(
description="Current wind conditions.",
)
unit_system: UnitSystem = Field(
description="The unit system used for all measurements in this object.",
)
observed_at: datetime = Field(
description="The UTC timestamp when this observation was recorded.",
)
@property
def temperature_unit(self) -> str:
"""Return the temperature unit symbol for the current unit system."""
mapping = {
UnitSystem.METRIC: "C",
UnitSystem.IMPERIAL: "F",
UnitSystem.STANDARD: "K",
}
return mapping[self.unit_system]
@property
def wind_unit(self) -> str:
"""Return the wind speed unit label for the current unit system."""
if self.unit_system == UnitSystem.IMPERIAL:
return "mph"
return "m/s"
def to_summary_string(self) -> str:
"""
Produce a concise, human-readable summary of current conditions.
This string is included in MCP tool responses for the LLM to read.
It is designed to be information-dense but still natural to parse.
"""
return (
f"{self.city_name}, {self.country_code}: "
f"{self.temperature:.1f}°{self.temperature_unit} "
f"({self.condition.description}), "
f"feels like {self.feels_like:.1f}°{self.temperature_unit}, "
f"humidity {self.humidity}%, "
f"wind {self.wind.speed:.1f} {self.wind_unit} "
f"from {self.wind.direction_label}."
)
class ForecastEntry(BaseModel):
"""A single time-step in a weather forecast."""
forecast_time: datetime = Field(
description="The UTC time for which this forecast is valid.",
)
temperature: float = Field(
description="Forecast temperature in the current unit system.",
)
feels_like: float = Field(
description="Forecast apparent temperature.",
)
condition: WeatherCondition = Field(
description="Forecast sky conditions.",
)
wind: WindInfo = Field(
description="Forecast wind conditions.",
)
precipitation_probability: float = Field(
ge=0.0,
le=1.0,
description=(
"Probability of precipitation as a fraction from 0.0 to 1.0. "
"Multiply by 100 to get a percentage."
),
)
rain_mm: Optional[float] = Field(
default=None,
description=(
"Expected rainfall in millimeters over the 3-hour forecast period."
),
)
class WeatherForecast(BaseModel):
"""A multi-step weather forecast for a location."""
city_name: str = Field(
description="The name of the city or location.",
)
country_code: str = Field(
description="ISO 3166-1 alpha-2 country code.",
)
unit_system: UnitSystem = Field(
description="The unit system used for all measurements.",
)
entries: list[ForecastEntry] = Field(
description=(
"The individual forecast time steps, ordered chronologically. "
"Each step covers a 3-hour window."
),
)
@property
def temperature_unit(self) -> str:
"""Return the temperature unit symbol for the current unit system."""
mapping = {
UnitSystem.METRIC: "C",
UnitSystem.IMPERIAL: "F",
UnitSystem.STANDARD: "K",
}
return mapping[self.unit_system]
Now we define our exceptions. Good exception design is something many tutorials skip, but it is genuinely important. Well-named exceptions make error handling readable and allow the MCP tools layer to give the LLM meaningful, actionable error messages rather than cryptic tracebacks.
# src/weather_mcp/domain/exceptions.py
#
# Domain-specific exceptions for the Weather MCP Server.
#
# These exceptions form a hierarchy rooted at WeatherServiceError.
# Catching the base exception catches everything; catching a specific
# subclass allows fine-grained error handling. This hierarchy mirrors
# the distinct kinds of things that can go wrong when fetching weather data.
from __future__ import annotations
class WeatherServiceError(Exception):
"""
Base exception for all errors originating from the weather service.
All other exceptions in this module inherit from this class, which
makes it easy to write a single except clause that catches all
weather-related errors when fine-grained handling is not needed.
"""
def __init__(self, message: str, details: str = "") -> None:
super().__init__(message)
self.message = message
self.details = details
def __str__(self) -> str:
if self.details:
return f"{self.message} Details: {self.details}"
return self.message
class LocationNotFoundError(WeatherServiceError):
"""
Raised when the requested city or coordinates cannot be found.
This typically means the user provided a city name that does not
exist in the OpenWeatherMap database, or geographic coordinates
that fall outside any recognized location. The fix is usually
to add a country code to the city name, e.g., 'Springfield,US'.
"""
pass
class ApiKeyError(WeatherServiceError):
"""
Raised when the API key is missing, invalid, or has been revoked.
This is a configuration error that requires human intervention to fix.
The server cannot recover from this automatically.
"""
pass
class RateLimitError(WeatherServiceError):
"""
Raised when the OpenWeatherMap API rate limit has been exceeded.
The free tier allows 60 calls per minute. When this exception is
raised, the caller should wait before retrying.
"""
def __init__(
self,
message: str,
retry_after_seconds: int = 60,
) -> None:
super().__init__(message)
self.retry_after_seconds = retry_after_seconds
class ApiCommunicationError(WeatherServiceError):
"""
Raised when the HTTP request to the OpenWeatherMap API fails for
any network-level reason: timeouts, DNS failures, TLS errors, and
similar transient problems. This error may resolve on retry.
"""
pass
class InvalidInputError(WeatherServiceError):
"""
Raised when the input provided is semantically invalid, even if it
passed schema validation. For example, a unit system value that is
not one of 'metric', 'imperial', or 'standard'.
"""
pass
CHAPTER 7: THE HTTP ADAPTER - TALKING TO OPENWEATHERMAP
The adapter layer is where we write the code that actually makes HTTP calls to the OpenWeatherMap API. This layer has one job and one job only: take parameters, make an HTTP request, parse the response, and return a domain model. It knows about HTTP status codes, about OpenWeatherMap's specific JSON structure, and about the error codes that the API returns. Nothing outside this layer needs to know any of those things.
We use httpx as our HTTP client. httpx is the modern, async-first HTTP client for Python. It has an interface very similar to the beloved requests library but supports async/await natively, which is essential for a high-performance server that needs to handle many concurrent requests.
We also build a token bucket rate limiter to protect ourselves from hitting OpenWeatherMap's API limits. The original implementation contained a critical concurrency bug: the asyncio lock was held during the sleep call, which would have blocked every other coroutine from even checking the rate limiter. The corrected implementation uses a while loop that releases the lock before sleeping and re-acquires it at the start of each iteration.
# src/weather_mcp/adapters/rate_limiter.py
#
# An async-safe token bucket rate limiter.
#
# The token bucket algorithm: imagine a bucket that holds up to N tokens.
# Tokens are added at a fixed rate (e.g., 55 tokens per 60 seconds).
# Each API call consumes one token. If the bucket is empty, the caller
# waits until a token becomes available.
#
# CONCURRENCY CORRECTNESS: The asyncio.Lock is acquired only for the brief
# moment of checking and updating the token count. It is NEVER held during
# the asyncio.sleep() call. This is critical: holding the lock during sleep
# would block every other coroutine from accessing the rate limiter at all,
# defeating its purpose entirely.
from __future__ import annotations
import asyncio
import time
from dataclasses import dataclass, field
@dataclass
class TokenBucketRateLimiter:
"""
An async-safe token bucket rate limiter.
Parameters
----------
max_calls : int
The maximum number of calls allowed per period. This is the
bucket capacity: the bucket never holds more than this many tokens.
period_seconds : float
The time period in seconds over which max_calls is measured.
For example, max_calls=55 and period_seconds=60.0 means at most
55 calls per minute (with a 5-call safety margin below OWM's limit).
"""
max_calls: int
period_seconds: float
_tokens: float = field(init=False)
_last_refill_time: float = field(init=False)
_lock: asyncio.Lock = field(init=False)
def __post_init__(self) -> None:
# Start with a full bucket so the first burst of requests goes
# through immediately without waiting.
self._tokens = float(self.max_calls)
self._last_refill_time = time.monotonic()
# asyncio.Lock in Python 3.10+ does not require a running event loop
# at construction time, so creating it here in __post_init__ is safe.
self._lock = asyncio.Lock()
async def acquire(self) -> None:
"""
Acquire one token, waiting if necessary until one is available.
This coroutine returns immediately if a token is available.
If the bucket is empty, it calculates the wait time, releases
the lock, sleeps for that duration, and then tries again.
The while loop handles the case where another coroutine consumed
the token during the sleep window.
"""
while True:
async with self._lock:
self._refill()
if self._tokens >= 1.0:
# A token is available. Consume it and return.
self._tokens -= 1.0
return
# Bucket is empty. Calculate how long to wait for one token.
# token_rate is the number of tokens generated per second.
token_rate = self.max_calls / self.period_seconds
wait_time = (1.0 - self._tokens) / token_rate
# The lock is released here, BEFORE we sleep.
# This allows other coroutines to check the bucket state
# while we are waiting. After sleeping, we loop back and
# try to acquire a token again under a fresh lock acquisition.
await asyncio.sleep(wait_time)
def _refill(self) -> None:
"""
Add tokens to the bucket based on elapsed time since last refill.
Called internally at the start of every lock-protected section.
The bucket capacity is capped at max_calls to prevent unbounded growth.
"""
now = time.monotonic()
elapsed = now - self._last_refill_time
self._last_refill_time = now
token_rate = self.max_calls / self.period_seconds
self._tokens = min(
float(self.max_calls),
self._tokens + elapsed * token_rate,
)
With the rate limiter in place, we can now build the OpenWeatherMap HTTP client. The most important structural improvement over the naive approach is the extraction of the shared JSON-to-domain mapping logic into a single private method. Both get_current_weather and get_current_weather_by_coords return a CurrentWeather object, and both need to perform exactly the same field mapping. Without the extraction, a change to the CurrentWeather model would require updating two identical blocks of code — a classic DRY violation that causes bugs when one copy is updated and the other is forgotten.
# src/weather_mcp/adapters/owm_client.py
#
# HTTP adapter for the OpenWeatherMap REST API.
#
# This module is the only place in the entire codebase that knows about
# OpenWeatherMap's specific URL structure, JSON field names, and error codes.
# It translates between the raw HTTP world and our clean domain models.
# All other layers interact only with domain objects; they never see raw
# HTTP responses or OpenWeatherMap-specific data structures.
from __future__ import annotations
from datetime import datetime, timezone
from typing import Any
import httpx
from weather_mcp.adapters.rate_limiter import TokenBucketRateLimiter
from weather_mcp.config import get_settings
from weather_mcp.domain.exceptions import (
ApiCommunicationError,
ApiKeyError,
LocationNotFoundError,
RateLimitError,
)
from weather_mcp.domain.models import (
CurrentWeather,
ForecastEntry,
UnitSystem,
WeatherCondition,
WeatherForecast,
WindInfo,
)
class OpenWeatherMapClient:
"""
Async HTTP client for the OpenWeatherMap REST API.
This class encapsulates all HTTP communication with OpenWeatherMap.
It handles authentication, rate limiting, error translation, and
response mapping. Callers receive clean domain objects; they never
see raw HTTP responses or OpenWeatherMap-specific data structures.
Usage
-----
Use this class as an async context manager to ensure that the
underlying HTTP connection pool is properly opened and closed:
async with OpenWeatherMapClient() as client:
weather = await client.get_current_weather("London")
"""
# API endpoint paths. Storing these as class constants makes them
# easy to locate and update if the API changes its URL structure.
_CURRENT_WEATHER_PATH = "/data/2.5/weather"
_FORECAST_PATH = "/data/2.5/forecast"
# HTTP status codes that have specific meanings in the OWM API.
_HTTP_UNAUTHORIZED = 401
_HTTP_NOT_FOUND = 404
_HTTP_TOO_MANY_REQUESTS = 429
# OWM sometimes returns HTTP 200 with this error code in the body
# when the city name is not recognized.
_OWM_CITY_NOT_FOUND_CODE = "404"
def __init__(self) -> None:
settings = get_settings()
self._api_key = settings.owm_api_key
self._base_url = settings.owm_base_url
self._default_units = settings.owm_default_units
# Configure for the free tier: 60 calls/minute.
# We use 55 as the limit to provide a 5-call safety margin,
# protecting against clock skew and burst timing issues.
self._rate_limiter = TokenBucketRateLimiter(
max_calls=55,
period_seconds=60.0,
)
# The httpx.AsyncClient is created in open() and closed in close().
self._http_client: httpx.AsyncClient | None = None
async def open(self) -> None:
"""Open the underlying HTTP connection pool."""
self._http_client = httpx.AsyncClient(
base_url=self._base_url,
timeout=httpx.Timeout(
connect=5.0, # Seconds to establish a TCP connection.
read=10.0, # Seconds to receive the full response body.
write=5.0, # Seconds to send the request body.
pool=2.0, # Seconds to acquire a connection from the pool.
),
headers={
"User-Agent": "WeatherMCPServer/1.0",
"Accept": "application/json",
},
)
async def close(self) -> None:
"""Close the underlying HTTP connection pool."""
if self._http_client is not None:
await self._http_client.aclose()
self._http_client = None
async def __aenter__(self) -> "OpenWeatherMapClient":
await self.open()
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: object,
) -> None:
await self.close()
def _ensure_open(self) -> httpx.AsyncClient:
"""
Return the HTTP client, raising a clear error if not yet opened.
This defensive check prevents confusing AttributeError messages
if someone forgets to use the client as a context manager.
"""
if self._http_client is None:
raise RuntimeError(
"OpenWeatherMapClient must be used as an async context manager "
"or opened explicitly with 'await client.open()'."
)
return self._http_client
def _build_common_params(self, units: str) -> dict[str, str]:
"""
Build the query parameters common to all API requests.
Every request to the OWM API needs the API key and unit system.
"""
return {
"appid": self._api_key,
"units": units,
}
async def _make_request(
self,
path: str,
params: dict[str, str],
) -> dict[str, Any]:
"""
Make a rate-limited GET request to the OpenWeatherMap API.
This private method is the single point through which all HTTP
calls flow. It handles rate limiting, HTTP errors, and network
errors in one place, so individual endpoint methods stay clean.
Parameters
----------
path : str
The API endpoint path, e.g., '/data/2.5/weather'.
params : dict
Query parameters to include in the request.
Returns
-------
dict
The parsed JSON response body.
Raises
------
ApiKeyError
If the API key is invalid or unauthorized.
LocationNotFoundError
If the requested location does not exist.
RateLimitError
If the API rate limit has been exceeded.
ApiCommunicationError
If any network-level error occurs.
"""
# Wait for a rate limiter token before proceeding.
await self._rate_limiter.acquire()
http = self._ensure_open()
try:
response = await http.get(path, params=params)
except httpx.TimeoutException as exc:
raise ApiCommunicationError(
"Request to OpenWeatherMap timed out.",
details=str(exc),
) from exc
except httpx.NetworkError as exc:
raise ApiCommunicationError(
"Network error while contacting OpenWeatherMap.",
details=str(exc),
) from exc
# Translate HTTP error codes into domain exceptions.
if response.status_code == self._HTTP_UNAUTHORIZED:
raise ApiKeyError(
"The OpenWeatherMap API key is invalid or unauthorized.",
details="Check that OWM_API_KEY in your .env file is correct.",
)
if response.status_code == self._HTTP_TOO_MANY_REQUESTS:
raise RateLimitError(
"OpenWeatherMap API rate limit exceeded.",
retry_after_seconds=60,
)
if response.status_code == self._HTTP_NOT_FOUND:
raise LocationNotFoundError(
"The requested location was not found.",
details=f"API returned 404 for path: {path}",
)
if response.status_code >= 400:
raise ApiCommunicationError(
"OpenWeatherMap returned an unexpected error.",
details=f"HTTP {response.status_code}: {response.text[:200]}",
)
data = response.json()
# OWM sometimes returns HTTP 200 with an error code in the body.
# This non-standard behavior must be handled explicitly.
if (
isinstance(data, dict)
and data.get("cod") == self._OWM_CITY_NOT_FOUND_CODE
):
raise LocationNotFoundError(
f"Location not found: {data.get('message', 'Unknown error')}",
)
return data
def _map_wind(self, wind_data: dict[str, Any]) -> WindInfo:
"""Map the 'wind' object from OWM JSON to a WindInfo domain model."""
return WindInfo(
speed=wind_data.get("speed", 0.0),
direction_degrees=wind_data.get("deg", 0),
gust=wind_data.get("gust"),
)
def _map_condition(self, weather_list: list[dict]) -> WeatherCondition:
"""
Map the 'weather' array from OWM JSON to a WeatherCondition model.
OpenWeatherMap returns a list of conditions; we use the first one,
which is always the primary condition for the location.
"""
if not weather_list:
return WeatherCondition(
main="Unknown",
description="No condition data available",
icon_code="",
)
primary = weather_list[0]
return WeatherCondition(
main=primary.get("main", "Unknown"),
description=primary.get("description", ""),
icon_code=primary.get("icon", ""),
)
def _map_current_weather(
self,
data: dict[str, Any],
units: str,
) -> CurrentWeather:
"""
Map a raw OWM current-weather JSON response to a CurrentWeather model.
This private helper is called by both get_current_weather() and
get_current_weather_by_coords(), eliminating the code duplication
that would otherwise exist between those two methods. Any change
to the CurrentWeather model only needs to be reflected here.
"""
return CurrentWeather(
city_name=data["name"],
country_code=data["sys"]["country"],
latitude=data["coord"]["lat"],
longitude=data["coord"]["lon"],
temperature=data["main"]["temp"],
feels_like=data["main"]["feels_like"],
temp_min=data["main"]["temp_min"],
temp_max=data["main"]["temp_max"],
humidity=data["main"]["humidity"],
pressure=data["main"]["pressure"],
visibility_meters=data.get("visibility"),
condition=self._map_condition(data.get("weather", [])),
wind=self._map_wind(data.get("wind", {})),
unit_system=UnitSystem(units),
observed_at=datetime.fromtimestamp(
data["dt"], tz=timezone.utc
),
)
async def get_current_weather(
self,
city: str,
units: str | None = None,
) -> CurrentWeather:
"""
Fetch current weather conditions for a named city.
Parameters
----------
city : str
The city name, optionally with country code: "London,GB".
Including the country code reduces ambiguity for common names.
units : str, optional
The unit system to use. Defaults to the configured default.
Returns
-------
CurrentWeather
A domain model containing the current weather conditions.
"""
effective_units = units or self._default_units
params = self._build_common_params(effective_units)
params["q"] = city
data = await self._make_request(self._CURRENT_WEATHER_PATH, params)
return self._map_current_weather(data, effective_units)
async def get_current_weather_by_coords(
self,
latitude: float,
longitude: float,
units: str | None = None,
) -> CurrentWeather:
"""
Fetch current weather for a specific geographic coordinate pair.
This is useful when the user provides coordinates rather than a
city name, or when city name lookup is ambiguous.
"""
effective_units = units or self._default_units
params = self._build_common_params(effective_units)
params["lat"] = str(latitude)
params["lon"] = str(longitude)
data = await self._make_request(self._CURRENT_WEATHER_PATH, params)
return self._map_current_weather(data, effective_units)
async def get_forecast(
self,
city: str,
units: str | None = None,
max_entries: int = 16,
) -> WeatherForecast:
"""
Fetch a multi-step weather forecast for a named city.
The /forecast endpoint returns up to 40 entries at 3-hour intervals
(5 days total). We limit the response to max_entries to keep LLM
context usage manageable.
Parameters
----------
city : str
The city name, optionally with country code.
units : str, optional
The unit system to use.
max_entries : int
Maximum number of 3-hour forecast steps to return.
Default is 16, which covers approximately 48 hours.
"""
effective_units = units or self._default_units
params = self._build_common_params(effective_units)
params["q"] = city
params["cnt"] = str(max_entries)
data = await self._make_request(self._FORECAST_PATH, params)
entries = [
ForecastEntry(
forecast_time=datetime.fromtimestamp(
entry["dt"], tz=timezone.utc
),
temperature=entry["main"]["temp"],
feels_like=entry["main"]["feels_like"],
condition=self._map_condition(entry.get("weather", [])),
wind=self._map_wind(entry.get("wind", {})),
precipitation_probability=entry.get("pop", 0.0),
rain_mm=entry.get("rain", {}).get("3h"),
)
for entry in data.get("list", [])
]
return WeatherForecast(
city_name=data["city"]["name"],
country_code=data["city"]["country"],
unit_system=UnitSystem(effective_units),
entries=entries,
)
CHAPTER 8: THE SERVICE LAYER - BUSINESS LOGIC LIVES HERE
The service layer sits between the MCP tools and the HTTP adapter. Its job is to implement business logic: combining data from multiple API calls, applying application-level rules, and preparing data in the format that the MCP tools need to present to the LLM.
Two important corrections from the original apply here. First, all imports — including asyncio and math — are placed at the top of the file where they belong, not buried inside method bodies or appended after the class definition. Second, get_comprehensive_report now calls self.get_forecast() rather than reaching directly into self._client.get_forecast(), which preserves the layered architecture and ensures that the service layer's own logic (the hours-to-entries conversion) is always applied consistently.
# src/weather_mcp/service/weather_service.py
#
# The weather service: business logic and orchestration.
#
# This layer orchestrates calls to the HTTP adapter and applies
# application-level logic. It is the only layer that knows about
# both domain models AND the adapter layer. The MCP tools layer
# only knows about this service; it never calls the adapter directly.
from __future__ import annotations
import asyncio
import math
from typing import Any
from weather_mcp.adapters.owm_client import OpenWeatherMapClient
from weather_mcp.domain.models import (
CurrentWeather,
UnitSystem,
WeatherForecast,
)
class WeatherService:
"""
Business logic layer for weather data retrieval and formatting.
This class owns the lifecycle of the HTTP adapter and provides
a clean, high-level interface to the MCP tools layer. It handles
the creation and cleanup of the HTTP client, ensuring that
connection pools are properly managed across the server's lifetime.
"""
def __init__(self) -> None:
# The client is created here but not opened yet.
# Opening the client (and its connection pool) happens in
# the async context manager, which is called during server startup.
self._client = OpenWeatherMapClient()
async def open(self) -> None:
"""Open the HTTP client connection pool."""
await self._client.open()
async def close(self) -> None:
"""Close the HTTP client connection pool."""
await self._client.close()
async def __aenter__(self) -> "WeatherService":
await self.open()
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: object,
) -> None:
await self.close()
async def get_current_weather(
self,
city: str,
units: str = "metric",
) -> CurrentWeather:
"""
Retrieve current weather conditions for a city.
This is a thin pass-through to the adapter for current weather,
as no additional business logic is required for this operation.
"""
return await self._client.get_current_weather(
city=city,
units=units,
)
async def get_current_weather_by_coords(
self,
latitude: float,
longitude: float,
units: str = "metric",
) -> CurrentWeather:
"""Retrieve current weather for geographic coordinates."""
return await self._client.get_current_weather_by_coords(
latitude=latitude,
longitude=longitude,
units=units,
)
async def get_forecast(
self,
city: str,
units: str = "metric",
hours_ahead: int = 48,
) -> WeatherForecast:
"""
Retrieve a weather forecast for a city.
Parameters
----------
city : str
The city name, optionally with country code.
units : str
The unit system: 'metric', 'imperial', or 'standard'.
hours_ahead : int
How many hours of forecast to return. The OWM API returns
data in 3-hour steps, so we request ceil(hours_ahead / 3)
entries to cover the full requested period.
"""
# Convert hours to the number of 3-hour forecast entries needed.
# We add 1 to ensure we cover the full requested period even when
# hours_ahead is not evenly divisible by 3.
max_entries = math.ceil(hours_ahead / 3) + 1
return await self._client.get_forecast(
city=city,
units=units,
max_entries=max_entries,
)
async def get_comprehensive_report(
self,
city: str,
units: str = "metric",
) -> dict[str, Any]:
"""
Produce a comprehensive weather report combining current conditions
and a 48-hour forecast into a single structured dictionary.
This method demonstrates genuine business logic in the service layer:
the decision to combine two API calls and format the result in a
specific way is a business rule, not a protocol concern or an
HTTP concern. It belongs here, not in the tools layer.
Both API calls are made concurrently using asyncio.gather() to
minimize total latency. The calls are independent, so there is
no reason to wait for one before starting the other.
Returns
-------
dict
A dictionary with structured current and forecast data,
ready for serialization into an MCP tool response.
"""
# Call self.get_forecast() (the service method), not
# self._client.get_forecast() (the adapter method directly).
# This ensures the hours_ahead -> max_entries conversion logic
# in self.get_forecast() is always applied consistently.
current, forecast = await asyncio.gather(
self.get_current_weather(city=city, units=units),
self.get_forecast(city=city, units=units, hours_ahead=48),
)
temp_unit = current.temperature_unit
wind_unit = current.wind_unit
# Build the forecast summary as a list of concise strings.
forecast_lines = []
for entry in forecast.entries:
time_str = entry.forecast_time.strftime("%a %d %b %H:%M UTC")
pop_pct = int(entry.precipitation_probability * 100)
line = (
f" {time_str}: {entry.temperature:.1f}°{temp_unit}, "
f"{entry.condition.description}, "
f"rain chance {pop_pct}%"
)
if entry.rain_mm is not None:
line += f", {entry.rain_mm:.1f}mm rain"
forecast_lines.append(line)
return {
"location": f"{current.city_name}, {current.country_code}",
"current_conditions": current.to_summary_string(),
"current_details": {
"temperature": (
f"{current.temperature:.1f}°{temp_unit}"
),
"feels_like": (
f"{current.feels_like:.1f}°{temp_unit}"
),
"humidity": f"{current.humidity}%",
"pressure": f"{current.pressure} hPa",
"visibility": (
f"{current.visibility_meters}m"
if current.visibility_meters is not None
else "Not available"
),
"wind": (
f"{current.wind.speed:.1f} {wind_unit} "
f"from {current.wind.direction_label}"
),
},
"48_hour_forecast": forecast_lines,
"unit_system": units,
}
CHAPTER 9: THE MCP TOOLS LAYER - WHERE THE MAGIC HAPPENS
We have arrived at the most exciting part of the entire tutorial: the MCP tools layer. This is where we take everything we have built so far and expose it to the world of LLMs through the Model Context Protocol. Every function we decorate with @mcp.tool() becomes a capability that any MCP-compatible LLM can discover and invoke.
The MCP Python SDK version 2.0.0 uses a beautifully ergonomic approach to tool definition. You write a normal Python async function with type-annotated parameters and a docstring. The SDK reads the type annotations to generate the JSON Schema that describes the tool's inputs, and it reads the docstring to generate the description that the LLM uses to decide when to call the tool. This means that writing good, clear docstrings is not just good practice: it directly affects how well the LLM understands and uses your tools.
Two important corrections from the original are applied here. First, the gust check is fixed from "if weather.wind.gust" (which would incorrectly evaluate to False when gust is 0.0) to "if weather.wind.gust is not None", which correctly handles the zero-gust case. Second, the unused import of InvalidInputError is removed, keeping the import block clean.
# src/weather_mcp/tools/weather_tools.py
#
# MCP tool definitions for the Weather MCP Server.
#
# Each function decorated with @mcp.tool() becomes a tool that LLMs
# can discover (via tools/list) and invoke (via tools/call).
#
# DOCSTRING QUALITY MATTERS: The docstring of each tool function is the
# description that the LLM sees when deciding which tool to use. Write
# docstrings as if explaining the tool to a colleague who needs to decide
# when to call it. Be specific about what it does, what parameters it
# expects, and what it returns. Vague docstrings lead to incorrect usage.
from __future__ import annotations
from typing import Annotated
from mcp.server.fastmcp import FastMCP
from pydantic import Field
from weather_mcp.domain.exceptions import (
ApiCommunicationError,
ApiKeyError,
LocationNotFoundError,
RateLimitError,
WeatherServiceError,
)
from weather_mcp.service.weather_service import WeatherService
def register_weather_tools(mcp: FastMCP, service: WeatherService) -> None:
"""
Register all weather-related MCP tools with the server.
This function is called once during server startup. It defines all
tool functions as closures over the 'service' parameter, giving each
tool access to the WeatherService without relying on global state.
Using closures instead of globals makes the tools testable in isolation:
you can pass a mock service to this function and the tools will use it.
Parameters
----------
mcp : FastMCP
The FastMCP server instance to register tools with.
service : WeatherService
The weather service instance that tools will use to fetch data.
"""
def _format_error(exc: WeatherServiceError) -> str:
"""
Format a domain exception into a clear, actionable error message
for the LLM. The LLM will include this message in its response
to the user, so it should be human-readable and genuinely helpful.
"""
if isinstance(exc, LocationNotFoundError):
return (
f"Location not found: {exc.message} "
"Please check the city name spelling and try again. "
"Adding a country code improves accuracy, "
"e.g., 'London,GB' instead of just 'London'."
)
if isinstance(exc, ApiKeyError):
return (
"The weather service is not properly configured. "
"The API key is missing or invalid. "
"Please contact the system administrator."
)
if isinstance(exc, RateLimitError):
return (
"The weather service is temporarily rate-limited. "
f"Please wait {exc.retry_after_seconds} seconds and try again."
)
if isinstance(exc, ApiCommunicationError):
return (
"Unable to reach the weather service due to a network error. "
"Please check your internet connection and try again."
)
return f"An unexpected error occurred: {exc.message}"
# ----------------------------------------------------------------
# TOOL 1: Get Current Weather by City Name
# ----------------------------------------------------------------
@mcp.tool()
async def get_current_weather(
city: Annotated[
str,
Field(
description=(
"The city name to get weather for. For best results, "
"include the country code: 'London,GB', 'Paris,FR', "
"'New York,US'. City names are case-insensitive."
),
),
],
units: Annotated[
str,
Field(
description=(
"The unit system for measurements. "
"Use 'metric' for Celsius and m/s (default), "
"'imperial' for Fahrenheit and mph, "
"or 'standard' for Kelvin and m/s."
),
),
] = "metric",
) -> str:
"""
Get the current weather conditions for a specific city.
Use this tool when the user asks about the current weather,
temperature, conditions, or climate in a specific city right now.
This tool returns real-time data including temperature, humidity,
wind speed and direction, atmospheric pressure, and sky conditions.
Do NOT use this tool for forecasts or historical data; use the
appropriate forecast tool for future weather questions.
"""
try:
weather = await service.get_current_weather(
city=city,
units=units,
)
temp_unit = weather.temperature_unit
wind_unit = weather.wind_unit
obs_time = weather.observed_at.strftime("%Y-%m-%d %H:%M UTC")
# Build the gust string only when gust data is actually present.
# The check uses 'is not None' rather than truthiness to correctly
# handle the edge case where gust speed is exactly 0.0 m/s.
gust_str = ""
if weather.wind.gust is not None:
gust_str = (
f" (gusts up to {weather.wind.gust:.1f} {wind_unit})"
)
visibility_str = ""
if weather.visibility_meters is not None:
visibility_str = (
f"Visibility: {weather.visibility_meters}m\n"
)
return (
f"Current weather for {weather.city_name}, "
f"{weather.country_code} "
f"(as of {obs_time}):\n\n"
f"Conditions: {weather.condition.description.capitalize()}\n"
f"Temperature: {weather.temperature:.1f}°{temp_unit} "
f"(feels like {weather.feels_like:.1f}°{temp_unit})\n"
f"Today's range: {weather.temp_min:.1f}°{temp_unit} "
f"to {weather.temp_max:.1f}°{temp_unit}\n"
f"Humidity: {weather.humidity}%\n"
f"Pressure: {weather.pressure} hPa\n"
f"Wind: {weather.wind.speed:.1f} {wind_unit} "
f"from the {weather.wind.direction_label}{gust_str}\n"
f"{visibility_str}"
f"Coordinates: {weather.latitude:.4f}N, "
f"{weather.longitude:.4f}E"
)
except WeatherServiceError as exc:
# Return errors as plain text rather than raising exceptions.
# MCP tools should return errors as content so the LLM can
# incorporate them gracefully into its response to the user.
return f"Error: {_format_error(exc)}"
# ----------------------------------------------------------------
# TOOL 2: Get Current Weather by Geographic Coordinates
# ----------------------------------------------------------------
@mcp.tool()
async def get_weather_by_coordinates(
latitude: Annotated[
float,
Field(
ge=-90.0,
le=90.0,
description=(
"Geographic latitude in decimal degrees. "
"Positive values are North, negative are South. "
"Valid range: -90.0 to 90.0."
),
),
],
longitude: Annotated[
float,
Field(
ge=-180.0,
le=180.0,
description=(
"Geographic longitude in decimal degrees. "
"Positive values are East, negative are West. "
"Valid range: -180.0 to 180.0."
),
),
],
units: Annotated[
str,
Field(
description=(
"Unit system: 'metric' (Celsius, default), "
"'imperial' (Fahrenheit), or 'standard' (Kelvin)."
),
),
] = "metric",
) -> str:
"""
Get current weather conditions for a location identified by
latitude and longitude coordinates.
Use this tool when the user provides coordinates instead of a city
name, or when you need weather for a precise location that may not
have a well-known city name, such as a remote area, a specific
point of interest, or an exact address.
"""
try:
weather = await service.get_current_weather_by_coords(
latitude=latitude,
longitude=longitude,
units=units,
)
temp_unit = weather.temperature_unit
wind_unit = weather.wind_unit
obs_time = weather.observed_at.strftime("%Y-%m-%d %H:%M UTC")
gust_str = ""
if weather.wind.gust is not None:
gust_str = (
f" (gusts up to {weather.wind.gust:.1f} {wind_unit})"
)
return (
f"Current weather at coordinates "
f"({latitude:.4f}, {longitude:.4f}) "
f"near {weather.city_name}, {weather.country_code} "
f"(as of {obs_time}):\n\n"
f"Conditions: {weather.condition.description.capitalize()}\n"
f"Temperature: {weather.temperature:.1f}°{temp_unit} "
f"(feels like {weather.feels_like:.1f}°{temp_unit})\n"
f"Humidity: {weather.humidity}%\n"
f"Pressure: {weather.pressure} hPa\n"
f"Wind: {weather.wind.speed:.1f} {wind_unit} "
f"from the {weather.wind.direction_label}{gust_str}"
)
except WeatherServiceError as exc:
return f"Error: {_format_error(exc)}"
# ----------------------------------------------------------------
# TOOL 3: Get Weather Forecast
# ----------------------------------------------------------------
@mcp.tool()
async def get_weather_forecast(
city: Annotated[
str,
Field(
description=(
"The city name to get the forecast for. "
"Include country code for precision, e.g., 'Tokyo,JP'."
),
),
],
hours_ahead: Annotated[
int,
Field(
ge=3,
le=120,
description=(
"How many hours of forecast to retrieve. "
"Minimum 3 hours, maximum 120 hours (5 days). "
"Default is 48 hours (2 days)."
),
),
] = 48,
units: Annotated[
str,
Field(
description=(
"Unit system: 'metric' (default), 'imperial', "
"or 'standard'."
),
),
] = "metric",
) -> str:
"""
Get a weather forecast for a city for the next 3 to 120 hours.
Use this tool when the user asks about future weather: what the
weather will be like tomorrow, this weekend, or over the next few
days. The forecast is provided in 3-hour intervals and includes
temperature, sky conditions, wind, and precipitation probability.
For a combined report of current conditions plus forecast, use
get_comprehensive_weather_report instead.
"""
try:
forecast = await service.get_forecast(
city=city,
units=units,
hours_ahead=hours_ahead,
)
temp_unit = forecast.temperature_unit
lines = [
f"Weather forecast for {forecast.city_name}, "
f"{forecast.country_code} "
f"(next {hours_ahead} hours, {units} units):\n"
]
for entry in forecast.entries:
time_str = entry.forecast_time.strftime(
"%a %d %b %H:%M UTC"
)
pop_pct = int(entry.precipitation_probability * 100)
line = (
f" {time_str}: "
f"{entry.temperature:.1f}°{temp_unit}, "
f"{entry.condition.description}"
)
if pop_pct > 0:
line += f", {pop_pct}% chance of rain"
if entry.rain_mm is not None:
line += f", {entry.rain_mm:.1f}mm expected"
lines.append(line)
return "\n".join(lines)
except WeatherServiceError as exc:
return f"Error: {_format_error(exc)}"
# ----------------------------------------------------------------
# TOOL 4: Get Comprehensive Weather Report
# ----------------------------------------------------------------
@mcp.tool()
async def get_comprehensive_weather_report(
city: Annotated[
str,
Field(
description=(
"The city name for the comprehensive report. "
"Include country code, e.g., 'Sydney,AU'."
),
),
],
units: Annotated[
str,
Field(
description=(
"Unit system: 'metric' (default), 'imperial', "
"or 'standard'."
),
),
] = "metric",
) -> str:
"""
Get a comprehensive weather report combining current conditions
and a 48-hour forecast for a city.
Use this tool when the user wants a full weather briefing: current
conditions plus upcoming forecast. This is the best tool to use
when someone asks "what's the weather like in [city]?" without
specifying whether they want current or forecast data, as it
provides both in a single, well-organized response.
This tool makes two API calls concurrently (current + forecast)
and combines them into a single structured report.
"""
try:
report = await service.get_comprehensive_report(
city=city,
units=units,
)
lines = [
f"Comprehensive Weather Report for {report['location']}",
"=" * 50,
"",
"CURRENT CONDITIONS",
"-" * 20,
report["current_conditions"],
"",
"DETAILED CURRENT DATA",
"-" * 20,
]
details = report["current_details"]
for key, value in details.items():
label = key.replace("_", " ").title()
lines.append(f" {label}: {value}")
lines.extend([
"",
"48-HOUR FORECAST",
"-" * 20,
])
lines.extend(report["48_hour_forecast"])
return "\n".join(lines)
except WeatherServiceError as exc:
return f"Error: {_format_error(exc)}"
CHAPTER 10: MCP RESOURCES AND PROMPTS
Tools are the workhorses of an MCP server, but resources and prompts complete the picture. Resources give the LLM access to structured data documents, and prompts give it reusable templates for common reasoning tasks. The forecast resource now explicitly documents its metric-only limitation in its docstring so that any developer reading the code understands the design choice immediately. The prompts module has its unused imports removed.
# src/weather_mcp/tools/weather_resources.py
#
# MCP resource definitions for the Weather MCP Server.
#
# Resources are readable data documents identified by URIs.
# Unlike tools (which execute actions), resources provide data
# that the LLM can read and incorporate into its context.
# Think of resources as "files" that the LLM can open and read.
from __future__ import annotations
import json
from mcp.server.fastmcp import FastMCP
from weather_mcp.domain.exceptions import WeatherServiceError
from weather_mcp.service.weather_service import WeatherService
def register_weather_resources(mcp: FastMCP, service: WeatherService) -> None:
"""
Register all weather-related MCP resources with the server.
Resources use URI templates to identify specific data documents.
The {city} placeholder in the URI is replaced with the actual city
name when the resource is fetched by an MCP client.
"""
@mcp.resource("weather://current/{city}")
async def current_weather_resource(city: str) -> str:
"""
A resource providing current weather data for a city as a JSON document.
This resource is fetched by the LLM when it needs to read current
weather data as a structured document rather than calling a tool.
The response is always in metric units (Celsius, m/s).
URI format: weather://current/{city}
Example: weather://current/Berlin,DE
Units: Always metric (Celsius, m/s, hPa).
"""
try:
weather = await service.get_current_weather(
city=city,
units="metric",
)
document = {
"resource_type": "current_weather",
"city": weather.city_name,
"country": weather.country_code,
"coordinates": {
"latitude": weather.latitude,
"longitude": weather.longitude,
},
"temperature": {
"current": weather.temperature,
"feels_like": weather.feels_like,
"min": weather.temp_min,
"max": weather.temp_max,
"unit": "C",
},
"conditions": {
"main": weather.condition.main,
"description": weather.condition.description,
},
"humidity_percent": weather.humidity,
"pressure_hpa": weather.pressure,
"wind": {
"speed": weather.wind.speed,
"direction_degrees": weather.wind.direction_degrees,
"direction_label": weather.wind.direction_label,
"gust": weather.wind.gust,
"unit": "m/s",
},
"visibility_meters": weather.visibility_meters,
"observed_at_utc": weather.observed_at.isoformat(),
"unit_system": "metric",
}
return json.dumps(document, indent=2)
except WeatherServiceError as exc:
error_doc = {
"error": True,
"message": str(exc),
"city_requested": city,
}
return json.dumps(error_doc, indent=2)
@mcp.resource("weather://forecast/{city}")
async def forecast_resource(city: str) -> str:
"""
A resource providing a 48-hour weather forecast for a city as JSON.
This resource always returns data in metric units (Celsius, m/s).
For imperial or standard units, use the get_weather_forecast tool
instead, which accepts a units parameter.
URI format: weather://forecast/{city}
Example: weather://forecast/Tokyo,JP
Units: Always metric (Celsius, m/s).
"""
try:
forecast = await service.get_forecast(
city=city,
units="metric",
hours_ahead=48,
)
document = {
"resource_type": "weather_forecast",
"city": forecast.city_name,
"country": forecast.country_code,
"unit_system": "metric",
"temperature_unit": "C",
"wind_unit": "m/s",
"forecast_entries": [
{
"time_utc": entry.forecast_time.isoformat(),
"temperature": entry.temperature,
"feels_like": entry.feels_like,
"condition": entry.condition.description,
"wind_speed": entry.wind.speed,
"wind_direction": entry.wind.direction_label,
"precipitation_probability_percent": int(
entry.precipitation_probability * 100
),
"rain_mm": entry.rain_mm,
}
for entry in forecast.entries
],
}
return json.dumps(document, indent=2)
except WeatherServiceError as exc:
return json.dumps(
{"error": True, "message": str(exc)},
indent=2,
)
# src/weather_mcp/tools/weather_prompts.py
#
# MCP prompt definitions for the Weather MCP Server.
#
# Prompts are reusable message templates. When an LLM client requests
# a prompt, the server returns a pre-formatted message that the client
# injects into the conversation. This guides the LLM toward specific
# reasoning patterns for common weather-related tasks.
#
# In MCP SDK 2.0, prompt functions decorated with @mcp.prompt() can
# return a plain str, which FastMCP automatically wraps into the
# required PromptMessage structure. Returning str is the simplest
# and most readable approach.
from __future__ import annotations
from mcp.server.fastmcp import FastMCP
def register_weather_prompts(mcp: FastMCP) -> None:
"""Register all weather-related MCP prompts with the server."""
@mcp.prompt()
async def clothing_recommendation(
city: str,
activity: str = "general",
) -> str:
"""
Generate a prompt for recommending clothing based on weather conditions.
This prompt template guides the LLM to fetch current weather and
forecast data, then reason about appropriate clothing choices for
the specified city and activity.
Parameters
----------
city : str
The city for which to recommend clothing.
activity : str
The type of activity: 'general', 'outdoor', 'formal', 'sports'.
Defaults to 'general'.
"""
return (
f"You are a helpful weather and lifestyle assistant. "
f"The user is in {city} and wants clothing recommendations "
f"for {activity} activities.\n\n"
f"Please:\n"
f"1. Use the get_comprehensive_weather_report tool to fetch "
f"current conditions and the 48-hour forecast for {city}.\n"
f"2. Based on the weather data, recommend appropriate clothing "
f"for today and tomorrow.\n"
f"3. Mention any weather changes the user should prepare for.\n"
f"4. Keep your recommendation practical and specific.\n\n"
f"Start by fetching the weather data now."
)
@mcp.prompt()
async def travel_weather_briefing(
origin_city: str,
destination_city: str,
) -> str:
"""
Generate a prompt for a travel weather briefing comparing two cities.
Use this prompt when a user is planning travel and wants to compare
weather conditions between their current location and destination.
Parameters
----------
origin_city : str
The city the user is traveling from.
destination_city : str
The city the user is traveling to.
"""
return (
f"You are a travel weather advisor. The user is planning to "
f"travel from {origin_city} to {destination_city}.\n\n"
f"Please:\n"
f"1. Fetch comprehensive weather reports for both "
f"{origin_city} and {destination_city} using the "
f"get_comprehensive_weather_report tool.\n"
f"2. Compare the weather conditions between the two cities.\n"
f"3. Highlight any significant weather differences the traveler "
f"should be aware of.\n"
f"4. Provide packing recommendations based on the destination "
f"weather and any forecast changes.\n"
f"5. Note any weather warnings or unusual conditions.\n\n"
f"Fetch both weather reports now and then provide your briefing."
)
CHAPTER 11: THE MCP SERVER - ASSEMBLING THE MACHINE
Now we bring everything together. The server module is the entry point for the entire application. It creates the FastMCP instance, registers all tools, resources, and prompts, and starts the HTTP server. The critical fix here is the addition of "import uvicorn" at the top of the file — without it, the main() function would raise a NameError the moment it was called, despite the rest of the code being perfectly correct.
# src/weather_mcp/server.py
#
# MCP server entry point and application assembly.
#
# This module is the composition root of the application: the one place
# where all layers are wired together. It creates the MCP server, registers
# all tools/resources/prompts, manages the service lifecycle, and starts
# the HTTP server.
#
# Design principle: this module should be thin. It wires things together
# but contains no business logic. If you find yourself writing business
# logic here, move it to the service or domain layer.
from __future__ import annotations
import logging
from contextlib import asynccontextmanager
from typing import AsyncGenerator
import uvicorn
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from mcp.server.fastmcp import FastMCP
from weather_mcp.config import get_settings
from weather_mcp.service.weather_service import WeatherService
from weather_mcp.tools.weather_prompts import register_weather_prompts
from weather_mcp.tools.weather_resources import register_weather_resources
from weather_mcp.tools.weather_tools import register_weather_tools
# Configure structured logging. In production, replace this with a JSON
# formatter that ships logs to your aggregation system (Datadog, ELK, etc.).
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("weather_mcp.server")
def create_mcp_server(service: WeatherService) -> FastMCP:
"""
Create and configure the FastMCP server instance.
This factory function creates the MCP server and registers all
capabilities. It accepts the WeatherService as a parameter so
that it can be called with a mock service during testing.
The service is passed as a closure to each registration function.
The tools store a reference to the service object; they do not call
it immediately. The service connection pool is opened by the FastAPI
lifespan manager before any request can reach the tools, so the
ordering is safe.
Parameters
----------
service : WeatherService
The weather service instance that tools will use.
Returns
-------
FastMCP
A fully configured MCP server ready to handle requests.
"""
mcp = FastMCP(
name="Weather MCP Server",
instructions=(
"This server provides real-time and forecast weather data "
"for cities worldwide via the OpenWeatherMap API. "
"Use get_current_weather for current conditions, "
"get_weather_forecast for upcoming weather, and "
"get_comprehensive_weather_report for a full briefing. "
"Always include the country code with city names to avoid "
"ambiguity, e.g., 'London,GB' not just 'London'. "
"The default unit system is metric (Celsius, m/s)."
),
)
register_weather_tools(mcp, service)
register_weather_resources(mcp, service)
register_weather_prompts(mcp)
logger.info("MCP server configured with tools, resources, and prompts.")
return mcp
def create_app() -> FastAPI:
"""
Create the FastAPI application that hosts the MCP server.
The FastAPI app serves two purposes:
1. It hosts the MCP endpoint at /mcp (the main protocol endpoint).
2. It provides operational endpoints like /health and /info that
are essential for running the server in a production environment
but are not part of the MCP protocol itself.
"""
settings = get_settings()
service = WeatherService()
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
"""
Manage the lifecycle of the WeatherService.
This context manager runs on application startup and shutdown.
It opens the HTTP connection pool when the server starts and
closes it cleanly when the server stops, preventing resource leaks.
All requests are served between the 'yield' and the cleanup code.
"""
logger.info("Starting Weather MCP Server...")
async with service:
logger.info(
"Weather MCP Server ready. "
f"MCP endpoint: http://{settings.mcp_host}:"
f"{settings.mcp_port}/mcp"
)
yield
logger.info("Weather MCP Server shut down cleanly.")
app = FastAPI(
title="Weather MCP Server",
description=(
"A production-ready MCP server wrapping the OpenWeatherMap API."
),
version="1.0.0",
lifespan=lifespan,
)
# Create the MCP server and mount it at /mcp.
# streamable_http_app() returns an ASGI application implementing
# the MCP 2026-07-28 stateless HTTP transport.
mcp_server = create_mcp_server(service)
app.mount("/mcp", mcp_server.streamable_http_app())
@app.get("/health", tags=["Operations"])
async def health_check() -> JSONResponse:
"""
Health check endpoint for load balancers and monitoring systems.
Returns HTTP 200 when the server is ready to handle requests.
"""
return JSONResponse(
{"status": "healthy", "service": "weather-mcp"}
)
@app.get("/info", tags=["Operations"])
async def server_info() -> JSONResponse:
"""
Server information endpoint. Returns metadata about this MCP server,
including the list of available tools, resources, and prompts.
"""
return JSONResponse({
"name": "Weather MCP Server",
"version": "1.0.0",
"mcp_spec_version": "2026-07-28",
"mcp_endpoint": "/mcp",
"tools": [
"get_current_weather",
"get_weather_by_coordinates",
"get_weather_forecast",
"get_comprehensive_weather_report",
],
"resources": [
"weather://current/{city}",
"weather://forecast/{city}",
],
"prompts": [
"clothing_recommendation",
"travel_weather_briefing",
],
})
return app
def main() -> None:
"""
Application entry point. Called by the 'weather-mcp' CLI command
defined in pyproject.toml under [project.scripts].
"""
settings = get_settings()
app = create_app()
logger.info(
f"Starting uvicorn on {settings.mcp_host}:{settings.mcp_port}"
)
uvicorn.run(
app,
host=settings.mcp_host,
port=settings.mcp_port,
workers=1,
log_level="info",
)
if __name__ == "__main__":
main()
CHAPTER 12: LLM CLIENTS - TALKING TO THE SERVER
Having a beautiful MCP server is only half the story. The other half is the client: the code that connects an LLM to our server and orchestrates the conversation. We will build two complete client implementations. The first uses cloud-hosted LLMs: GPT-6 Astra, Claude Fable 5.1, and Gemini 3.8 Flash. The second uses a locally running Llama 4 model through Ollama.
The key insight in building an MCP client is the tool-use loop. Modern LLMs do not just call a tool and stop; they may call multiple tools in sequence, use the results of one tool to inform the next, and only produce a final answer when they have gathered all the information they need. Our client must implement this loop correctly.
Three important corrections apply to the cloud client. First, the Anthropic client is changed from the synchronous anthropic.Anthropic to the async anthropic.AsyncAnthropic, and the messages.create call is properly awaited. Calling a synchronous blocking function inside an async function without awaiting it or running it in an executor would freeze the entire event loop, preventing any other coroutine from running. Second, the Gemini tool conversion now uses a proper type-mapping function instead of blindly converting every property to STRING type. Third, the Gemini text extraction is made safe by iterating over response parts rather than calling response.text, which can raise a ValueError when the response contains no text content. The MCP server URL is also read from settings rather than hardcoded.
# clients/cloud_llm_client.py
#
# MCP client for cloud-hosted LLMs.
#
# This client demonstrates how to connect three different cloud LLMs
# (Claude Fable 5.1, GPT-6 Astra, Gemini 3.8 Flash) to our Weather
# MCP Server and run a multi-turn conversation with tool use.
#
# The tool-use agentic loop: send a user message to the LLM, process
# any tool calls the LLM makes (by calling our MCP server), feed the
# results back to the LLM, and repeat until the LLM produces a final
# answer with no tool calls.
from __future__ import annotations
import asyncio
import json
import os
from enum import Enum
from typing import Any
import anthropic
import google.genai as genai
import google.genai.types as genai_types
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
from openai import AsyncOpenAI
from weather_mcp.config import get_settings
class LLMProvider(str, Enum):
"""Supported cloud LLM providers."""
ANTHROPIC = "anthropic"
OPENAI = "openai"
GOOGLE = "google"
def _json_schema_type_to_genai(type_str: str) -> genai_types.Type:
"""
Map a JSON Schema type string to the corresponding google.genai Type enum.
This mapping is necessary because MCP tool schemas use JSON Schema type
names (strings like "integer", "number") while the Google Generative AI
SDK uses its own Type enum. Without this mapping, all parameters would
be treated as strings, causing type errors when the model passes numeric
values as tool arguments.
"""
mapping: dict[str, genai_types.Type] = {
"string": genai_types.Type.STRING,
"integer": genai_types.Type.INTEGER,
"number": genai_types.Type.NUMBER,
"boolean": genai_types.Type.BOOLEAN,
"array": genai_types.Type.ARRAY,
"object": genai_types.Type.OBJECT,
}
return mapping.get(type_str.lower(), genai_types.Type.STRING)
class MCPWeatherClient:
"""
A generic MCP client that connects any supported cloud LLM to the
Weather MCP Server and manages the full tool-use conversation loop.
This client uses the MCP Python SDK's ClientSession to communicate
with the server. The ClientSession handles the low-level details of
the MCP protocol (tool discovery, tool invocation, response parsing)
so we can focus on the LLM-specific parts of the integration.
"""
def __init__(self, provider: LLMProvider) -> None:
self.provider = provider
settings = get_settings()
# Read the server URL from settings rather than hardcoding it.
# This allows the URL to be configured per environment via .env.
self._mcp_server_url = settings.mcp_server_url
self._tools_cache: list[dict] | None = None
async def _fetch_mcp_tools(
self,
session: ClientSession,
) -> list[dict]:
"""
Discover available tools from the MCP server.
The MCP SDK's session.list_tools() sends a tools/list request and
returns the list of available tools with their names, descriptions,
and input schemas. We cache the result for the duration of a
conversation to avoid redundant network calls.
"""
if self._tools_cache is not None:
return self._tools_cache
tools_result = await session.list_tools()
self._tools_cache = [
{
"name": tool.name,
"description": tool.description or "",
"input_schema": tool.inputSchema or {
"type": "object",
"properties": {},
},
}
for tool in tools_result.tools
]
return self._tools_cache
async def _call_mcp_tool(
self,
session: ClientSession,
tool_name: str,
tool_arguments: dict[str, Any],
) -> str:
"""
Invoke a tool on the MCP server and return its text result.
Sends a tools/call request and extracts all text content items
from the response, joining them into a single string.
"""
result = await session.call_tool(
name=tool_name,
arguments=tool_arguments,
)
text_parts = [
item.text
for item in result.content
if hasattr(item, "text")
]
return "\n".join(text_parts) if text_parts else "Tool returned no output."
# ----------------------------------------------------------------
# ANTHROPIC CLAUDE FABLE 5.1 INTEGRATION
# ----------------------------------------------------------------
async def chat_with_claude(
self,
user_message: str,
session: ClientSession,
) -> str:
"""
Run a multi-turn conversation with Claude Fable 5.1.
Uses anthropic.AsyncAnthropic so that the messages.create() call
is properly awaited without blocking the event loop. The tool-use
loop continues until Claude produces a stop_reason of "end_turn",
meaning it has finished calling tools and is ready to respond.
"""
settings = get_settings()
# AsyncAnthropic is the non-blocking async variant of the client.
# Never use the synchronous Anthropic client inside an async function.
client = anthropic.AsyncAnthropic(
api_key=settings.anthropic_api_key or os.environ.get(
"ANTHROPIC_API_KEY", ""
),
)
mcp_tools = await self._fetch_mcp_tools(session)
anthropic_tools = [
{
"name": tool["name"],
"description": tool["description"],
"input_schema": tool["input_schema"],
}
for tool in mcp_tools
]
messages: list[dict] = [
{"role": "user", "content": user_message},
]
while True:
response = await client.messages.create(
model="claude-fable-5-1",
max_tokens=4096,
tools=anthropic_tools,
messages=messages,
)
if response.stop_reason == "end_turn":
final_text = "".join(
block.text
for block in response.content
if hasattr(block, "text")
)
return final_text
# Claude wants to call tools. Add its response to history first.
messages.append({"role": "assistant", "content": response.content})
# Execute each tool call and collect results.
tool_results = []
for block in response.content:
if block.type == "tool_use":
print(
f" [Claude -> tool: {block.name} "
f"args: {json.dumps(block.input, indent=2)}]"
)
tool_result_text = await self._call_mcp_tool(
session=session,
tool_name=block.name,
tool_arguments=block.input,
)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": tool_result_text,
})
messages.append({"role": "user", "content": tool_results})
# ----------------------------------------------------------------
# OPENAI GPT-6 ASTRA INTEGRATION
# ----------------------------------------------------------------
async def chat_with_gpt(
self,
user_message: str,
session: ClientSession,
) -> str:
"""
Run a multi-turn conversation with GPT-6 Astra.
AsyncOpenAI is already async-native, so all calls are properly
awaited without blocking the event loop.
"""
settings = get_settings()
client = AsyncOpenAI(
api_key=settings.openai_api_key or os.environ.get(
"OPENAI_API_KEY", ""
),
)
mcp_tools = await self._fetch_mcp_tools(session)
openai_tools = [
{
"type": "function",
"function": {
"name": tool["name"],
"description": tool["description"],
"parameters": tool["input_schema"],
},
}
for tool in mcp_tools
]
messages: list[dict] = [
{
"role": "system",
"content": (
"You are a helpful weather assistant. Use the available "
"tools to fetch real-time weather data and provide "
"accurate, helpful responses to the user's questions."
),
},
{"role": "user", "content": user_message},
]
while True:
response = await client.chat.completions.create(
model="gpt-6-astra",
messages=messages,
tools=openai_tools,
tool_choice="auto",
max_tokens=4096,
)
message = response.choices[0].message
if not message.tool_calls:
return message.content or ""
messages.append(message)
for tool_call in message.tool_calls:
tool_args = json.loads(tool_call.function.arguments)
print(
f" [GPT-6 -> tool: {tool_call.function.name} "
f"args: {json.dumps(tool_args, indent=2)}]"
)
tool_result = await self._call_mcp_tool(
session=session,
tool_name=tool_call.function.name,
tool_arguments=tool_args,
)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": tool_result,
})
# ----------------------------------------------------------------
# GOOGLE GEMINI 3.8 FLASH INTEGRATION
# ----------------------------------------------------------------
async def chat_with_gemini(
self,
user_message: str,
session: ClientSession,
) -> str:
"""
Run a multi-turn conversation with Gemini 3.8 Flash.
Uses the google-genai SDK (the 2026 successor to google-generativeai).
Tool schemas are converted using the proper type-mapping function to
ensure numeric and boolean parameters are typed correctly.
Text extraction iterates over response parts rather than calling
response.text, which is fragile when the response contains no text.
"""
settings = get_settings()
client = genai.Client(
api_key=settings.google_api_key or os.environ.get(
"GOOGLE_API_KEY", ""
),
)
mcp_tools = await self._fetch_mcp_tools(session)
# Convert MCP tool definitions to Gemini FunctionDeclaration objects.
# Each property's JSON Schema type is mapped to the correct genai Type.
gemini_functions = []
for tool in mcp_tools:
schema_props = tool["input_schema"].get("properties", {})
required_props = tool["input_schema"].get("required", [])
parameters = genai_types.Schema(
type=genai_types.Type.OBJECT,
properties={
prop_name: genai_types.Schema(
type=_json_schema_type_to_genai(
prop_data.get("type", "string")
),
description=prop_data.get("description", ""),
)
for prop_name, prop_data in schema_props.items()
},
required=required_props,
)
gemini_functions.append(
genai_types.FunctionDeclaration(
name=tool["name"],
description=tool["description"],
parameters=parameters,
)
)
gemini_tools = [genai_types.Tool(function_declarations=gemini_functions)]
# Build the conversation history for Gemini's multi-turn chat.
contents: list[genai_types.Content] = [
genai_types.Content(
role="user",
parts=[genai_types.Part(text=user_message)],
)
]
while True:
response = await asyncio.to_thread(
client.models.generate_content,
model="gemini-3.8-flash",
contents=contents,
config=genai_types.GenerateContentConfig(tools=gemini_tools),
)
# Extract any function calls from the response.
function_calls = []
for candidate in response.candidates:
for part in candidate.content.parts:
if part.function_call and part.function_call.name:
function_calls.append(part.function_call)
if not function_calls:
# No function calls: extract text safely by iterating parts.
text_parts = []
for candidate in response.candidates:
for part in candidate.content.parts:
if part.text:
text_parts.append(part.text)
return "\n".join(text_parts) if text_parts else ""
# Add the model's response (with function calls) to history.
contents.append(response.candidates[0].content)
# Execute each function call and collect results.
function_response_parts = []
for fc in function_calls:
tool_args = dict(fc.args)
print(
f" [Gemini -> tool: {fc.name} "
f"args: {json.dumps(tool_args, indent=2)}]"
)
tool_result = await self._call_mcp_tool(
session=session,
tool_name=fc.name,
tool_arguments=tool_args,
)
function_response_parts.append(
genai_types.Part(
function_response=genai_types.FunctionResponse(
name=fc.name,
response={"result": tool_result},
)
)
)
# Add function results to the conversation history.
contents.append(
genai_types.Content(
role="tool",
parts=function_response_parts,
)
)
async def chat(
self,
user_message: str,
provider: LLMProvider | None = None,
) -> str:
"""
Send a message to the configured LLM provider and get a response.
This is the main entry point for the client. It establishes an
MCP session with the server and dispatches to the appropriate
provider-specific method.
"""
active_provider = provider or self.provider
async with streamablehttp_client(self._mcp_server_url) as (
read,
write,
_,
):
async with ClientSession(read, write) as session:
if active_provider == LLMProvider.ANTHROPIC:
return await self.chat_with_claude(user_message, session)
elif active_provider == LLMProvider.OPENAI:
return await self.chat_with_gpt(user_message, session)
elif active_provider == LLMProvider.GOOGLE:
return await self.chat_with_gemini(user_message, session)
else:
raise ValueError(
f"Unsupported provider: {active_provider}"
)
async def main() -> None:
"""
Demonstrate the cloud LLM client with all three providers.
Ensure the MCP server is running before executing this script.
"""
questions = [
"What's the weather like in Munich, Germany right now?",
"Should I bring an umbrella if I'm going to Tokyo tomorrow?",
"Compare the weather in New York and London for the next 48 hours.",
]
providers = [
LLMProvider.ANTHROPIC,
LLMProvider.OPENAI,
LLMProvider.GOOGLE,
]
provider_names = {
LLMProvider.ANTHROPIC: "Claude Fable 5.1",
LLMProvider.OPENAI: "GPT-6 Astra",
LLMProvider.GOOGLE: "Gemini 3.8 Flash",
}
for provider in providers:
client = MCPWeatherClient(provider=provider)
question = questions[providers.index(provider)]
print(f"\n{'=' * 60}")
print(f"Provider: {provider_names[provider]}")
print(f"Question: {question}")
print("-" * 60)
try:
answer = await client.chat(question)
print(f"Answer:\n{answer}")
except Exception as exc:
print(f"Error: {type(exc).__name__}: {exc}")
if __name__ == "__main__":
asyncio.run(main())
CHAPTER 13: THE LOCAL LLM CLIENT WITH OLLAMA AND LLAMA 4
The local LLM client is arguably the most interesting piece of the entire tutorial, because it demonstrates something remarkable: you can run a state-of-the-art language model entirely on your own hardware, connect it to a real-world REST API through MCP, and get results that rival cloud-hosted models for many practical tasks. The Llama 4 Maverick model runs comfortably on a modern consumer GPU with 16 GB of VRAM, or on CPU with 32 GB of RAM.
The critical fix here is the removal of the erroneous "await" before the ollama.pull() call. The ollama async client's pull() method with stream=True returns an async generator directly, not a coroutine. Placing "await" before it would cause a TypeError at runtime because you cannot await an async generator. The correct pattern is to use "async for" directly without "await". The MCP server URL is also read from settings.
# clients/local_llm_client.py
#
# MCP client for local LLMs running through Ollama.
#
# This client bridges between Ollama's OpenAI-compatible API and our
# MCP server. Since Ollama is not natively an MCP client, we implement
# the tool-use loop manually: fetch tools from MCP, convert them to
# Ollama's function-calling format, run the model, detect tool calls,
# execute them via MCP, feed results back, and repeat until done.
from __future__ import annotations
import asyncio
import json
import os
from typing import Any
import ollama
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
from weather_mcp.config import get_settings
class LocalMCPWeatherClient:
"""
An MCP client that uses a locally running Ollama model.
The tool-use agentic loop:
1. Fetch available tools from the MCP server (once, then cache).
2. Send the user's message to Ollama with tool definitions.
3. If Ollama's response contains tool calls, execute them via MCP.
4. Add tool results to the conversation history.
5. Send the updated conversation back to Ollama.
6. Repeat from step 3 until Ollama produces a response with no calls.
7. Return the final text response.
"""
# Maximum tool-call iterations to prevent infinite loops.
# Well-designed conversations rarely exceed 5 turns.
MAX_TOOL_ITERATIONS = 10
def __init__(
self,
model: str | None = None,
ollama_base_url: str | None = None,
) -> None:
settings = get_settings()
self.model = model or settings.ollama_model
self.ollama_base_url = ollama_base_url or settings.ollama_base_url
# Read the MCP server URL from settings for configurability.
self._mcp_server_url = settings.mcp_server_url
self._ollama = ollama.AsyncClient(host=self.ollama_base_url)
self._tools_cache: list[dict] | None = None
async def _ensure_model_available(self) -> None:
"""
Check that the requested model is available in Ollama.
If not, pull it automatically. This convenience feature makes
the first run smoother by avoiding a separate 'ollama pull' step.
"""
try:
models_response = await self._ollama.list()
available = [m.model for m in models_response.models]
if self.model not in available:
print(
f"Model '{self.model}' not found locally. "
"Pulling from Ollama registry... "
"(this may take several minutes on first run)"
)
# ollama.AsyncClient.pull() with stream=True returns an
# async generator directly. Do NOT use 'await' here;
# use 'async for' to iterate over the progress events.
async for progress in self._ollama.pull(
self.model,
stream=True,
):
if progress.status:
print(f" {progress.status}", end="\r")
print(f"\nModel '{self.model}' is ready.")
except Exception as exc:
print(
f"Warning: Could not verify model availability: {exc}. "
"Proceeding anyway."
)
async def _fetch_tools_as_ollama_format(
self,
session: ClientSession,
) -> list[dict]:
"""
Fetch MCP tools and convert them to Ollama's function-calling format.
Ollama uses the OpenAI function-calling format, where each tool is
an object with type "function" and a "function" sub-object containing
the name, description, and parameter schema. The input_schema from
MCP is already valid JSON Schema, so it can be used directly as
the "parameters" field without any conversion.
"""
if self._tools_cache is not None:
return self._tools_cache
tools_result = await session.list_tools()
self._tools_cache = [
{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description or "",
"parameters": tool.inputSchema or {
"type": "object",
"properties": {},
},
},
}
for tool in tools_result.tools
]
return self._tools_cache
async def _execute_tool_call(
self,
session: ClientSession,
tool_name: str,
tool_arguments: dict[str, Any],
) -> str:
"""
Execute a single tool call via the MCP server and return the result.
This is the bridge between Ollama's tool-call request and the MCP server.
"""
result = await session.call_tool(
name=tool_name,
arguments=tool_arguments,
)
text_parts = [
item.text
for item in result.content
if hasattr(item, "text")
]
return "\n".join(text_parts) if text_parts else "No result returned."
async def chat(
self,
user_message: str,
system_prompt: str | None = None,
) -> str:
"""
Send a message to the local Llama 4 model and get a response,
with full MCP tool-use support.
Parameters
----------
user_message : str
The user's question or request.
system_prompt : str, optional
A custom system prompt. If not provided, a sensible default
instructs the model to use weather tools for real-time data.
Returns
-------
str
The model's final text response after all tool calls complete.
"""
await self._ensure_model_available()
default_system = (
"You are a helpful weather assistant with access to real-time "
"weather data tools. When the user asks about weather, always "
"use the available tools to fetch current data rather than "
"relying on your training knowledge, which may be outdated. "
"Be concise, accurate, and helpful."
)
async with streamablehttp_client(self._mcp_server_url) as (
read,
write,
_,
):
async with ClientSession(read, write) as session:
ollama_tools = await self._fetch_tools_as_ollama_format(session)
messages: list[dict] = [
{
"role": "system",
"content": system_prompt or default_system,
},
{
"role": "user",
"content": user_message,
},
]
print(
f"\n[Using local model: {self.model} "
f"via Ollama at {self.ollama_base_url}]"
)
for iteration in range(self.MAX_TOOL_ITERATIONS):
print(
f"[Iteration {iteration + 1}/"
f"{self.MAX_TOOL_ITERATIONS}] "
f"Sending {len(messages)} messages to model..."
)
response = await self._ollama.chat(
model=self.model,
messages=messages,
tools=ollama_tools,
options={
# Temperature 0.1 for factual tool-use tasks.
"temperature": 0.1,
# Conservative context size for fast inference.
# Llama 4 Maverick supports up to 128K tokens.
"num_ctx": 8192,
},
)
assistant_message = response.message
tool_calls = assistant_message.tool_calls or []
# Add the assistant's response to the conversation history.
# We store tool_calls as a list (empty if none) for
# consistency in the messages format.
messages.append({
"role": "assistant",
"content": assistant_message.content or "",
"tool_calls": [
{
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments,
}
}
for tc in tool_calls
],
})
if not tool_calls:
# No tool calls: the model has produced its final answer.
print(
f"[Final answer produced after "
f"{iteration + 1} iteration(s).]"
)
return assistant_message.content or ""
print(
f"[Model requested {len(tool_calls)} tool call(s).]"
)
for tool_call in tool_calls:
fn = tool_call.function
# Ollama may return arguments as a JSON string or a dict.
# Handle both cases defensively.
if isinstance(fn.arguments, str):
try:
args = json.loads(fn.arguments)
except json.JSONDecodeError:
args = {}
else:
args = fn.arguments or {}
print(
f" -> Calling MCP tool: '{fn.name}' "
f"with args: {json.dumps(args, indent=4)}"
)
tool_result = await self._execute_tool_call(
session=session,
tool_name=fn.name,
tool_arguments=args,
)
preview = (
tool_result[:100] + "..."
if len(tool_result) > 100
else tool_result
)
print(f" <- Tool result: {preview}")
# Append the tool result with the 'tool' role.
# Ollama matches tool results to tool calls by
# position in the conversation, not by ID.
messages.append({
"role": "tool",
"content": tool_result,
})
return (
"I was unable to complete the request within the maximum "
f"number of tool-use iterations ({self.MAX_TOOL_ITERATIONS}). "
"Please try rephrasing your question."
)
async def demo_local_llm() -> None:
"""
Demonstrate the local LLM client with a series of weather questions.
Prerequisites:
- Terminal 1: uv run weather-mcp (MCP server must be running)
- Terminal 2: ollama serve (Ollama must be running)
- Terminal 3: uv run python clients/local_llm_client.py
"""
client = LocalMCPWeatherClient()
questions = [
"What is the current temperature in Tokyo?",
"Will it rain in London over the next two days?",
"Give me a full weather briefing for Berlin, Germany.",
]
for question in questions:
print(f"\n{'=' * 60}")
print(f"Question: {question}")
print("=" * 60)
try:
answer = await client.chat(question)
print(f"\nAnswer:\n{answer}")
except Exception as exc:
print(f"Error: {type(exc).__name__}: {exc}")
print()
if __name__ == "__main__":
asyncio.run(demo_local_llm())
CHAPTER 14: TESTING - TRUST BUT VERIFY
A server without tests is a server waiting to fail at the worst possible moment. We write tests at two levels: unit tests for the domain and service layers, and integration tests for the HTTP adapter. Our testing strategy uses respx to mock the OpenWeatherMap HTTP calls, which means tests run without making real network requests and without consuming API quota.
The critical addition here is conftest.py, which provides a pytest fixture that sets the OWM_API_KEY environment variable to a dummy value before any test runs. Without this fixture, every test that instantiates OpenWeatherMapClient would fail immediately with a Pydantic ValidationError because the Settings class requires OWM_API_KEY to be present.
# tests/conftest.py
#
# Shared pytest fixtures for the Weather MCP Server test suite.
#
# This file is automatically discovered by pytest and its fixtures are
# available to all test modules without explicit imports. The most
# important fixture here is 'mock_settings', which prevents tests from
# failing due to missing environment variables by providing dummy values
# for all required settings.
from __future__ import annotations
import pytest
from weather_mcp.config import get_settings
@pytest.fixture(autouse=True)
def mock_settings(monkeypatch: pytest.MonkeyPatch) -> None:
"""
Set dummy environment variables for all required settings.
This fixture runs automatically for every test (autouse=True).
It ensures that tests never fail because OWM_API_KEY or other
required environment variables are absent from the test environment.
The fixture also clears the lru_cache on get_settings() before
and after each test, ensuring that each test starts with a fresh
settings object that reflects the monkeypatched environment.
"""
# Clear any cached settings from previous tests.
get_settings.cache_clear()
# Set dummy values for all required environment variables.
monkeypatch.setenv("OWM_API_KEY", "test_api_key_dummy_value")
monkeypatch.setenv("OWM_BASE_URL", "https://api.openweathermap.org")
monkeypatch.setenv("OWM_DEFAULT_UNITS", "metric")
monkeypatch.setenv("MCP_SERVER_URL", "http://localhost:8000/mcp")
yield
# Clear the cache again after the test so the next test gets fresh settings.
get_settings.cache_clear()
# tests/__init__.py
# This file is intentionally empty.
# Its presence tells Python that the tests/ directory is a package,
# which enables relative imports within the test suite if needed.
# tests/test_tools.py
#
# Unit tests for the domain models and service layer.
#
# These tests verify that domain logic is correct in isolation.
# They mock the OpenWeatherMapClient to test the service without
# making real HTTP calls or needing a real API key.
from __future__ import annotations
from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from weather_mcp.domain.models import (
CurrentWeather,
UnitSystem,
WeatherCondition,
WindInfo,
)
from weather_mcp.service.weather_service import WeatherService
def make_test_weather(
city: str = "London",
country: str = "GB",
temperature: float = 18.5,
humidity: int = 62,
gust: float | None = None,
) -> CurrentWeather:
"""
Factory function that creates a CurrentWeather object for testing.
Having a single factory function means that if the CurrentWeather
model changes, we only need to update one place in the test suite.
"""
return CurrentWeather(
city_name=city,
country_code=country,
latitude=51.5085,
longitude=-0.1257,
temperature=temperature,
feels_like=17.9,
temp_min=16.2,
temp_max=20.1,
humidity=humidity,
pressure=1018,
visibility_meters=10000,
condition=WeatherCondition(
main="Clear",
description="clear sky",
icon_code="01d",
),
wind=WindInfo(speed=3.6, direction_degrees=220, gust=gust),
unit_system=UnitSystem.METRIC,
observed_at=datetime(2026, 9, 8, 12, 0, 0, tzinfo=timezone.utc),
)
class TestWeatherService:
"""Tests for the WeatherService layer using a mocked HTTP adapter."""
@pytest.fixture
def mock_client(self) -> MagicMock:
"""Create a mock OpenWeatherMapClient for injection into the service."""
mock = MagicMock()
mock.open = AsyncMock()
mock.close = AsyncMock()
mock.get_current_weather = AsyncMock(
return_value=make_test_weather()
)
return mock
@pytest.mark.asyncio
async def test_get_current_weather_delegates_to_client(
self,
mock_client: MagicMock,
) -> None:
"""
Verify that get_current_weather calls the underlying client with
the correct parameters and returns the result unchanged.
"""
with patch(
"weather_mcp.service.weather_service.OpenWeatherMapClient",
return_value=mock_client,
):
service = WeatherService()
service._client = mock_client
result = await service.get_current_weather(
city="London,GB",
units="metric",
)
mock_client.get_current_weather.assert_called_once_with(
city="London,GB",
units="metric",
)
assert result.city_name == "London"
assert result.temperature == 18.5
assert result.humidity == 62
@pytest.mark.asyncio
async def test_get_current_weather_returns_correct_units(
self,
mock_client: MagicMock,
) -> None:
"""
Verify that temperature_unit and wind_unit properties return
the correct symbols for the metric unit system.
"""
with patch(
"weather_mcp.service.weather_service.OpenWeatherMapClient",
return_value=mock_client,
):
service = WeatherService()
service._client = mock_client
result = await service.get_current_weather("London,GB")
assert result.temperature_unit == "C"
assert result.wind_unit == "m/s"
@pytest.mark.asyncio
async def test_to_summary_string_contains_all_fields(
self,
mock_client: MagicMock,
) -> None:
"""
Verify that to_summary_string() produces a string containing
all expected pieces of information.
"""
weather = make_test_weather(
city="Berlin",
country="DE",
temperature=22.0,
)
summary = weather.to_summary_string()
assert "Berlin" in summary
assert "DE" in summary
assert "22.0°C" in summary
assert "clear sky" in summary
assert "62%" in summary
class TestWindDirectionLabel:
"""
Tests for the wind direction label computation in WindInfo.
We test cardinal and intercardinal directions plus the boundary
condition at 359 degrees (which should map back to North).
"""
def test_north_at_zero_degrees(self) -> None:
wind = WindInfo(speed=5.0, direction_degrees=0)
assert wind.direction_label == "N"
def test_north_at_359_degrees(self) -> None:
"""359 degrees is just west of North and should still map to N."""
wind = WindInfo(speed=5.0, direction_degrees=359)
assert wind.direction_label == "N"
def test_east_at_90_degrees(self) -> None:
wind = WindInfo(speed=5.0, direction_degrees=90)
assert wind.direction_label == "E"
def test_south_at_180_degrees(self) -> None:
wind = WindInfo(speed=5.0, direction_degrees=180)
assert wind.direction_label == "S"
def test_southwest_at_225_degrees(self) -> None:
wind = WindInfo(speed=5.0, direction_degrees=225)
assert wind.direction_label == "SW"
def test_west_at_270_degrees(self) -> None:
wind = WindInfo(speed=5.0, direction_degrees=270)
assert wind.direction_label == "W"
class TestGustHandling:
"""
Tests for the None-safety of gust-related logic.
The gust field is Optional[float], meaning it can be None when the
weather station does not report gust data. The to_summary_string()
method and tool formatting must handle None correctly and must also
correctly handle gust=0.0 (a valid value that is falsy in Python).
"""
def test_gust_none_does_not_appear_in_summary(self) -> None:
"""When gust is None, to_summary_string should not mention gusts."""
weather = make_test_weather(gust=None)
# to_summary_string doesn't include gust; this tests that
# accessing wind.gust is None-safe in general usage.
assert weather.wind.gust is None
def test_gust_zero_is_not_none(self) -> None:
"""
Verify that gust=0.0 is correctly identified as not None.
This is the key test: 'if gust' would be False for 0.0,
but 'if gust is not None' is correctly True.
"""
weather = make_test_weather(gust=0.0)
assert weather.wind.gust is not None
assert weather.wind.gust == 0.0
def test_gust_positive_value(self) -> None:
weather = make_test_weather(gust=5.1)
assert weather.wind.gust == 5.1
assert weather.wind.gust is not None
# tests/test_client_integration.py
#
# Integration tests for the OpenWeatherMap HTTP adapter.
#
# These tests use respx to mock HTTP calls, verifying that the adapter
# correctly maps API responses to domain models and correctly translates
# API errors into domain exceptions. No real network calls are made.
from __future__ import annotations
import httpx
import pytest
import respx
from weather_mcp.adapters.owm_client import OpenWeatherMapClient
from weather_mcp.domain.exceptions import (
ApiKeyError,
LocationNotFoundError,
RateLimitError,
)
# A realistic sample response from /data/2.5/weather.
# This mirrors the actual OWM API response format exactly.
SAMPLE_CURRENT_WEATHER_RESPONSE = {
"coord": {"lon": -0.1257, "lat": 51.5085},
"weather": [
{
"id": 800,
"main": "Clear",
"description": "clear sky",
"icon": "01d",
}
],
"main": {
"temp": 18.5,
"feels_like": 17.9,
"temp_min": 16.2,
"temp_max": 20.1,
"pressure": 1018,
"humidity": 62,
},
"visibility": 10000,
"wind": {"speed": 3.6, "deg": 220, "gust": 5.1},
"dt": 1725796800,
"sys": {"country": "GB"},
"name": "London",
"cod": 200,
}
@pytest.fixture
def mock_owm_api():
"""
Activate respx mocking for all HTTP calls during a test.
Any HTTP call not explicitly mocked will raise an error, preventing
accidental real network calls.
"""
with respx.mock(assert_all_called=False) as mock:
yield mock
@pytest.mark.asyncio
async def test_get_current_weather_success(
mock_owm_api: respx.MockRouter,
) -> None:
"""
Verify that a successful API response is correctly mapped to a
CurrentWeather domain model with all fields populated accurately.
"""
mock_owm_api.get(
"https://api.openweathermap.org/data/2.5/weather"
).mock(
return_value=httpx.Response(
200,
json=SAMPLE_CURRENT_WEATHER_RESPONSE,
)
)
async with OpenWeatherMapClient() as client:
weather = await client.get_current_weather("London,GB")
assert weather.city_name == "London"
assert weather.country_code == "GB"
assert weather.temperature == 18.5
assert weather.feels_like == 17.9
assert weather.humidity == 62
assert weather.pressure == 1018
assert weather.condition.main == "Clear"
assert weather.condition.description == "clear sky"
assert weather.wind.speed == 3.6
assert weather.wind.direction_degrees == 220
assert weather.wind.gust == 5.1
assert weather.wind.direction_label == "SW"
assert weather.visibility_meters == 10000
@pytest.mark.asyncio
async def test_get_current_weather_city_not_found(
mock_owm_api: respx.MockRouter,
) -> None:
"""
Verify that a 404 HTTP response raises LocationNotFoundError.
"""
mock_owm_api.get(
"https://api.openweathermap.org/data/2.5/weather"
).mock(
return_value=httpx.Response(
404,
json={"cod": "404", "message": "city not found"},
)
)
async with OpenWeatherMapClient() as client:
with pytest.raises(LocationNotFoundError):
await client.get_current_weather("NonExistentCity12345")
@pytest.mark.asyncio
async def test_get_current_weather_owm_200_with_error_body(
mock_owm_api: respx.MockRouter,
) -> None:
"""
Verify that HTTP 200 with OWM's non-standard error body raises
LocationNotFoundError. OWM sometimes returns 200 OK with a
{"cod":"404","message":"city not found"} body, which is non-standard
but must be handled correctly.
"""
mock_owm_api.get(
"https://api.openweathermap.org/data/2.5/weather"
).mock(
return_value=httpx.Response(
200,
json={"cod": "404", "message": "city not found"},
)
)
async with OpenWeatherMapClient() as client:
with pytest.raises(LocationNotFoundError):
await client.get_current_weather("GhostCity")
@pytest.mark.asyncio
async def test_get_current_weather_invalid_api_key(
mock_owm_api: respx.MockRouter,
) -> None:
"""
Verify that a 401 HTTP response raises ApiKeyError.
"""
mock_owm_api.get(
"https://api.openweathermap.org/data/2.5/weather"
).mock(
return_value=httpx.Response(
401,
json={"cod": 401, "message": "Invalid API key."},
)
)
async with OpenWeatherMapClient() as client:
with pytest.raises(ApiKeyError):
await client.get_current_weather("London,GB")
@pytest.mark.asyncio
async def test_get_current_weather_rate_limited(
mock_owm_api: respx.MockRouter,
) -> None:
"""
Verify that a 429 HTTP response raises RateLimitError.
"""
mock_owm_api.get(
"https://api.openweathermap.org/data/2.5/weather"
).mock(
return_value=httpx.Response(429, text="Too Many Requests")
)
async with OpenWeatherMapClient() as client:
with pytest.raises(RateLimitError):
await client.get_current_weather("London,GB")
@pytest.mark.asyncio
async def test_wind_direction_sw_from_220_degrees(
mock_owm_api: respx.MockRouter,
) -> None:
"""
Verify that 220 degrees maps to the SW compass direction.
220 degrees is in the southwest quadrant (202.5 to 225 degrees).
"""
mock_owm_api.get(
"https://api.openweathermap.org/data/2.5/weather"
).mock(
return_value=httpx.Response(
200,
json=SAMPLE_CURRENT_WEATHER_RESPONSE,
)
)
async with OpenWeatherMapClient() as client:
weather = await client.get_current_weather("London,GB")
assert weather.wind.direction_label == "SW"
CHAPTER 15: RUNNING THE COMPLETE SYSTEM
Let us now walk through running the entire system end-to-end. This section covers starting the server, verifying it works, and running both the cloud and local LLM clients against it.
The first step is to install the project. Open a terminal, navigate to your project directory, and run:
# Install all dependencies and create the virtual environment
uv sync
# Verify the installation by checking the entry point
uv run weather-mcp --help
Now configure your environment:
# Copy the example environment file
cp .env.example .env
# Edit .env with your actual API key
# On Linux/macOS:
nano .env
# On Windows:
notepad .env
Start the MCP server:
# Activate the virtual environment (optional with uv run)
uv run weather-mcp
You should see output like this:
2026-09-08 10:00:00,123 [INFO] weather_mcp.server: Starting Weather MCP Server...
2026-09-08 10:00:00,145 [INFO] weather_mcp.server: MCP server configured with
tools, resources, and prompts.
2026-09-08 10:00:00,147 [INFO] weather_mcp.server: Weather MCP Server ready.
MCP endpoint: http://0.0.0.0:8000/mcp
INFO: Started server process [12345]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
Verify the server is healthy by checking the operational endpoints:
# Health check
curl http://localhost:8000/health
# Expected: {"status":"healthy","service":"weather-mcp"}
# Server metadata
curl -s http://localhost:8000/info | python3 -m json.tool
Verify the MCP protocol endpoint directly by sending a raw tools/list request:
curl -X POST http://localhost:8000/mcp \
-H "Content-Type: application/json" \
-H "Mcp-Method: tools/list" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {}
}'
The response will be a JSON object listing all four tools with their complete input schemas. This is exactly what an LLM client sees when it discovers your server's capabilities.
Run the tests to verify everything is working correctly:
# Run all tests with verbose output
uv run pytest tests/ -v
# Run with coverage report
uv run pytest tests/ --cov=src/weather_mcp --cov-report=term-missing
# Run only the domain model tests
uv run pytest tests/test_tools.py -v
# Run only the HTTP adapter tests
uv run pytest tests/test_client_integration.py -v
For the local LLM client, you need Ollama running in a separate terminal:
# Terminal 2: Start Ollama (if not already running as a system service)
ollama serve
# Terminal 3: Pull the Llama 4 Maverick model (first time only, ~10 GB)
ollama pull llama4:maverick
# Terminal 4: Run the local LLM client
# (MCP server must already be running in Terminal 1)
uv run python clients/local_llm_client.py
A successful run of the local client produces output like this:
============================================================
Question: What is the current temperature in Tokyo?
============================================================
[Using local model: llama4:maverick via Ollama at http://localhost:11434]
[Iteration 1/10] Sending 2 messages to model...
[Model requested 1 tool call(s).]
-> Calling MCP tool: 'get_current_weather' with args: {
"city": "Tokyo,JP",
"units": "metric"
}
<- Tool result: Current weather for Tokyo, JP (as of 2026-09-08 ...
[Iteration 2/10] Sending 4 messages to model...
[Final answer produced after 2 iteration(s).]
Answer:
The current temperature in Tokyo is 28.3 degrees Celsius (feels like
31.2 degrees due to high humidity). The sky is partly cloudy with
humidity at 78% and winds from the southeast at 4.2 m/s.
For the cloud LLM client:
# Ensure OPENAI_API_KEY, ANTHROPIC_API_KEY, and GOOGLE_API_KEY
# are set in your .env file, then run:
uv run python clients/cloud_llm_client.py
CHAPTER 16: PRODUCTION CONSIDERATIONS AND SECURITY
Getting a server to work in development is one thing; running it in production is another. This chapter covers the most important operational concerns for a production MCP server.
The first and most critical concern is secrets management. Your OpenWeatherMap API key is a secret that should never appear in source code, logs, or error messages. In development, we use a .env file. In production, use your cloud provider's secrets management service: AWS Secrets Manager, Azure Key Vault, Google Cloud Secret Manager, or HashiCorp Vault. Configure your deployment environment to inject secrets as environment variables at runtime, not at build time.
The second concern is observability. In production, you need to know when your server is failing, how fast it is responding, and which tools are being called most frequently. The following module shows how to add structured logging with timing to your tools:
# Production-grade structured logging and tool timing.
# This pattern can be applied to any tool in weather_tools.py.
# Add this decorator to the tools you want to monitor.
from __future__ import annotations
import json
import logging
import time
from functools import wraps
from typing import Any, Callable
class JsonFormatter(logging.Formatter):
"""
A logging formatter that outputs one JSON object per log line.
JSON logs integrate cleanly with aggregation systems like Datadog,
Splunk, or the ELK Stack.
"""
def format(self, record: logging.LogRecord) -> str:
log_entry: dict[str, Any] = {
"timestamp": self.formatTime(record),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}
if record.exc_info:
log_entry["exception"] = self.formatException(record.exc_info)
return json.dumps(log_entry)
def timed_tool(tool_name: str) -> Callable:
"""
A decorator factory that wraps an MCP tool with timing and logging.
Apply this decorator to any @mcp.tool() function to automatically
log its execution time and success/failure status on every call.
Example:
@mcp.tool()
@timed_tool("get_current_weather")
async def get_current_weather(city: str, units: str = "metric") -> str:
...
"""
def decorator(func: Callable) -> Callable:
logger = logging.getLogger(f"weather_mcp.tools.{tool_name}")
@wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any:
start = time.monotonic()
try:
result = await func(*args, **kwargs)
elapsed_ms = (time.monotonic() - start) * 1000
logger.info(
"Tool call succeeded",
extra={
"tool": tool_name,
"duration_ms": round(elapsed_ms, 2),
"success": True,
},
)
return result
except Exception as exc:
elapsed_ms = (time.monotonic() - start) * 1000
logger.error(
"Tool call failed",
extra={
"tool": tool_name,
"duration_ms": round(elapsed_ms, 2),
"success": False,
"error_type": type(exc).__name__,
"error": str(exc),
},
)
raise
return wrapper
return decorator
The third production concern is authentication and authorization. For a public-facing MCP server, you should require clients to present a valid bearer token. The following FastAPI middleware validates tokens on every request to the /mcp endpoint:
# Authentication middleware for the MCP server.
# Add this to server.py and register it with app.middleware("http").
from __future__ import annotations
import secrets
from fastapi import Request
from fastapi.responses import JSONResponse
# In production, load valid tokens from your secrets manager.
# Never hardcode tokens in source code.
VALID_TOKENS: set[str] = {
"load-this-from-secrets-manager-at-startup",
}
async def bearer_token_middleware(
request: Request,
call_next: object,
) -> object:
"""
Middleware that validates bearer tokens on all MCP requests.
Public endpoints (/health, /info) are excluded from authentication
so that monitoring systems can reach them without credentials.
All requests to /mcp must include a valid Authorization header.
"""
# Allow monitoring endpoints without authentication.
if request.url.path in ("/health", "/info"):
return await call_next(request)
# Require authentication for the MCP endpoint.
if request.url.path.startswith("/mcp"):
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
return JSONResponse(
status_code=401,
content={
"error": "Missing or malformed Authorization header. "
"Expected: 'Authorization: Bearer <token>'"
},
)
token = auth_header[len("Bearer "):]
# Use secrets.compare_digest for constant-time comparison.
# This prevents timing attacks where an attacker could infer
# the correct token by measuring response times.
token_valid = any(
secrets.compare_digest(token, valid_token)
for valid_token in VALID_TOKENS
)
if not token_valid:
return JSONResponse(
status_code=403,
content={"error": "Invalid or expired bearer token."},
)
return await call_next(request)
# Register in create_app() after creating the FastAPI instance:
# app.middleware("http")(bearer_token_middleware)
CHAPTER 17: THE BIGGER PICTURE - WHERE THIS ALL LEADS
You have now built a complete, production-ready MCP server from scratch. You have a layered architecture that separates concerns cleanly, a correctly implemented rate limiter that does not hold locks during sleep, a rich set of MCP tools and resources, properly async cloud LLM clients, a local LLM client that bridges Ollama and MCP, and a test suite with proper environment isolation. But let us zoom out and think about what this all means.
The pattern you have learned in this tutorial is not specific to weather data. It is a universal pattern for connecting any REST API to the world of AI agents. Consider what you could build by applying the same architecture to different APIs. You could wrap the GitHub REST API and give LLMs the ability to read repositories, create issues, and review pull requests. You could wrap a company's internal CRM API and give sales agents real-time access to customer data. You could wrap a financial data API and give investment analysis agents access to market data. You could wrap a healthcare data API and give medical decision-support systems access to patient records, with appropriate authorization controls.
In each case, the architecture is the same. The domain layer defines the concepts. The adapter layer handles the HTTP communication. The service layer implements business logic. The tools layer exposes capabilities to LLMs through the MCP protocol. The clients connect LLMs to the server and implement the tool-use loop.
The MCP 2026-07-28 specification's stateless design is particularly important for this future. Because every request is self-describing and sessions are gone, MCP servers can be deployed as serverless functions, as containers in a Kubernetes cluster, or as edge functions running in dozens of data centers simultaneously. An LLM can call your MCP server from anywhere in the world and get a response in milliseconds, with no session state to manage and no sticky load balancing required.
The rise of local LLMs, as demonstrated by our Ollama/Llama 4 client, adds another dimension to this picture. For organizations with strict data privacy requirements, the ability to run both the LLM and the MCP server entirely on-premises, with no data ever leaving the corporate network, is transformative. The same MCP server you built in this tutorial works identically whether the client is a cloud-hosted GPT-6 Astra or a locally running Llama 4 Maverick. The protocol is the equalizer.
The future of software development is increasingly agentic: systems that can reason, plan, and act autonomously to accomplish complex goals. MCP is the connective tissue that makes those systems possible. Every REST API you wrap in an MCP server becomes a capability that any AI agent can discover and use. Every tool you define carefully and document well becomes a reliable building block in systems that can accomplish things no single human could do alone.
You are not just building a weather server. You are learning the grammar of the agentic web.
APPENDIX A: QUICK REFERENCE - MCP 2026-07-28 ESSENTIALS
Protocol Version : 2026-07-28
Transport : Stateless HTTP (no sessions, no handshakes)
Key Headers : Mcp-Method, Mcp-Name
Python SDK : pip install "mcp[cli]>=2.0.0"
Server Class : mcp.server.fastmcp.FastMCP
Tool Decorator : @mcp.tool()
Resource Decorator : @mcp.resource("scheme://path/{param}")
Prompt Decorator : @mcp.prompt()
ASGI Mount : mcp.streamable_http_app()
Client Session : mcp.ClientSession
HTTP Transport : mcp.client.streamable_http.streamablehttp_client
LLM Models Referenced (as of September 8, 2026)
------------------------------------------------
Cloud (Remote):
Anthropic : claude-fable-5-1 (1M token context, agentic work)
OpenAI : gpt-6-astra (1M token context, coding/science)
Google : gemini-3.8-flash (1M token context, multimodal)
Local (via Ollama):
Meta : llama4:maverick (17B params, function calling)
Meta : llama4:scout (smaller, faster, lighter GPU req.)
Meta : llama4:behemoth (largest, most capable, needs H100)
Install Ollama : https://ollama.com
Pull a model : ollama pull llama4:maverick
List local models : ollama list
OpenWeatherMap API Quick Reference
-----------------------------------
Base URL : https://api.openweathermap.org
Current Weather : GET /data/2.5/weather?q={city}&appid={key}&units={u}
5-Day Forecast : GET /data/2.5/forecast?q={city}&appid={key}&units={u}
One Call 3.0 : GET /data/3.0/onecall?lat={lat}&lon={lon}&appid={key}
Unit Systems : metric (C, m/s), imperial (F, mph), standard (K, m/s)
Free Tier Limit : 1,000 calls/day, 60 calls/minute
Sign Up : https://openweathermap.org/api
APPENDIX B: COMPLETE FILE INVENTORY
Source Files (src/weather_mcp/)
--------------------------------
__init__.py - Empty package marker
config.py - Pydantic Settings configuration
server.py - FastAPI + FastMCP server assembly
domain/__init__.py - Empty package marker
domain/models.py - Pydantic domain models
domain/exceptions.py - Domain exception hierarchy
service/__init__.py - Empty package marker
service/weather_service.py - Business logic layer
adapters/__init__.py - Empty package marker
adapters/rate_limiter.py - Token bucket rate limiter
adapters/owm_client.py - OpenWeatherMap HTTP adapter
tools/__init__.py - Empty package marker
tools/weather_tools.py - MCP tool definitions
tools/weather_resources.py - MCP resource definitions
tools/weather_prompts.py - MCP prompt definitions
Test Files (tests/)
--------------------
__init__.py - Empty package marker
conftest.py - Shared fixtures (env var setup)
test_tools.py - Domain and service layer tests
test_client_integration.py - HTTP adapter tests with respx mocking
Client Files (clients/)
------------------------
cloud_llm_client.py - Claude / GPT-6 / Gemini clients
local_llm_client.py - Ollama + Llama 4 client
Project Root
------------
pyproject.toml - Dependencies and entry points
.env.example - Environment variable template
README.md - Quick-start guide
APPENDIX C: TROUBLESHOOTING COMMON ISSUES
Problem : "OWM_API_KEY validation error" on startup
Fix : Ensure .env exists in the project root and contains a valid
OWM_API_KEY. Run: grep OWM_API_KEY .env
Problem : "LocationNotFoundError" for a city you know exists
Fix : Add the country code: "Springfield,US" not "Springfield".
Many city names are shared by multiple countries.
Problem : Ollama tool calls return empty or wrong arguments
Fix : Ensure you are using llama4:maverick or another model that
supports function calling. Run: ollama run llama4:maverick
Problem : "Connection refused" when running a client
Fix : Start the MCP server first: uv run weather-mcp
Verify it is listening: curl http://localhost:8000/health
Problem : Tests fail with "ValidationError: OWM_API_KEY field required"
Fix : The conftest.py fixture sets dummy env vars automatically.
Ensure conftest.py exists in the tests/ directory and that
pytest can find it (tests/__init__.py must also exist).
Problem : "TypeError: object AsyncGenerator can't be used in 'await'"
Fix : This error means 'await' was used before an async generator.
In local_llm_client.py, ensure pull() uses 'async for',
not 'async for ... in await ...'.
Problem : Event loop blocked during Claude API calls
Fix : Use anthropic.AsyncAnthropic, not anthropic.Anthropic.
The synchronous client blocks the event loop.
Problem : Gemini tool calls pass wrong types (strings instead of ints)
Fix : The _json_schema_type_to_genai() function in cloud_llm_client.py
maps JSON Schema types to genai types. Verify it is present
and called correctly in the Gemini tool conversion loop.
