From ddc20437e00584f9a9139f6275772b7a99c837f7 Mon Sep 17 00:00:00 2001 From: David Soria Parra Date: Thu, 21 Nov 2024 00:37:13 +0000 Subject: [PATCH 1/2] Improve Python First Server This is a first pass as improving the Python First Server documentation. There is still much left and I haven't tested it yet. --- docs/first-server/python.mdx | 647 ++++++++++++++++------------------- 1 file changed, 287 insertions(+), 360 deletions(-) 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 From f1649a310f8a6427ade85356cf20471165c5b9a6 Mon Sep 17 00:00:00 2001 From: David Soria Parra Date: Thu, 21 Nov 2024 14:37:46 +0000 Subject: [PATCH 2/2] update the python impl --- docs/first-server/python.mdx | 624 +++++++++++++++++++++-------------- 1 file changed, 380 insertions(+), 244 deletions(-) diff --git a/docs/first-server/python.mdx b/docs/first-server/python.mdx index 03ae26c..363d6ee 100644 --- a/docs/first-server/python.mdx +++ b/docs/first-server/python.mdx @@ -17,7 +17,6 @@ Let's build your first MCP server in Python! We'll create a weather server that ```bash python --version # Should be 3.10 or higher - pip --version ``` @@ -30,16 +29,15 @@ Let's build your first MCP server in Python! We'll create a weather server that ```bash - mkdir weather-server - cd weather-server - uv init --package --app --name weather-server - source .venv/bin/activate # activate our virtual environment + mkdir weather-service + cd weather-service + uv init --package --app --name weather-service ``` ```bash - uv add mcp + uv add mcp requests python-dotenv ``` @@ -57,203 +55,174 @@ Let's build your first MCP server in Python! We'll create a weather server that ```bash - touch weather_server.py + mkdir -p weather_service/src/weather_service + touch weather_service/src/weather_service/server.py ``` - - In `weather_server.py` + + In `weather_service/src/weather_service/server.py` ```python import os import json import logging - from datetime import datetime - from typing import Any + from datetime import datetime, timedelta + from collections.abc import Sequence + from functools import lru_cache import requests + import asyncio 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, + Resource, + Tool, + TextContent, + ImageContent, + EmbeddedResource, + LoggingLevel, ) + from pydantic import AnyUrl + # Load environment variables load_dotenv() # 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_BASE_URL = "http://api.openweathermap.org/data/2.5" - DEFAULT_CITY = "San Francisco" - CURRENT_WEATHER_ENDPOINT = "weather" - FORECAST_ENDPOINT = "forecast" - - class WeatherServer(Server): - def __init__(self): - super().__init__("example-weather-server") - - # Set up HTTP session - self.http = requests.Session() - self.http.params = { - "appid": API_KEY, - "units": "metric" - } - - @self.list_resources() - async def list_resources(self) -> ListResourcesResult: - # Implementation will come later - pass - - @self.read_resource() - async def read_resource(self, uri: str) -> ReadResourceResult: - # Implementation will come later - pass - - @self.list_tools() - async def list_tools(self) -> ListToolsResult: - # Implementation will come later - pass - - @self.call_tool() - async def call_tool(self, name: str, args: Any) -> CallToolResult: - # Implementation will come later - pass - + 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 ``` - - Add these methods to the `WeatherServer` class: + + Add this functionality: - ```python - @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" - ) - ] - ) + ```python + # Create reusable session + http = requests.Session() + http.params = { + "appid": API_KEY, + "units": "metric" + } - @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( - ErrorData( - code=INVALID_REQUEST, - message=f"Unknown resource: {uri}" - ) - ) + 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() - try: - response = self.http.get( - f"{API_BASE_URL}/{CURRENT_WEATHER_ENDPOINT}", - 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() + } - weather_data = { - "temperature": data["main"]["temp"], - "conditions": data["weather"][0]["description"], - "humidity": data["main"]["humidity"], - "wind_speed": data["wind"]["speed"], - "timestamp": datetime.now().isoformat() - } - return ReadResourceResult( - contents=[ - TextResourceContents( - uri=uri, - mimeType="application/json", - text=json.dumps(weather_data, indent=2) - ) - ] - ) - except requests.RequestException as e: - raise McpError( - ErrorData( - code=INTERNAL_ERROR, - message=f"Weather API error: {str(e)}" - ) - ) + app = Server("weather-server") ``` - - Add these methods to the `WeatherServer` class: + + Add these resource-related handlers to our main function: ```python - @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 - } - }, - "required": ["city"] - } - ) - ] - ) + app = Server("weather-server") - @self.call_tool() - async def call_tool(self, name: str, arguments: Any) -> CallToolResult: + @app.list_resources() + async def list_resources() -> list[Resource]: + """List available weather resources.""" + uri = AnyUrl(f"weather://{DEFAULT_CITY}/current") + return [ + Resource( + uri=uri, + name=f"Current weather in {DEFAULT_CITY}", + mimeType="application/json", + description="Real-time weather data" + ) + ] + + @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 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: + + ```python + app = Server("weather-server") + + # Resource implementation ... + + @app.list_tools() + async def list_tools() -> list[Tool]: + """List available weather tools.""" + return [ + 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"] + } + ) + ] + + @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 McpError( - ErrorData( - code=METHOD_NOT_FOUND, - message=f"Unknown tool: {name}" - ) - ) + raise ValueError(f"Unknown tool: {name}") if not isinstance(arguments, dict) or "city" not in arguments: - raise McpError( - ErrorData( - code=INVALID_PARAMS, - message="Invalid forecast arguments" - ) - ) + raise ValueError("Invalid forecast arguments") city = arguments["city"] days = min(int(arguments.get("days", 3)), 5) try: - response = self.session.get( + response = http.get( f"{API_BASE_URL}/{FORECAST_ENDPOINT}", params={ "q": city, @@ -272,48 +241,50 @@ Let's build your first MCP server in Python! We'll create a weather server that "conditions": day_data["weather"][0]["description"] }) - return CallToolResult( - content=[ - TextContent( - type="text", - text=json.dumps(forecasts, indent=2) - ) - ], - isError=False - ) - except requests.RequestException as e: - raise McpError( - ErrorData( - code=INTERNAL_ERROR, - message=f"Weather API error: {str(e)}" + return [ + TextContent( + type="text", + text=json.dumps(forecasts, indent=2) ) - ) + ] + except requests.RequestException as e: + logger.error(f"Weather API error: {str(e)}") + raise RuntimeError(f"Weather API error: {str(e)}") ``` - - Add this to the end of `weather_server.py`: + + Add this to the end of `weather_service/src/weather_service/server.py`: ```python - async def main(): - server = WeatherServer() + async def main(): + # Import here to avoid issues with event loops + from mcp.server.stdio import stdio_server - # 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 server.run( - read_stream, - write_stream, - server.create_initialization_options() - ) - - if __name__ == "__main__": - import anyio - anyio.run(main) + 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 @@ -330,7 +301,7 @@ Let's build your first MCP server in Python! We'll create a weather server that ```python - @self.list_resources() + @app.list_resources() async def list_resources(self) -> ListResourcesResult: return ListResourcesResult( resources=[ @@ -373,6 +344,33 @@ Let's build your first MCP server in Python! We'll create a weather server that 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 @@ -417,8 +415,36 @@ While this guide uses stdio transport, MCP supports additonal transport options: ### SSE (Server-Sent Events) ```python -from modelcontextprotocol import SSEServerTransport -transport = SSEServerTransport("/events", response) +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 @@ -431,18 +457,8 @@ python --version # Reinstall dependencies uv sync --reinstall - -# Ensure we are using the right virtual environment -source .venv/bin/activate ``` -### Runtime errors - -```bash -# Enable debug logging -export MCP_LOG_LEVEL=DEBUG -python weather_server.py -``` ### Type checking @@ -451,7 +467,7 @@ python weather_server.py uv add --dev pyright # Run type checker -pyright src +uv run pyright src ``` ## Connect to Claude Desktop @@ -515,30 +531,74 @@ pyright src ## Advanced features + + The request context provides access to the current request's metadata and the active client session. Access it through `server.request_context`: + + ```python + @app.call_tool() + async def call_tool(name: str, arguments: Any) -> Sequence[TextContent]: + # Access the current request context + ctx = self.request_context + + # Get request metadata like progress tokens + if progress_token := ctx.meta.progressToken: + # Send progress notifications via the session + await ctx.session.send_progress_notification( + progress_token=progress_token, + progress=0.5, + total=1.0 + ) + + # Sample from the LLM client + result = await ctx.session.create_message( + messages=[ + SamplingMessage( + role="user", + content=TextContent( + type="text", + text="Analyze this weather data: " + json.dumps(arguments) + ) + ) + ], + max_tokens=100 + ) + + return [TextContent(type="text", text=result.content.text)] + ``` + + ```python - from functools import lru_cache - from datetime import datetime, timedelta + # Cache settings + cache_timeout = timedelta(minutes=15) + last_cache_time = None + cached_weather = None - class WeatherServer: - def __init__(self): - self._cache_timeout = timedelta(minutes=15) - self._last_cache_time = None - self._cached_weather = None + async def fetch_weather(city: str) -> dict[str, Any]: + global cached_weather, last_cache_time - @lru_cache(maxsize=32) - 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 - self._last_cache_time is None or - now - self._last_cache_time > self._cache_timeout): + now = datetime.now() + if (cached_weather is None or + last_cache_time is None or + now - last_cache_time > cache_timeout): - response = self.session.get(...) - self._cached_weather = response.json() - self._last_cache_time = now + response = http.get( + f"{API_BASE_URL}/{CURRENT_WEATHER_ENDPOINT}", + params={"q": city} + ) + response.raise_for_status() + data = response.json() - return self._cached_weather + cached_weather = { + "temperature": data["main"]["temp"], + "conditions": data["weather"][0]["description"], + "humidity": data["main"]["humidity"], + "wind_speed": data["wind"]["speed"], + "timestamp": datetime.now().isoformat() + } + last_cache_time = now + + return cached_weather ``` @@ -615,46 +675,122 @@ pyright src - Create `src/weather_service/tests/test_weather_server.py`: + Create `tests/weather_test.py`: ```python import pytest - from unittest.mock import Mock, patch - from weather_service.weather_server import WeatherServer + import os + from unittest.mock import patch, Mock + from datetime import datetime + import json + from pydantic import AnyUrl + os.environ["OPENWEATHER_API_KEY"] = "TEST" + + from weather_service.server import ( + fetch_weather, + read_resource, + call_tool, + list_resources, + list_tools, + DEFAULT_CITY + ) @pytest.fixture - async def server(): - server = WeatherServer() - yield server + def mock_weather_response(): + return { + "main": { + "temp": 20.5, + "humidity": 65 + }, + "weather": [ + {"description": "scattered clouds"} + ], + "wind": { + "speed": 3.6 + } + } - @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": [{ + @pytest.fixture + def mock_forecast_response(): + return { + "list": [ + { "dt_txt": "2024-01-01 12:00:00", - "main": {"temp": 20}, + "main": {"temp": 18.5}, "weather": [{"description": "sunny"}] - }] + }, + { + "dt_txt": "2024-01-02 12:00:00", + "main": {"temp": 17.2}, + "weather": [{"description": "cloudy"}] + } + ] + } + + @pytest.mark.asyncio + async def test_fetch_weather(mock_weather_response): + with patch('requests.Session.get') as mock_get: + mock_get.return_value.json.return_value = mock_weather_response + mock_get.return_value.raise_for_status = Mock() + + weather = await fetch_weather("London") + + assert weather["temperature"] == 20.5 + assert weather["conditions"] == "scattered clouds" + assert weather["humidity"] == 65 + assert weather["wind_speed"] == 3.6 + assert "timestamp" in weather + + @pytest.mark.asyncio + async def test_read_resource(): + with patch('weather_service.server.fetch_weather') as mock_fetch: + mock_fetch.return_value = { + "temperature": 20.5, + "conditions": "clear sky", + "timestamp": datetime.now().isoformat() } - 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 + uri = AnyUrl("weather://London/current") + result = await read_resource(uri) + + assert isinstance(result, str) + assert "temperature" in result + assert "clear sky" in result + + @pytest.mark.asyncio + async def test_call_tool(mock_forecast_response): + with patch('weather_service.server.http.get') as mock_get: + mock_get.return_value.json.return_value = mock_forecast_response + mock_get.return_value.raise_for_status = Mock() + + result = await call_tool("get_forecast", {"city": "London", "days": 2}) + + assert len(result) == 1 + assert result[0].type == "text" + forecast_data = json.loads(result[0].text) + assert len(forecast_data) == 1 + assert forecast_data[0]["temperature"] == 18.5 + assert forecast_data[0]["conditions"] == "sunny" + + @pytest.mark.asyncio + async def test_list_resources(): + resources = await list_resources() + assert len(resources) == 1 + assert resources[0].name == f"Current weather in {DEFAULT_CITY}" + assert resources[0].mimeType == "application/json" + + @pytest.mark.asyncio + async def test_list_tools(): + tools = await list_tools() + assert len(tools) == 1 + assert tools[0].name == "get_forecast" + assert "city" in tools[0].inputSchema["properties"] ``` ```bash uv add --dev pytest pytest-asyncio - uv run pytest src/weather_service/ + uv run pytest ```