12/1/24, 9:44 PM Python - Model Context Protocol
Your First MCP Server Python
Your First MCP Server
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.
1 Install Python
You’ll need Python 3.10 or higher:
[Link] 1/26
12/1/24, 9:44 PM Python - Model Context Protocol
python --version # Should be 3.10 or higher
2 Your Install
First MCP
uv Server Python
via homebrew
See [Link] 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 httpx python-dotenv
5 Set up environment
Create :
[Link] 2/26
12/1/24, 9:44 PM Python - Model Context Protocol
OPENWEATHER_API_KEY=your-api-key-here
Your First MCP Server Python
Create your server
1 Add the base imports and setup
In
import os
import json
import logging
from datetime import datetime, timedelta
from [Link] import Sequence
from functools import lru_cache
from typing import Any
import httpx
import asyncio
from dotenv import load_dotenv
from [Link] import Server
from [Link] import (
Resource,
Tool,
TextContent,
ImageContent,
[Link] 3/26
12/1/24, 9:44 PM Python - Model Context Protocol
EmbeddedResource,
LoggingLevel
)
from
Your First MCPpydantic
Server import
Python AnyUrl
# Load environment variables
load_dotenv()
# Configure logging
[Link](level=[Link])
logger = [Link]("weather-server")
# API configuration
API_KEY = [Link]("OPENWEATHER_API_KEY")
if not API_KEY:
raise ValueError("OPENWEATHER_API_KEY environment variable required")
API_BASE_URL = "[Link]
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:
[Link] 4/26
12/1/24, 9:44 PM Python - Model Context Protocol
# Create reusable params
http_params = {
"appid": API_KEY,
Your First MCP Server Python
"units": "metric"
}
async def fetch_weather(city: str) -> dict[str, Any]:
async with [Link]() as client:
response = await [Link](
f"{API_BASE_URL}/weather",
params={"q": city, **http_params}
)
response.raise_for_status()
data = [Link]()
return {
"temperature": data["main"]["temp"],
"conditions": data["weather"][0]["description"],
"humidity": data["main"]["humidity"],
"wind_speed": data["wind"]["speed"],
"timestamp": [Link]().isoformat()
}
app = Server("weather-server")
3 Implement resource handlers
Add these resource-related handlers to our main function:
[Link] 5/26
12/1/24, 9:44 PM Python - Model Context Protocol
app = Server("weather-server")
@app.list_resources()
Your First MCP Server Python
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 [Link](weather_data, indent=2)
except [Link] as e:
[Link] 6/26
12/1/24, 9:44 PM Python - Model Context Protocol
raise RuntimeError(f"Weather API error: {str(e)}")
4 Your Implement
First MCP Server Python
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,
[Link] 7/26
12/1/24, 9:44 PM Python - Model Context Protocol
"maximum": 5
}
},
Your First MCP Server "required": ["city"]
Python
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: Any) -> Sequence[TextContent | ImageContent | EmbeddedResour
"""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([Link]("days", 3)), 5)
try:
async with [Link]() as client:
response = await [Link](
f"{API_BASE_URL}/{FORECAST_ENDPOINT}",
params={
"q": city,
"cnt": days * 8, # API returns 3-hour intervals
**http_params,
}
)
[Link] 8/26
12/1/24, 9:44 PM Python - Model Context Protocol
response.raise_for_status()
data = [Link]()
forecasts
Your First MCP Server = []
Python
for i in range(0, len(data["list"]), 8):
day_data = data["list"][i]
[Link]({
"date": day_data["dt_txt"].split()[0],
"temperature": day_data["main"]["temp"],
"conditions": day_data["weather"][0]["description"]
})
return [
TextContent(
type="text",
text=[Link](forecasts, indent=2)
)
]
except [Link] as e:
[Link](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 :
async def main():
# Import here to avoid issues with event loops
from [Link] import stdio_server
[Link] 9/26
12/1/24, 9:44 PM Python - Model Context Protocol
async with stdio_server() as (read_stream, write_stream):
await [Link](
Your First MCP Server read_stream,
Python
write_stream,
app.create_initialization_options()
)
6 Check your entry point in __init__.py
Add this to the end of :
from . import server
import asyncio
def main():
"""Main entry point for the package."""
[Link]([Link]())
# Optionally expose other important items at package level
__all__ = ['main', 'server']
Connect to Claude Desktop
[Link] 10/26
12/1/24, 9:44 PM Python - Model Context Protocol
1 Update Claude config
Add to :
Your First MCP Server Python
{
"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
[Link] 11/26
12/1/24, 9:44 PM Python - Model Context Protocol
3. Look for your weather server in the 🔌 menu
TryYour
it out!
First MCP Server Python
Check Current Weather
Get a Forecast
Compare Weather
Understanding the code
Type Hints Resources Tools Server Structure
async def read_resource(self, uri: str) -> ReadResourceResult:
# ...
Python type hints help catch errors early and improve code maintainability.
Best practices
[Link] 12/26
12/1/24, 9:44 PM Python - Model Context Protocol
Error Handling
Your First MCP Server Python
try:
async with [Link]() as client:
response = await [Link](..., params={..., **http_params})
response.raise_for_status()
except [Link] 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,
[Link] 13/26
12/1/24, 9:44 PM Python - Model Context Protocol
"Invalid forecast arguments"
)
Your First MCP Server Python
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 [Link] import SseServerTransport
from [Link] import Starlette
from [Link] import Route
# Create SSE transport with endpoint
sse = SseServerTransport("/messages")
[Link] 14/26
12/1/24, 9:44 PM Python - Model Context Protocol
# Handler for SSE connections
async def handle_sse(scope, receive, send):
Yourasync withServer
First MCP sse.connect_sse(scope,
Python receive, send) as streams:
await [Link](
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
[Link](app, host="[Link]", port=8000)
Advanced features
[Link] 15/26
12/1/24, 9:44 PM Python - Model Context Protocol
1 Understanding Request Context
The request context provides access to the current request’s metadata and the active client session. Access it
Your through
First MCP Server Python :
@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 := [Link]:
# Send progress notifications via the session
await [Link].send_progress_notification(
progress_token=progress_token,
progress=0.5,
total=1.0
)
# Sample from the LLM client
result = await [Link].create_message(
messages=[
SamplingMessage(
role="user",
content=TextContent(
type="text",
text="Analyze this weather data: " + [Link](arguments)
)
)
[Link] 16/26
12/1/24, 9:44 PM Python - Model Context Protocol
],
max_tokens=100
)
Your First MCP Server Python
return [TextContent(type="text", text=[Link])]
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 = [Link]()
if (cached_weather is None or
last_cache_time is None or
now - last_cache_time > cache_timeout):
async with [Link]() as client:
response = await [Link](
f"{API_BASE_URL}/{CURRENT_WEATHER_ENDPOINT}",
params={"q": city, **http_params}
)
response.raise_for_status()
data = [Link]()
[Link] 17/26
12/1/24, 9:44 PM Python - Model Context Protocol
cached_weather = {
"temperature": data["main"]["temp"],
"conditions": data["weather"][0]["description"],
Your First MCP Server "humidity":
Python data["main"]["humidity"],
"wind_speed": data["wind"]["speed"],
"timestamp": [Link]().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.[Link]:
# 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
[Link] 18/26
12/1/24, 9:44 PM Python - Model Context Protocol
# Rest of the method implementation...
Your First MCP Server Python
4 Add logging support
# Set up logging
logger = [Link]("weather-server")
[Link]([Link])
@app.set_logging_level()
async def set_logging_level(level: LoggingLevel) -> EmptyResult:
[Link]([Link]())
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:
# [Link]("Weather data fetched successfully")
# [Link](f"Error fetching weather data: {str(e)}")
5 Add resource templates
[Link] 19/26
12/1/24, 9:44 PM Python - Model Context Protocol
@app.list_resource_templates()
async def list_resource_templates() -> list[ResourceTemplate]:
return [
Your First MCP Server Python
ResourceTemplate(
uriTemplate="weather://{city}/current",
name="Current weather for any city",
mimeType="application/json"
)
]
Testing
1 Create test file
Create :
import pytest
import os
from [Link] import patch, Mock
from datetime import datetime
import json
from pydantic import AnyUrl
[Link]["OPENWEATHER_API_KEY"] = "TEST"
[Link] 20/26
12/1/24, 9:44 PM Python - Model Context Protocol
from weather_service.server import (
fetch_weather,
read_resource,
call_tool,
Your First MCP Server Python
list_resources,
list_tools,
DEFAULT_CITY
)
@[Link]
def anyio_backend():
return "asyncio"
@[Link]
def mock_weather_response():
return {
"main": {
"temp": 20.5,
"humidity": 65
},
"weather": [
{"description": "scattered clouds"}
],
"wind": {
"speed": 3.6
}
}
@[Link]
def mock_forecast_response():
[Link] 21/26
12/1/24, 9:44 PM Python - Model Context Protocol
return {
"list": [
{
"dt_txt": "2024-01-01 12:00:00",
Your First MCP Server Python
"main": {"temp": 18.5},
"weather": [{"description": "sunny"}]
},
{
"dt_txt": "2024-01-02 12:00:00",
"main": {"temp": 17.2},
"weather": [{"description": "cloudy"}]
}
]
}
@[Link]
async def test_fetch_weather(mock_weather_response):
with patch('[Link]') 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
@[Link]
[Link] 22/26
12/1/24, 9:44 PM Python - Model Context Protocol
async def test_read_resource():
with patch('weather_service.server.fetch_weather') as mock_fetch:
mock_fetch.return_value = {
Your First MCP Server "temperature":
Python 20.5,
"conditions": "clear sky",
"timestamp": [Link]().isoformat()
}
uri = AnyUrl("weather://London/current")
result = await read_resource(uri)
assert isinstance(result, str)
assert "temperature" in result
assert "clear sky" in result
@[Link]
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():
def __aenter__(self):
return self
async def __aexit__(self, *exc_info):
pass
[Link] 23/26
12/1/24, 9:44 PM Python - Model Context Protocol
async def get(self, *args, **kwargs):
return Response()
Your First MCP Server Python
with patch('[Link]', 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 = [Link](result[0].text)
assert len(forecast_data) == 1
assert forecast_data[0]["temperature"] == 18.5
assert forecast_data[0]["conditions"] == "sunny"
@[Link]
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"
@[Link]
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
[Link] 24/26
12/1/24, 9:44 PM Python - Model Context Protocol
uv add --dev pytest
uv run pytest
Your First MCP Server Python
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
[Link] 25/26
12/1/24, 9:44 PM Python - Model Context Protocol
Next steps
Your First MCP Server Python
Architecture overview Python SDK
Learn more about the MCP architecture Check out the Python SDK on GitHub
Clients TypeScript
[Link] 26/26