Let’s build your first MCP server in Python! We’ll create a weather server that provides current weather data as a resource and lets Claude fetch forecasts using tools.

This guide uses the OpenWeatherMap API. You’ll need a free API key from OpenWeatherMap to follow along.

​
Prerequisites

The following steps are for macOS. Guides for other platforms are coming soon.

1

Install Python

You’ll need Python 3.10 or higher:

python --version  # Should be 3.10 or higher
2

Install uv via homebrew

See https://docs.astral.sh/uv/ for more information.

brew install uv
uv --version # Should be 0.4.18 or higher
3

Create a new project using the MCP project creator

uvx create-mcp-server --path weather_service
cd weather_service
4

Install additional dependencies

uv add requests python-dotenv
5

Set up environment

Create .env:

OPENWEATHER_API_KEY=your-api-key-here

​
Create your server

1

Add the base imports and setup

In weather_service/src/weather_service/server.py

import os
import json
import logging
from datetime import datetime, timedelta
from collections.abc import Sequence
from functools import lru_cache
from typing import Any

import requests
import asyncio
from dotenv import load_dotenv
from mcp.server import Server
from mcp.types import (
    Resource,
    Tool,
    TextContent,
    ImageContent,
    EmbeddedResource,
    LoggingLevel
)
from pydantic import AnyUrl

# Load environment variables
load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("weather-server")

# API configuration
API_KEY = os.getenv("OPENWEATHER_API_KEY")
if not API_KEY:
    raise ValueError("OPENWEATHER_API_KEY environment variable required")

API_BASE_URL = "http://api.openweathermap.org/data/2.5"
DEFAULT_CITY = "London"
CURRENT_WEATHER_ENDPOINT = "weather"
FORECAST_ENDPOINT = "forecast"

# The rest of our server implementation will go here
2

Add weather fetching functionality

Add this functionality:

# Create reusable session
http = requests.Session()
http.params = {
    "appid": API_KEY,
    "units": "metric"
}

async def fetch_weather(city: str) -> dict[str, Any]:
    response = http.get(
        f"{API_BASE_URL}/weather",
        params={"q": city}
    )
    response.raise_for_status()
    data = response.json()

    return {
        "temperature": data["main"]["temp"],
        "conditions": data["weather"][0]["description"],
        "humidity": data["main"]["humidity"],
        "wind_speed": data["wind"]["speed"],
        "timestamp": datetime.now().isoformat()
    }


app = Server("weather-server")
3

Implement resource handlers

Add these resource-related handlers to our main function:

app = Server("weather-server")

@app.list_resources()
async def list_resources() -> list[Resource]:
    """List available weather resources."""
    uri = AnyUrl(f"weather://{DEFAULT_CITY}/current")
    return [
        Resource(
            uri=uri,
            name=f"Current weather in {DEFAULT_CITY}",
            mimeType="application/json",
            description="Real-time weather data"
        )
    ]

@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:
    """Read current weather data for a city."""
    city = DEFAULT_CITY
    if str(uri).startswith("weather://") and str(uri).endswith("/current"):
        city = str(uri).split("/")[-2]
    else:
        raise ValueError(f"Unknown resource: {uri}")

    try:
        weather_data = await fetch_weather(city)
        return json.dumps(weather_data, indent=2)
    except requests.RequestException as e:
        raise RuntimeError(f"Weather API error: {str(e)}")
4

Implement tool handlers

Add these tool-related handlers:

app = Server("weather-server")

# Resource implementation ...

@app.list_tools()
async def list_tools() -> list[Tool]:
    """List available weather tools."""
    return [
        Tool(
            name="get_forecast",
            description="Get weather forecast for a city",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name"
                    },
                    "days": {
                        "type": "number",
                        "description": "Number of days (1-5)",
                        "minimum": 1,
                        "maximum": 5
                    }
                },
                "required": ["city"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
    """Handle tool calls for weather forecasts."""
    if name != "get_forecast":
        raise ValueError(f"Unknown tool: {name}")

    if not isinstance(arguments, dict) or "city" not in arguments:
        raise ValueError("Invalid forecast arguments")

    city = arguments["city"]
    days = min(int(arguments.get("days", 3)), 5)

    try:
        response = http.get(
            f"{API_BASE_URL}/{FORECAST_ENDPOINT}",
            params={
                "q": city,
                "cnt": days * 8  # API returns 3-hour intervals
            }
        )
        response.raise_for_status()
        data = response.json()

        forecasts = []
        for i in range(0, len(data["list"]), 8):
            day_data = data["list"][i]
            forecasts.append({
                "date": day_data["dt_txt"].split()[0],
                "temperature": day_data["main"]["temp"],
                "conditions": day_data["weather"][0]["description"]
            })

        return [
            TextContent(
                type="text",
                text=json.dumps(forecasts, indent=2)
            )
        ]
    except requests.RequestException as e:
        logger.error(f"Weather API error: {str(e)}")
        raise RuntimeError(f"Weather API error: {str(e)}")
5

Add the main function

Add this to the end of weather_service/src/weather_service/server.py:

async def main():
    # Import here to avoid issues with event loops
    from mcp.server.stdio import stdio_server

    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )
