Python
Create a simple MCP server in Python in 15 minutes
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.
Install Python
You’ll need Python 3.10 or higher:
python --version # Should be 3.10 or higher
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
Create a new project using the MCP project creator
uvx create-mcp-server --path weather_service
cd weather_service
Install additional dependencies
uv add httpx python-dotenv
Set up environment
Create .env
:
OPENWEATHER_API_KEY=your-api-key-here
Create your server
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 httpx
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
Add weather fetching functionality
Add this functionality:
# Create reusable params
http_params = {
"appid": API_KEY,
"units": "metric"
}
async def fetch_weather(city: str) -> dict[str, Any]:
async with httpx.AsyncClient() as client:
response = await client.get(
f"{API_BASE_URL}/weather",
params={"q": city, **http_params}
)
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")
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 httpx.HTTPError as e:
raise RuntimeError(f"Weather API error: {str(e)}")
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:
async with httpx.AsyncClient() as client:
response = await client.get(
f"{API_BASE_URL}/{FORECAST_ENDPOINT}",
params={
"q": city,
"cnt": days * 8, # API returns 3-hour intervals
**http_params,
}
)
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 httpx.HTTPError as e:
logger.error(f"Weather API error: {str(e)}")
raise RuntimeError(f"Weather API error: {str(e)}")
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()
)
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
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"
}
}
}
}
Restart Claude
-
Quit Claude completely
-
Start Claude again
-
Look for your weather server in the 🔌 menu
Try it out!
Ask Claude:
What's the current weather in San Francisco? Can you analyze the conditions and tell me if it's a good day for outdoor activities?
Ask Claude:
Can you get me a 5-day forecast for Tokyo and help me plan what clothes to pack for my trip?
Ask Claude:
Can you analyze the forecast for both Tokyo and San Francisco and tell me which city would be better for outdoor photography this week?
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:
async with httpx.AsyncClient() as client:
response = await client.get(..., params={..., **http_params})
response.raise_for_status()
except httpx.HTTPError 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 additional 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
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)]
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):
async with httpx.AsyncClient() as client:
response = await client.get(
f"{API_BASE_URL}/{CURRENT_WEATHER_ENDPOINT}",
params={"q": city, **http_params}
)
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
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...
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)}")
Add resource templates
@app.list_resource_templates()
async def list_resource_templates() -> list[ResourceTemplate]:
return [
ResourceTemplate(
uriTemplate="weather://{city}/current",
name="Current weather for any city",
mimeType="application/json"
)
]
Testing
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 anyio_backend():
return "asyncio"
@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.anyio
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.anyio
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.anyio
async def test_call_tool(mock_forecast_response):
class Response():
def raise_for_status(self):
pass
def json(self):
return mock_forecast_response
class AsyncClient():
async def __aenter__(self):
return self
async def __aexit__(self, *exc_info):
pass
async def get(self, *args, **kwargs):
return Response()
with patch('httpx.AsyncClient', new=AsyncClient) as mock_client:
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.anyio
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.anyio
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"]
Run tests
uv add --dev pytest
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
Was this page helpful?