modelcontextprotocol/docs
Watch
1
0
Fork
You've already forked docs
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Updates to Python guide

Browse source
This commit is contained in:
Zack Lee 2024-11-21 14:59:47 -05:00
commit 50c7f4cc44
No known key found for this signature in database
1 changed files with 307 additions and 176 deletions
Download patch file Download diff file Expand all files Collapse all files

485
docs/first-server/python.mdx
View file

@ -37,8 +37,6 @@ Let's build your first MCP server in Python! We'll create a weather server that
```bash
uvx create-mcp-server --path weather_service
cd weather_service
# Installs base dependencies
uv sync --dev --all-extras
```
</Step>
@ -57,175 +55,99 @@ Let's build your first MCP server in Python! We'll create a weather server that
</Step>
</Steps>
## Understanding the starter code
The starter code establishes a basic MCP server structure that we'll build upon to create our weather service. Let's break down its key components:
### Server class definition
```python
class McpServer(Server):
def __init__(self):
super().__init__("weather_service")
self.notes: dict[str, str] = {}
```
This is our main server class that inherits from the base `Server` class. Currently, it initializes with a simple notes dictionary that we'll replace with weather-related functionality.
### Three core primitives
The server implements three fundamental MCP primitives, two of which we'll use for our weather service:
1. **Resources (`@self.list_resources` and `@self.read_resource`)**
- Will be modified to expose a single weather endpoint for DEFAULT_CITY
- `list_resources`: Returns current weather resource for DEFAULT_CITY
- `read_resource`: Fetches and returns current weather data for DEFAULT_CITY
- Uses custom URI schemes for data access
2. **Tools (`@self.list_tools` and `@self.call_tool`)**
- Will implement a single "get_forecast" tool
- Accepts a city name as input
- Returns weather forecast data for the specified city
- Uses JSON Schema for input validation
### Server communication
```python
async def main():
server = McpServer()
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(...)
```
This section handles server initialization and communication setup using stdin/stdout streams.
## Steps to implement our weather server
To transform this starter code into our weather server, we'll:
1. **Set up OpenWeather integration**
- Add OpenWeather API key configuration
- Create helper functions for API communication
- Implement weather data parsing
2. **Implement weather resources**
- Replace note storage with weather data handling
- Create weather-specific URI scheme
- Add current weather fetching for DEFAULT_CITY
3. **Create forecast tool**
- Remove note-related tools
- Add "get_forecast" tool implementation
- Implement city-based forecast fetching
- Add response formatting
4. **Add error handling**
- Implement API error management
- Add input validation for city names
- Create user-friendly error messages
Let's dive in!
## Implementing the weather server
Let's start by modifiying our existing `server.py` to transform it from a notes server into a weather service. We'll go through each section of the code and make the necessary changes.
## Create your server
<Steps>
<Step title="Updating imports and configuration">
First, let's add the imports we need at the top of the file:
<Step title="Add the base imports and setup">
In `weather_service/src/weather_service/server.py`
```python
# Keep existing imports
import asyncio
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
from pydantic import AnyUrl
import mcp.server.stdio
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
# Add new imports
import os
import json
import logging
from datetime import datetime
import requests
from dotenv import load_dotenv
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
# Add configuration
load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("weather-server")
# Load environment variables
load_dotenv()
# API configuration
API_KEY = os.getenv("OPENWEATHER_API_KEY")
if not API_KEY:
raise ValueError("OPENWEATHER_API_KEY environment variable required")
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("weather-server")
API_BASE_URL = "http://api.openweathermap.org/data/2.5"
DEFAULT_CITY = "London"
CURRENT_WEATHER_ENDPOINT = "weather"
FORECAST_ENDPOINT = "forecast"
# 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
```
</Step>
<Step title="Update the class initialization">
First, let's modify the class docstring and initialization to remove the notes storage and add our HTTP session:
<Step title="Add weather fetching functionality">
Add this functionality:
```python
class McpServer(Server):
"""
Weather service MCP server implementing two core primitives:
1. Resources - Current weather data accessed via custom URIs
2. Tools - Ability to fetch weather forecasts for any city
"""
def __init__(self):
super().__init__("weather_service")
# Create reusable session for API calls
self.http = requests.Session()
self.http.params = {
"appid": API_KEY,
"units": "metric"
}
# 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")
```
</Step>
<Step title="Add weather helper method">
Next, let's add a helper method to fetch weather data that both our resource and tool handlers will use:
```python
async def fetch_weather(self, city: str) -> dict:
"""Helper method to fetch current weather data."""
response = self.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()
}
```
</Step>
<Step title="Update resource handlers">
Now, let's replace the note-related resource handlers with weather ones:
<Step title="Implement resource handlers">
Add these resource-related handlers to our main function:
```python
# RESOURCE HANDLERS
@self.list_resources()
async def handle_list_resources(self) -> list[types.Resource]:
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 [
types.Resource(
Resource(
uri=uri,
name=f"Current weather in {DEFAULT_CITY}",
mimeType="application/json",
@ -233,36 +155,37 @@ Let's start by modifiying our existing `server.py` to transform it from a notes
)
]
@self.read_resource()
async def handle_read_resource(self, uri: AnyUrl) -> str:
@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 self.fetch_weather(city)
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)}")
```
</Step>
<Step title="Remove prompt handlers">
Remove the entire prompt section from the starter code - we won't be using prompts in our weather service.
</Step>
<Step title="Update tool handlers">
Finally, let's replace the note-adding tool with our weather forecast tool:
<Step title="Implement tool handlers">
Add these tool-related handlers:
```python
# TOOL HANDLERS
@self.list_tools()
async def handle_list_tools(self) -> list[types.Tool]:
app = Server("weather-server")
# Resource implementation ...
@app.list_tools()
async def list_tools() -> list[Tool]:
"""List available weather tools."""
return [
types.Tool(
Tool(
name="get_forecast",
description="Get weather forecast for a city",
inputSchema={
@ -284,10 +207,8 @@ Let's start by modifiying our existing `server.py` to transform it from a notes
)
]
@self.call_tool()
async def handle_call_tool(
self, name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
@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}")
@ -299,7 +220,7 @@ Let's start by modifiying our existing `server.py` to transform it from a notes
days = min(int(arguments.get("days", 3)), 5)
try:
response = self.http.get(
response = http.get(
f"{API_BASE_URL}/{FORECAST_ENDPOINT}",
params={
"q": city,
@ -319,7 +240,7 @@ Let's start by modifiying our existing `server.py` to transform it from a notes
})
return [
types.TextContent(
TextContent(
type="text",
text=json.dumps(forecasts, indent=2)
)
@ -329,16 +250,226 @@ Let's start by modifiying our existing `server.py` to transform it from a notes
raise RuntimeError(f"Weather API error: {str(e)}")
```
</Step>
<Step title="Add the main function">
Add this to the end of `weather_service/src/weather_service/server.py`:
```python
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()
)
```
</Step>
<Step title="Check your entry point in __init__.py">
Add this to the end of `weather_service/src/weather_service/__init__.py`:
```python
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']
```
</Step>
</Steps>
## Understanding the code
<Tabs>
<Tab title="Type Hints">
```python
async def read_resource(self, uri: str) -> ReadResourceResult:
# ...
```
Python type hints help catch errors early and improve code maintainability.
</Tab>
<Tab title="Resources">
```python
@app.list_resources()
async def list_resources(self) -> ListResourcesResult:
return ListResourcesResult(
resources=[
Resource(
uri=f"weather://{DEFAULT_CITY}/current",
name=f"Current weather in {DEFAULT_CITY}",
mimeType="application/json",
description="Real-time weather data"
)
]
)
```
Resources provide data that Claude can access as context.
</Tab>
<Tab title="Tools">
```python
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"]
}
)
```
Tools let Claude take actions through your server with validated inputs.
</Tab>
<Tab title="Server Structure">
```python
# Create server instance with name
app = Server("weather-server")
# Register resource handler
@app.list_resources()
async def list_resources() -> list[Resource]:
"""List available resources"""
return [...]
# Register tool handler
@app.call_tool()
async def call_tool(name: str, arguments: Any) -> Sequence[TextContent]:
"""Handle tool execution"""
return [...]
# Register additional handlers
@app.read_resource()
...
@app.list_tools()
...
```
The MCP server uses a simple app pattern - create a Server instance and register handlers with decorators. Each handler maps to a specific MCP protocol operation.
</Tab>
</Tabs>
## Best practices
<CardGroup cols={1}>
<Card title="Error Handling" icon="shield">
```python
try:
response = self.http.get(...)
response.raise_for_status()
except requests.RequestException as e:
raise McpError(
ErrorCode.INTERNAL_ERROR,
f"API error: {str(e)}"
)
```
</Card>
<Card title="Type Validation" icon="check">
```python
if not isinstance(args, dict) or "city" not in args:
raise McpError(
ErrorCode.INVALID_PARAMS,
"Invalid forecast arguments"
)
```
</Card>
<Card title="Environment Variables" icon="gear">
```python
if not API_KEY:
raise ValueError("OPENWEATHER_API_KEY is required")
```
</Card>
</CardGroup>
## Available transports
While this guide uses stdio transport, MCP supports additonal transport options:
### SSE (Server-Sent Events)
```python
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)
```
## Troubleshooting
### Installation issues
```bash
# Check Python version
python --version
# Reinstall dependencies
uv sync --reinstall
```
### Type checking
```bash
# Install mypy
uv add --dev pyright
# Run type checker
uv run pyright src
```
## Connect to Claude Desktop
Now that our server is implemented, let's make sure Claude desktop can talk to it properly.
<Note>
Note that this step is unnecessary if you've opted to install the server into Claude.app with the MCP project creator script.
</Note>
<Steps>
<Step title="Update Claude config">
Add to `claude_desktop_config.json`:
@ -672,4 +803,4 @@ Note that this step is unnecessary if you've opted to install the server into Cl
<Card title="Python SDK" icon="python" href="/api-reference/python/class-reference">
Read more about the Python SDK
</Card>
</CardGroup>
</CardGroup>