6

Check your entry point in __init__.py

Add this to the end of weather_service/src/weather_service/__init__.py:

from . import server
import asyncio

def main():
   """Main entry point for the package."""
   asyncio.run(server.main())

# Optionally expose other important items at package level
__all__ = ['main', 'server']

​
Connect to Claude Desktop

1

Update Claude config

Add to claude_desktop_config.json:

{
  "mcpServers": {
    "weather": {
      "command": "uv",
      "args": [
        "--directory",
        "path/to/your/project",
        "run",
        "weather-service"
      ],
      "env": {
        "OPENWEATHER_API_KEY": "your-api-key"
      }
    }
  }
}
2

Restart Claude

  1. Quit Claude completely

  2. Start Claude again

  3. Look for your weather server in the 🔌 menu

​
Try it out!

​
Understanding the code

async def read_resource(self, uri: str) -> ReadResourceResult:
    # ...

Python type hints help catch errors early and improve code maintainability.

​
Best practices

Error Handling

try:
    response = self.http.get(...)
    response.raise_for_status()
except requests.RequestException as e:
    raise McpError(
        ErrorCode.INTERNAL_ERROR,
        f"API error: {str(e)}"
    )

Type Validation

if not isinstance(args, dict) or "city" not in args:
    raise McpError(
        ErrorCode.INVALID_PARAMS,
        "Invalid forecast arguments"
    )

Environment Variables

if not API_KEY:
    raise ValueError("OPENWEATHER_API_KEY is required")

​
Available transports

While this guide uses stdio transport, MCP supports additonal transport options:

​
SSE (Server-Sent Events)

from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route

# Create SSE transport with endpoint
sse = SseServerTransport("/messages")

# Handler for SSE connections
async def handle_sse(scope, receive, send):
    async with sse.connect_sse(scope, receive, send) as streams:
        await app.run(
            streams[0], streams[1], app.create_initialization_options()
        )

# Handler for client messages
async def handle_messages(scope, receive, send):
    await sse.handle_post_message(scope, receive, send)

# Create Starlette app with routes
app = Starlette(
    debug=True,
    routes=[
        Route("/sse", endpoint=handle_sse),
        Route("/messages", endpoint=handle_messages, methods=["POST"]),
    ],
)

# Run with any ASGI server
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

​
Advanced features

1

Understanding Request Context

The request context provides access to the current request’s metadata and the active client session. Access it through server.request_context:

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> Sequence[TextContent]:
    # Access the current request context
    ctx = self.request_context

    # Get request metadata like progress tokens
    if progress_token := ctx.meta.progressToken:
        # Send progress notifications via the session
        await ctx.session.send_progress_notification(
            progress_token=progress_token,
            progress=0.5,
            total=1.0
        )

    # Sample from the LLM client
    result = await ctx.session.create_message(
        messages=[
            SamplingMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text="Analyze this weather data: " + json.dumps(arguments)
                )
            )
        ],
        max_tokens=100
    )

    return [TextContent(type="text", text=result.content.text)]
2

Add caching

# Cache settings
cache_timeout = timedelta(minutes=15)
last_cache_time = None
cached_weather = None

async def fetch_weather(city: str) -> dict[str, Any]:
    global cached_weather, last_cache_time

    now = datetime.now()
    if (cached_weather is None or
        last_cache_time is None or
        now - last_cache_time > cache_timeout):

        response = http.get(
            f"{API_BASE_URL}/{CURRENT_WEATHER_ENDPOINT}",
            params={"q": city}
        )
        response.raise_for_status()
        data = response.json()

        cached_weather = {
            "temperature": data["main"]["temp"],
            "conditions": data["weather"][0]["description"],
            "humidity": data["main"]["humidity"],
            "wind_speed": data["wind"]["speed"],
            "timestamp": datetime.now().isoformat()
        }
        last_cache_time = now

    return cached_weather
3

Add progress notifications

@self.call_tool()
async def call_tool(self, name: str, arguments: Any) -> CallToolResult:
    if progress_token := self.request_context.meta.progressToken:
        # Send progress notifications
        await self.request_context.session.send_progress_notification(
            progress_token=progress_token,
            progress=1,
            total=2
        )

        # Fetch data...

        await self.request_context.session.send_progress_notification(
            progress_token=progress_token,
            progress=2,
            total=2
        )

    # Rest of the method implementation...
