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

Almost done with Python guide

Browse source
This commit is contained in:
Zack Lee 2024-11-21 13:00:50 -05:00
commit 2ea1edebae
No known key found for this signature in database
1 changed files with 175 additions and 306 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -37,6 +37,8 @@ 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>
@ -55,99 +57,175 @@ Let's build your first MCP server in Python! We'll create a weather server that
</Step>
</Steps>
## Create your server
## 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.
<Steps>
<Step title="Add the base imports and setup">
In `weather_service/src/weather_service/server.py`
<Step title="Updating imports and configuration">
First, let's add the imports we need at the top of the file:
```python
import os
import json
import logging
from datetime import datetime, timedelta
from collections.abc import Sequence
from functools import lru_cache
# 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 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 new imports
import os
import json
import logging
from datetime import datetime
import requests
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Add configuration
load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("weather-server")
# 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 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"
async def mcp_server():
# our server implementation will go here
API_BASE_URL = "http://api.openweathermap.org/data/2.5"
DEFAULT_CITY = "London"
CURRENT_WEATHER_ENDPOINT = "weather"
FORECAST_ENDPOINT = "forecast"
```
</Step>
<Step title="Add weather fetching functionality">
Add this functionality:
<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:
```python
# 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")
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"
}
```
</Step>
<Step title="Implement resource handlers">
Add these resource-related handlers to our main function:
<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:
```python
app = Server("weather-server")
@app.list_resources()
async def list_resources() -> list[Resource]:
# RESOURCE HANDLERS
@self.list_resources()
async def handle_list_resources(self) -> list[types.Resource]:
"""List available weather resources."""
uri = AnyUrl(f"weather://{DEFAULT_CITY}/current")
return [
Resource(
types.Resource(
uri=uri,
name=f"Current weather in {DEFAULT_CITY}",
mimeType="application/json",
@ -155,37 +233,36 @@ Let's build your first MCP server in Python! We'll create a weather server that
)
]
@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:
@self.read_resource()
async def handle_read_resource(self, 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)
weather_data = await self.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="Implement tool handlers">
Add these tool-related handlers:
<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:
```python
app = Server("weather-server")
# Resource implementation ...
@app.list_tools()
async def list_tools() -> list[Tool]:
# TOOL HANDLERS
@self.list_tools()
async def handle_list_tools(self) -> list[types.Tool]:
"""List available weather tools."""
return [
Tool(
types.Tool(
name="get_forecast",
description="Get weather forecast for a city",
inputSchema={
@ -207,8 +284,10 @@ Let's build your first MCP server in Python! We'll create a weather server that
)
]
@app.call_tool()
async def call_tool(name: str, arguments: Any) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
@self.call_tool()
async def handle_call_tool(
self, name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""Handle tool calls for weather forecasts."""
if name != "get_forecast":
raise ValueError(f"Unknown tool: {name}")
@ -220,7 +299,7 @@ Let's build your first MCP server in Python! We'll create a weather server that
days = min(int(arguments.get("days", 3)), 5)
try:
response = http.get(
response = self.http.get(
f"{API_BASE_URL}/{FORECAST_ENDPOINT}",
params={
"q": city,
@ -240,7 +319,7 @@ Let's build your first MCP server in Python! We'll create a weather server that
})
return [
TextContent(
types.TextContent(
type="text",
text=json.dumps(forecasts, indent=2)
)
@ -250,226 +329,16 @@ Let's build your first MCP server in Python! We'll create a weather server that
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`: