diff --git a/docs/first-server/python.mdx b/docs/first-server/python.mdx
index 73ac3b8..03ae26c 100644
--- a/docs/first-server/python.mdx
+++ b/docs/first-server/python.mdx
@@ -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
```
@@ -61,113 +62,110 @@ Let's build your first MCP server in Python! We'll create a weather server that
- 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
- ```
+
+ ```
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)}"
+ )
)
```
@@ -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)}"
+ )
)
```
@@ -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)
```
-
- ```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"
- ],
- )
- ```
-
## Understanding the code
@@ -333,10 +321,7 @@ Let's build your first MCP server in Python! We'll create a weather server that
```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
```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
```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
```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
```
-
- ```python
- API_CONFIG = {
- "BASE_URL": "...",
- "ENDPOINTS": {
- "CURRENT": "weather",
- "FORECAST": "forecast"
- }
- }
- ```
-
## 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
```
```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...
```
@@ -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)}")
```
```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"
+ )
+ ]
+ )
```
@@ -648,90 +615,50 @@ mypy weather_server.py
- 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
+ ```
-
- ```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/
+ ```
-## Deployment
-
-
-
- ```dockerfile
- FROM python:3.11-slim
-
- WORKDIR /app
- COPY requirements.txt .
- RUN pip install -r requirements.txt
-
- COPY . .
-
- CMD ["python", "weather_server.py"]
- ```
-
-
-
- ```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
- ```
-
-
-
## Next steps