4

Add logging support

# Set up logging
logger = logging.getLogger("weather-server")
logger.setLevel(logging.INFO)

@app.set_logging_level()
async def set_logging_level(level: LoggingLevel) -> EmptyResult:
    logger.setLevel(level.upper())
    await app.request_context.session.send_log_message(
        level="info",
        data=f"Log level set to {level}",
        logger="weather-server"
    )
    return EmptyResult()

# Use logger throughout the code
# For example:
# logger.info("Weather data fetched successfully")
# logger.error(f"Error fetching weather data: {str(e)}")
5

Add resource templates

@self.list_resources()
async def list_resources(self) -> ListResourcesResult:
    return ListResourcesResult(
        resources=[...],
        resourceTemplates=[
            ResourceTemplate(
                uriTemplate="weather://{city}/current",
                name="Current weather for any city",
                mimeType="application/json"
            )
        ]
    )

​
Testing

1

Create test file

Create tests/weather_test.py:

import pytest
import os
from unittest.mock import patch, Mock
from datetime import datetime
import json
from pydantic import AnyUrl
os.environ["OPENWEATHER_API_KEY"] = "TEST"

from weather_service.server import (
    fetch_weather,
    read_resource,
    call_tool,
    list_resources,
    list_tools,
    DEFAULT_CITY
)

@pytest.fixture
def mock_weather_response():
    return {
        "main": {
            "temp": 20.5,
            "humidity": 65
        },
        "weather": [
            {"description": "scattered clouds"}
        ],
        "wind": {
            "speed": 3.6
        }
    }

@pytest.fixture
def mock_forecast_response():
    return {
        "list": [
            {
                "dt_txt": "2024-01-01 12:00:00",
                "main": {"temp": 18.5},
                "weather": [{"description": "sunny"}]
            },
            {
                "dt_txt": "2024-01-02 12:00:00",
                "main": {"temp": 17.2},
                "weather": [{"description": "cloudy"}]
            }
        ]
    }

@pytest.mark.asyncio
async def test_fetch_weather(mock_weather_response):
    with patch('requests.Session.get') as mock_get:
        mock_get.return_value.json.return_value = mock_weather_response
        mock_get.return_value.raise_for_status = Mock()

        weather = await fetch_weather("London")

        assert weather["temperature"] == 20.5
        assert weather["conditions"] == "scattered clouds"
        assert weather["humidity"] == 65
        assert weather["wind_speed"] == 3.6
        assert "timestamp" in weather

@pytest.mark.asyncio
async def test_read_resource():
    with patch('weather_service.server.fetch_weather') as mock_fetch:
        mock_fetch.return_value = {
            "temperature": 20.5,
            "conditions": "clear sky",
            "timestamp": datetime.now().isoformat()
        }

        uri = AnyUrl("weather://London/current")
        result = await read_resource(uri)

        assert isinstance(result, str)
        assert "temperature" in result
        assert "clear sky" in result

@pytest.mark.asyncio
async def test_call_tool(mock_forecast_response):
    with patch('weather_service.server.http.get') as mock_get:
        mock_get.return_value.json.return_value = mock_forecast_response
        mock_get.return_value.raise_for_status = Mock()

        result = await call_tool("get_forecast", {"city": "London", "days": 2})

        assert len(result) == 1
        assert result[0].type == "text"
        forecast_data = json.loads(result[0].text)
        assert len(forecast_data) == 1
        assert forecast_data[0]["temperature"] == 18.5
        assert forecast_data[0]["conditions"] == "sunny"

@pytest.mark.asyncio
async def test_list_resources():
    resources = await list_resources()
    assert len(resources) == 1
    assert resources[0].name == f"Current weather in {DEFAULT_CITY}"
    assert resources[0].mimeType == "application/json"

@pytest.mark.asyncio
async def test_list_tools():
    tools = await list_tools()
    assert len(tools) == 1
    assert tools[0].name == "get_forecast"
    assert "city" in tools[0].inputSchema["properties"]
2

Run tests

uv add --dev pytest pytest-asyncio
uv run pytest

​
Troubleshooting

​
Installation issues

# Check Python version
python --version

# Reinstall dependencies
uv sync --reinstall

​
Type checking

# Install mypy
uv add --dev pyright

# Run type checker
uv run pyright src

​
Next steps

Architecture overview

Learn more about the MCP architecture

Python SDK

Check out the Python SDK on GitHub

ClientsTypeScript