diff --git a/docs/first-server/python.mdx b/docs/first-server/python.mdx index d86d897..217e83c 100644 --- a/docs/first-server/python.mdx +++ b/docs/first-server/python.mdx @@ -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 ``` @@ -57,175 +55,99 @@ Let's build your first MCP server in Python! We'll create a weather server that -## 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 - - First, let's add the imports we need at the top of the file: - + + 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 ``` - - - First, let's modify the class docstring and initialization to remove the notes storage and add our HTTP session: - + + 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") ``` - - 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() - } - ``` - - - - Now, let's replace the note-related resource handlers with weather ones: + + 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)}") + ``` - - Remove the entire prompt section from the starter code - we won't be using prompts in our weather service. - - - - Finally, let's replace the note-adding tool with our weather forecast tool: + + 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)}") ``` + + + 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() + ) + ``` + + + + 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'] + ``` + +## Understanding the code + + + + ```python + async def read_resource(self, uri: str) -> ReadResourceResult: + # ... + ``` + + Python type hints help catch errors early and improve code maintainability. + + + + ```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. + + + + ```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. + + + + ```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. + + + +## Best practices + + + + ```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)}" + ) + ``` + + + + ```python + if not isinstance(args, dict) or "city" not in args: + raise McpError( + ErrorCode.INVALID_PARAMS, + "Invalid forecast arguments" + ) + ``` + + + + ```python + if not API_KEY: + raise ValueError("OPENWEATHER_API_KEY is required") + ``` + + + + +## 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 that this step is unnecessary if you've opted to install the server into Claude.app with the MCP project creator script. - - 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 Read more about the Python SDK - + \ No newline at end of file