From 2ea1edebaec7d340ad692e39b5e8d3fc5f41a061 Mon Sep 17 00:00:00 2001 From: Zack Lee Date: Thu, 21 Nov 2024 13:00:50 -0500 Subject: [PATCH] Almost done with Python guide --- docs/first-server/python.mdx | 483 +++++++++++++---------------------- 1 file changed, 176 insertions(+), 307 deletions(-) diff --git a/docs/first-server/python.mdx b/docs/first-server/python.mdx index 8936a9d..d86d897 100644 --- a/docs/first-server/python.mdx +++ b/docs/first-server/python.mdx @@ -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 ``` @@ -55,99 +57,175 @@ Let's build your first MCP server in Python! We'll create a weather server that -## 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. - - In `weather_service/src/weather_service/server.py` + + 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" ``` - - Add this functionality: + + + 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" + } ``` - - Add these resource-related handlers to our main function: + + 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: ```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)}") - ``` - - Add these tool-related handlers: + + 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: ```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)}") ``` - - - 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`: