docs/docs/first-server/python.mdx

---
title: "Python"
description: "Create a simple MCP server in Python in 15 minutes"
---

Let's build your first MCP server in Python! We'll create a weather server that provides current weather data as a resource and lets Claude fetch forecasts using tools.

<Note>
  This guide uses the OpenWeatherMap API. You'll need a free API key from [OpenWeatherMap](https://openweathermap.org/api) to follow along.
</Note>

## Prerequisites

<Steps>
  <Step title="Install Python">
    You'll need Python 3.10 or higher:

    ```bash
    python --version  # Should be 3.10 or higher
    ```
  </Step>

  <Step title="Install uv (https://docs.astral.sh/uv/) via homebrew">
    ```bash
    brew install uv
    uv --version # Should be 0.4.18 or higher
    ```
  </Step>

  <Step title="Create a new project">
    ```bash
    mkdir weather-service
    cd weather-service
    uv init --package --app --name weather-service
    ```
  </Step>

  <Step title="Install dependencies">
    ```bash
    uv add mcp requests python-dotenv
    ```
  </Step>

  <Step title="Set up environment">
    Create `.env`:

    ```bash
    OPENWEATHER_API_KEY=your-api-key-here
    ```
  </Step>
</Steps>

## Create your server

<Steps>
  <Step title="Create project structure">
    ```bash
    mkdir -p weather_service/src/weather_service
    touch weather_service/src/weather_service/server.py
    ```
  </Step>

  <Step title="Add the base imports and setup">
  In `weather_service/src/weather_service/server.py`
  ```python
  import os
  import json
  import logging
  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.types import (
      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 = "London"
   CURRENT_WEATHER_ENDPOINT = "weather"
   FORECAST_ENDPOINT = "forecast"

  async def mcp_server():
     # our server implementation will go here
  ```
  </Step>

  <Step title="Add weather fetching functionality">
  Add this functionality:

  ```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")
    ```
  </Step>

  <Step title="Implement resource handlers">
    Add these resource-related handlers to our main function:

    ```python
    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 [
            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)}")

    ```
  </Step>

  <Step title="Implement tool handlers">
    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 ValueError(f"Unknown tool: {name}")

        if not isinstance(arguments, dict) or "city" not in arguments:
            raise ValueError("Invalid forecast arguments")

        city = arguments["city"]
        days = min(int(arguments.get("days", 3)), 5)

        try:
            response = http.get(
                f"{API_BASE_URL}/{FORECAST_ENDPOINT}",
                params={
                    "q": city,
                    "cnt": days * 8  # API returns 3-hour intervals
                }
            )
            response.raise_for_status()
            data = response.json()

            forecasts = []
            for i in range(0, len(data["list"]), 8):
                day_data = data["list"][i]
                forecasts.append({
                    "date": day_data["dt_txt"].split()[0],
                    "temperature": day_data["main"]["temp"],
                    "conditions": day_data["weather"][0]["description"]
                })

            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)}")
    ```
  </Step>

  <Step title="Add the main function">
    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()
            )
    ```
  </Step>

  <Step title="Check your entry point in __init__.py">
    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']
    ```
  </Step>
</Steps>

## Understanding the code

<Tabs>
  <Tab title="Type Hints">
    ```python
    async def read_resource(self, uri: str) -> ReadResourceResult:
        # ...
    ```

    Python type hints help catch errors early and improve code maintainability.
  </Tab>

  <Tab title="Resources">
    ```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.
  </Tab>

  <Tab title="Tools">
    ```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.
  </Tab>

  <Tab title="Server Structure">
    ```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.
  </Tab>
</Tabs>

## Best practices

<CardGroup cols={1}>
  <Card title="Error Handling" icon="shield">
    ```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)}"
        )
    ```
  </Card>

  <Card title="Type Validation" icon="check">
    ```python
    if not isinstance(args, dict) or "city" not in args:
        raise McpError(
            ErrorCode.INVALID_PARAMS,
            "Invalid forecast arguments"
        )
    ```
  </Card>

  <Card title="Environment Variables" icon="gear">
    ```python
    if not API_KEY:
        raise ValueError("OPENWEATHER_API_KEY is required")
    ```
  </Card>

