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

Improve Python First Server

Browse source 
This is a first pass as improving the Python First Server
documentation. There is still much left and I haven't
tested it yet.
This commit is contained in:
David Soria Parra 2024-11-21 00:37:13 +00:00
commit ddc20437e0
No known key found for this signature in database
1 changed files with 295 additions and 368 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -33,6 +33,7 @@ Let's build your first MCP server in Python! We'll create a weather server that
mkdir weather-server
cd weather-server
uv init --package --app --name weather-server
source .venv/bin/activate # activate our virtual environment
```
</Step>
@ -61,113 +62,110 @@ Let's build your first MCP server in Python! We'll create a weather server that
</Step>
<Step title="Add the base code">
In `weather_server.py`:
In `weather_server.py`
```python
import os
import json
import logging
from datetime import datetime
from typing import Any
```python
import os
import json
from datetime import datetime
from typing import Any, Dict, List, Optional
import requests
from dotenv import load_dotenv
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.types import (
ServerCapabilities,
ListResourcesResult,
ReadResourceResult,
ListToolsResult,
CallToolResult,
McpError,
)
import requests
from dotenv import load_dotenv
from modelcontextprotocol import Server, StdioServerTransport
from modelcontextprotocol.types import (
ListResourcesRequest,
ReadResourceRequest,
ListToolsRequest,
CallToolRequest,
ErrorCode,
McpError
)
load_dotenv()
load_dotenv()
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("weather-server")
API_KEY = os.getenv("OPENWEATHER_API_KEY")
if not API_KEY:
raise ValueError("OPENWEATHER_API_KEY environment variable is required")
API_KEY = os.getenv("OPENWEATHER_API_KEY")
if not API_KEY:
raise ValueError("OPENWEATHER_API_KEY environment variable required")
API_CONFIG = {
"BASE_URL": "http://api.openweathermap.org/data/2.5",
"DEFAULT_CITY": "San Francisco",
"ENDPOINTS": {
"CURRENT": "weather",
"FORECAST": "forecast"
}
}
API_BASE_URL = "http://api.openweathermap.org/data/2.5"
DEFAULT_CITY = "San Francisco"
CURRENT_WEATHER_ENDPOINT = "weather"
FORECAST_ENDPOINT = "forecast"
class WeatherServer:
def __init__(self):
self.server = Server(
name="example-weather-server",
version="0.1.0"
)
self.session = requests.Session()
self.session.params = {
"appid": API_KEY,
"units": "metric"
}
class WeatherServer(Server):
def __init__(self):
super().__init__("example-weather-server")
self.setup_handlers()
self.setup_error_handling()
# Set up HTTP session
self.http = requests.Session()
self.http.params = {
"appid": API_KEY,
"units": "metric"
}
def setup_error_handling(self) -> None:
def on_error(error: Exception) -> None:
print("[MCP Error]", error, flush=True)
@self.list_resources()
async def list_resources(self) -> ListResourcesResult:
# Implementation will come later
pass
self.server.onerror = on_error
@self.read_resource()
async def read_resource(self, uri: str) -> ReadResourceResult:
# Implementation will come later
pass
def setup_handlers(self) -> None:
self.server.set_request_handler(
ListResourcesRequest, self.handle_list_resources)
self.server.set_request_handler(
ReadResourceRequest, self.handle_read_resource)
self.server.set_request_handler(
ListToolsRequest, self.handle_list_tools)
self.server.set_request_handler(
CallToolRequest, self.handle_call_tool)
@self.list_tools()
async def list_tools(self) -> ListToolsResult:
# Implementation will come later
pass
async def run(self) -> None:
transport = StdioServerTransport()
await self.server.connect(transport)
print("Weather MCP server running on stdio", flush=True)
@self.call_tool()
async def call_tool(self, name: str, args: Any) -> CallToolResult:
# Implementation will come later
pass
# Handler implementations will continue in next sections
```
```
</Step>
<Step title="Add resource handlers">
Add these methods to the `WeatherServer` class:
```python
async def handle_list_resources(
self,
request: ListResourcesRequest
) -> Dict[str, Any]:
return {
"resources": [{
"uri": f"weather://{API_CONFIG['DEFAULT_CITY']}/current",
"name": f"Current weather in {API_CONFIG['DEFAULT_CITY']}",
"mimeType": "application/json",
"description": "Real-time weather data including temperature, conditions, humidity, and wind speed"
}]
}
@self.list_resources()
async def list_resources(self) -> ListResourcesResult:
"""List available weather resources."""
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"
)
]
)
async def handle_read_resource(
self,
request: ReadResourceRequest
) -> Dict[str, Any]:
city = API_CONFIG["DEFAULT_CITY"]
if request.params.uri != f"weather://{city}/current":
@self.read_resource()
async def read_resource(self, uri: AnyUrl) -> ReadResourceResult:
"""Read current weather data for a city."""
city = DEFAULT_CITY
if str(uri) != f"weather://{city}/current":
raise McpError(
ErrorCode.INVALID_REQUEST,
f"Unknown resource: {request.params.uri}"
ErrorData(
code=INVALID_REQUEST,
message=f"Unknown resource: {uri}"
)
)
try:
response = self.session.get(
f"{API_CONFIG['BASE_URL']}/{API_CONFIG['ENDPOINTS']['CURRENT']}",
response = self.http.get(
f"{API_BASE_URL}/{CURRENT_WEATHER_ENDPOINT}",
params={"q": city}
)
response.raise_for_status()
@ -181,17 +179,21 @@ Let's build your first MCP server in Python! We'll create a weather server that
"timestamp": datetime.now().isoformat()
}
return {
"contents": [{
"uri": request.params.uri,
"mimeType": "application/json",
"text": json.dumps(weather_data, indent=2)
}]
}
return ReadResourceResult(
contents=[
TextResourceContents(
uri=uri,
mimeType="application/json",
text=json.dumps(weather_data, indent=2)
)
]
)
except requests.RequestException as e:
raise McpError(
ErrorCode.INTERNAL_ERROR,
f"Weather API error: {str(e)}"
ErrorData(
code=INTERNAL_ERROR,
message=f"Weather API error: {str(e)}"
)
)
```
</Step>
@ -200,56 +202,59 @@ Let's build your first MCP server in Python! We'll create a weather server that
Add these methods to the `WeatherServer` class:
```python
async def handle_list_tools(
self,
request: ListToolsRequest
) -> Dict[str, Any]:
return {
"tools": [{
"name": "get_forecast",
"description": "Get weather forecast for a city",
"inputSchema": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "City name"
@self.list_tools()
async def list_tools(self) -> ListToolsResult:
"""List available weather tools."""
return ListToolsResult(
tools=[
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
}
},
"days": {
"type": "number",
"description": "Number of days (1-5)",
"minimum": 1,
"maximum": 5
}
},
"required": ["city"]
}
}]
}
"required": ["city"]
}
)
]
)
async def handle_call_tool(
self,
request: CallToolRequest
) -> Dict[str, Any]:
if request.params.name != "get_forecast":
@self.call_tool()
async def call_tool(self, name: str, arguments: Any) -> CallToolResult:
"""Handle tool calls for weather forecasts."""
if name != "get_forecast":
raise McpError(
ErrorCode.METHOD_NOT_FOUND,
f"Unknown tool: {request.params.name}"
ErrorData(
code=METHOD_NOT_FOUND,
message=f"Unknown tool: {name}"
)
)
args = request.params.arguments
if not isinstance(args, dict) or "city" not in args:
if not isinstance(arguments, dict) or "city" not in arguments:
raise McpError(
ErrorCode.INVALID_PARAMS,
"Invalid forecast arguments"
ErrorData(
code=INVALID_PARAMS,
message="Invalid forecast arguments"
)
)
city = args["city"]
days = min(int(args.get("days", 3)), 5)
city = arguments["city"]
days = min(int(arguments.get("days", 3)), 5)
try:
response = self.session.get(
f"{API_CONFIG['BASE_URL']}/{API_CONFIG['ENDPOINTS']['FORECAST']}",
f"{API_BASE_URL}/{FORECAST_ENDPOINT}",
params={
"q": city,
"cnt": days * 8 # API returns 3-hour intervals
@ -267,11 +272,21 @@ Let's build your first MCP server in Python! We'll create a weather server that
"conditions": day_data["weather"][0]["description"]
})
return {"toolResult": forecasts}
return CallToolResult(
content=[
TextContent(
type="text",
text=json.dumps(forecasts, indent=2)
)
],
isError=False
)
except requests.RequestException as e:
raise McpError(
ErrorCode.INTERNAL_ERROR,
f"Weather API error: {str(e)}"
ErrorData(
code=INTERNAL_ERROR,
message=f"Weather API error: {str(e)}"
)
)
```
</Step>
@ -280,52 +295,25 @@ Let's build your first MCP server in Python! We'll create a weather server that
Add this to the end of `weather_server.py`:
```python
if __name__ == "__main__":
import asyncio
import signal
async def main():
server = WeatherServer()
async def main():
server = WeatherServer()
def signal_handler(sig, frame):
print("Shutting down...", flush=True)
asyncio.create_task(server.server.close())
signal.signal(signal.SIGINT, signal_handler)
await server.run()
# Import here to avoid issues with event loops
from mcp.server.stdio import stdio_server
asyncio.run(main())
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
if __name__ == "__main__":
import anyio
anyio.run(main)
```
</Step>
<Step title="Make the script executable">
```bash
chmod +x weather_server.py
pip install -e .
```
Create `setup.py`:
```python
from setuptools import setup
setup(
name="weather-server",
version="0.1.0",
py_modules=["weather_server"],
entry_points={
"console_scripts": [
"weather-server=weather_server:main",
],
},
install_requires=[
"modelcontextprotocol",
"requests",
"python-dotenv"
],
)
```
</Step>
</Steps>
## Understanding the code
@ -333,10 +321,7 @@ Let's build your first MCP server in Python! We'll create a weather server that
<Tabs>
<Tab title="Type Hints">
```python
async def handle_read_resource(
self,
request: ReadResourceRequest
) -> Dict[str, Any]:
async def read_resource(self, uri: str) -> ReadResourceResult:
# ...
```
@ -345,17 +330,18 @@ Let's build your first MCP server in Python! We'll create a weather server that
<Tab title="Resources">
```python
async def handle_list_resources(
self,
request: ListResourcesRequest
) -> Dict[str, Any]:
return {
"resources": [{
"uri": f"weather://{DEFAULT_CITY}/current",
"name": f"Current weather in {DEFAULT_CITY}",
"mimeType": "application/json"
}]
}
@self.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.
@ -363,17 +349,26 @@ Let's build your first MCP server in Python! We'll create a weather server that
<Tab title="Tools">
```python
{
"name": "get_forecast",
"description": "Get weather forecast for a city",
"inputSchema": {
Tool(
name="get_forecast",
description="Get weather forecast for a city",
inputSchema={
"type": "object",
"properties": {
"city": { "type": "string" },
"days": { "type": "number" }
}
"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.
@ -386,7 +381,7 @@ Let's build your first MCP server in Python! We'll create a weather server that
<Card title="Error Handling" icon="shield">
```python
try:
response = self.session.get(...)
response = self.http.get(...)
response.raise_for_status()
except requests.RequestException as e:
raise McpError(
@ -413,29 +408,11 @@ Let's build your first MCP server in Python! We'll create a weather server that
```
</Card>
<Card title="Constants" icon="box">
```python
API_CONFIG = {
"BASE_URL": "...",
"ENDPOINTS": {
"CURRENT": "weather",
"FORECAST": "forecast"
}
}
```
</Card>
</CardGroup>
## Available transports
While this guide uses stdio transport, MCP supports multiple transport options:
### WebSocket
```python
from modelcontextprotocol import WebSocketServerTransport
transport = WebSocketServerTransport()
```
While this guide uses stdio transport, MCP supports additonal transport options:
### SSE (Server-Sent Events)
@ -453,8 +430,10 @@ transport = SSEServerTransport("/events", response)
python --version
# Reinstall dependencies
pip uninstall -r requirements.txt
pip install -r requirements.txt
uv sync --reinstall
# Ensure we are using the right virtual environment
source .venv/bin/activate
```
### Runtime errors
@ -469,10 +448,10 @@ python weather_server.py
```bash
# Install mypy
pip install mypy
uv add --dev pyright
# Run type checker
mypy weather_server.py
pyright src
```
## Connect to Claude Desktop
@ -485,7 +464,8 @@ mypy weather_server.py
{
"mcpServers": {
"weather": {
"command": "weather-server",
"command": "uv",
"args": "--directory path/to/your/project run weather-service",
"env": {
"OPENWEATHER_API_KEY": "your-api-key",
}
@ -550,45 +530,39 @@ mypy weather_server.py
async def _fetch_weather(self, city: str) -> Dict[str, Any]:
# Cache expires after 15 minutes
now = datetime.now()
if (self._cached_weather is None or
if (self._cached_weather is None or
self._last_cache_time is None or
now - self._last_cache_time > self._cache_timeout):
response = self.session.get(...)
self._cached_weather = response.json()
self._last_cache_time = now
return self._cached_weather
```
</Step>
<Step title="Add progress notifications">
```python
async def handle_call_tool(
self,
request: CallToolRequest
) -> Dict[str, Any]:
if progress_token := request.params._meta.get("progressToken"):
@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.server.notification({
"method": "notifications/progress",
"params": {
"progress": 1,
"total": 2,
"progressToken": progress_token
}
})
await self.request_context.session.send_progress_notification(
progress_token=progress_token,
progress=1,
total=2
)
# Fetch data...
await self.server.notification({
"method": "notifications/progress",
"params": {
"progress": 2,
"total": 2,
"progressToken": progress_token
}
})
await self.request_context.session.send_progress_notification(
progress_token=progress_token,
progress=2,
total=2
)
# Rest of the method implementation...
```
</Step>
@ -596,50 +570,43 @@ mypy weather_server.py
```python
import logging
class WeatherServer:
class WeatherServer(Server):
def __init__(self):
super().__init__("example-weather-server")
self.logger = logging.getLogger("weather-server")
self.logger.setLevel(logging.INFO)
# Set up logging handler
self.server.set_request_handler(
SetLevelRequest,
self.handle_set_log_level
)
async def handle_set_log_level(
self,
request: SetLevelRequest
) -> Dict[str, Any]:
level = request.params.level.upper()
self.logger.setLevel(getattr(logging, level))
await self.server.notification({
"method": "notifications/message",
"params": {
"level": "info",
"logger": "weather-server",
"data": f"Log level set to {level}"
}
})
return {}
@self.set_logging_level()
async def set_logging_level(self, level: LoggingLevel) -> EmptyResult:
self.logger.setLevel(level.upper())
await self.request_context.session.send_log_message(
level="info",
data=f"Log level set to {level}",
logger="weather-server"
)
return EmptyResult()
# Use self.logger for logging throughout the class methods
# For example:
# self.logger.info("Weather data fetched successfully")
# self.logger.error(f"Error fetching weather data: {str(e)}")
```
</Step>
<Step title="Add resource templates">
```python
async def handle_list_resources(
self,
request: ListResourcesRequest
) -> Dict[str, Any]:
return {
"resources": [...],
"resourceTemplates": [{
"uriTemplate": "weather://{city}/current",
"name": "Current weather for any city",
"mimeType": "application/json"
}]
}
@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"
)
]
)
```
</Step>
</Steps>
@ -648,90 +615,50 @@ mypy weather_server.py
<Steps>
<Step title="Create test file">
Create `test_weather_server.py`:
Create `src/weather_service/tests/test_weather_server.py`:
```python
import pytest
from unittest.mock import Mock, patch
from weather_server import WeatherServer
```python
import pytest
from unittest.mock import Mock, patch
from weather_service.weather_server import WeatherServer
@pytest.fixture
async def server():
server = WeatherServer()
yield server
await server.server.close()
@pytest.fixture
async def server():
server = WeatherServer()
yield server
@pytest.mark.asyncio
async def test_list_resources(server):
request = Mock()
result = await server.handle_list_resources(request)
assert len(result["resources"]) > 0
assert result["resources"][0]["mimeType"] == "application/json"
@pytest.mark.asyncio
async def test_list_resources(server):
result = await server.list_resources()
assert len(result.resources) > 0
assert result.resources[0].mimeType == "application/json"
@pytest.mark.asyncio
async def test_get_forecast():
with patch("requests.Session") as mock_session:
mock_session.return_value.get.return_value.json.return_value = {
"list": [{
"dt_txt": "2024-01-01 12:00:00",
"main": {"temp": 20},
"weather": [{"description": "sunny"}]
}]
}
server = WeatherServer()
request = Mock()
request.params.name = "get_forecast"
request.params.arguments = {"city": "Tokyo", "days": 1}
result = await server.handle_call_tool(request)
assert len(result["toolResult"]) == 1
assert result["toolResult"][0]["temperature"] == 20
```
@pytest.mark.asyncio
async def test_get_forecast():
with patch("requests.Session") as mock_session:
mock_session.return_value.get.return_value.json.return_value = {
"list": [{
"dt_txt": "2024-01-01 12:00:00",
"main": {"temp": 20},
"weather": [{"description": "sunny"}]
}]
}
server = WeatherServer()
result = await server.call_tool("get_forecast", {"city": "Tokyo", "days": 1})
assert len(result.content) == 1
assert "temperature" in result.content[0].text
assert "20" in result.content[0].text
```
</Step>
<Step title="Run tests">
```bash
pip install pytest pytest-asyncio
pytest test_weather_server.py -v
```
```bash
uv add --dev pytest pytest-asyncio
uv run pytest src/weather_service/
```
</Step>
</Steps>
## Deployment
<CardGroup cols={1}>
<Card title="Docker" icon="docker">
```dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "weather_server.py"]
```
</Card>
<Card title="Systemd Service" icon="server">
```ini
[Unit]
Description=Weather MCP Server
After=network.target
[Service]
ExecStart=/path/to/venv/bin/python /path/to/weather_server.py
Environment=OPENWEATHER_API_KEY=your-key-here
Restart=always
[Install]
WantedBy=multi-user.target
```
</Card>
</CardGroup>
## Next steps
<CardGroup cols={2}>