</CardGroup>

## 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

<Steps>
  <Step title="Update Claude config">
    Add to `claude_desktop_config.json`:

    ```json
    {
      "mcpServers": {
        "weather": {
          "command": "uv",
          "args": "--directory path/to/your/project run weather-service",
          "env": {
            "OPENWEATHER_API_KEY": "your-api-key",
          }
        }
      }
    }
    ```
  </Step>

  <Step title="Restart Claude">
    1. Quit Claude completely

    2. Start Claude again

    3. Look for your weather server in the 🔌 menu
  </Step>
</Steps>

## Try it out!

<AccordionGroup>
  <Accordion title="Check Current Weather" active>
    Ask Claude:

    ```
    What's the current weather in San Francisco? Can you analyze the conditions and tell me if it's a good day for outdoor activities?
    ```
  </Accordion>

  <Accordion title="Get a Forecast">
    Ask Claude:

    ```
    Can you get me a 5-day forecast for Tokyo and help me plan what clothes to pack for my trip?
    ```
  </Accordion>

  <Accordion title="Compare Weather">
    Ask Claude:

    ```
    Can you analyze the forecast for both Tokyo and San Francisco and tell me which city would be better for outdoor photography this week?
    ```
  </Accordion>
</AccordionGroup>

## Advanced features

<Steps>
  <Step title="Understanding Request Context">
    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)]
    ```
  </Step>

  <Step title="Add caching">
    ```python
    # Cache settings
    cache_timeout = timedelta(minutes=15)
    last_cache_time = None
    cached_weather = None

    async def fetch_weather(city: str) -> dict[str, Any]:
        global cached_weather, last_cache_time

        now = datetime.now()
        if (cached_weather is None or
            last_cache_time is None or
            now - last_cache_time > cache_timeout):

            response = http.get(
                f"{API_BASE_URL}/{CURRENT_WEATHER_ENDPOINT}",
                params={"q": city}
            )
            response.raise_for_status()
            data = response.json()

            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
    ```
  </Step>

  <Step title="Add progress notifications">
    ```python
    @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.request_context.session.send_progress_notification(
                progress_token=progress_token,
                progress=1,
                total=2
            )

            # Fetch data...

            await self.request_context.session.send_progress_notification(
                progress_token=progress_token,
                progress=2,
                total=2
            )

        # Rest of the method implementation...
    ```
  </Step>

  <Step title="Add logging support">
    ```python
    import logging

    class WeatherServer(Server):
        def __init__(self):
            super().__init__("example-weather-server")
            self.logger = logging.getLogger("weather-server")
            self.logger.setLevel(logging.INFO)

        @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
    @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>

## Testing

<Steps>
  <Step title="Create test file">
  Create `tests/weather_test.py`:

  ```python
  import pytest
  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
  def mock_weather_response():
      return {
          "main": {
              "temp": 20.5,
              "humidity": 65
          },
          "weather": [
              {"description": "scattered clouds"}
          ],
          "wind": {
              "speed": 3.6
          }
      }

  @pytest.fixture
  def mock_forecast_response():
      return {
          "list": [
              {
                  "dt_txt": "2024-01-01 12:00:00",
                  "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()
          }

          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"]
  ```
  </Step>
  <Step title="Run tests">
  ```bash
  uv add --dev pytest pytest-asyncio
  uv run pytest
  ```
  </Step>
</Steps>

## Next steps

<CardGroup cols={2}>
  <Card title="Architecture overview" icon="sitemap" href="/docs/concepts/architecture">
    Learn more about the MCP architecture
  </Card>

  <Card title="Python SDK" icon="python" href="/api-reference/python/class-reference">
    Read more about the Python SDK
  </Card>
</CardGroup>