Costruire un MCP server in Python

Molti LLM non sono attualmente in grado di recuperare le previsioni e gli avvisi di maltempo. Usiamo MCP per risolvere questo problema! Costruiremo un server che espone due strumenti: get-alerts e get-forecast. I server possono connettersi a qualsiasi client come quello visto nel precedente articolo di questo blog.

Concetti fondamentali di MCP

I server MCP possono offrire tre tipi principali di funzionalità:

Risorse: dati simili a file che possono essere letti dai client (come risposte API o contenuto di file)
Strumenti: funzioni che possono essere richiamate dall'LLM (con l'approvazione dell'utente)
Prompt: modelli predefiniti che aiutano gli utenti a svolgere attività specifiche

Questo tutorial si concentrerà principalmente sugli strumenti.

Installazione in ambiente Ubuntu

Assicurati di avere Python 3.10 o superiore installato con MCP SDK 1.2.0 o superiore.
1) Installiamo uv con il seguente comando:
curl -LsSf https://astral.sh/uv/install.sh | sh

2) Creiamo il progetto e un ambiente virtuale:

# Creaiamo una nuova cartella
uv init weather
cd weather

# Creiamo un ambiente virtuale e lo attiviamo
uv venv --python 3.10
source .venv/bin/activate

# Installiamo le dipendenze necessarie
uv add --python 3.10 "mcp[cli]" httpx

# Creiamo il file del server
touch weather.py
		

3) Aggiungi all'inizio del file weather.py le seguenti importazioni e i seguenti settaggi:

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP

# Initialize FastMCP server
mcp = FastMCP("weather")

# Constants
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

La classe FastMCP utilizza suggerimenti sui tipi e docstring Python per generare automaticamente definizioni di strumenti, semplificando la creazione e la gestione degli strumenti MCP.

4) Definiamo delle funzioni utili per il recupero di informazioni dal sito NWS (National Weather Service) degli Stati Uniti:

async def make_nws_request(url: str) -> dict[str, Any] | None:
    """Make a request to the NWS API with proper error handling."""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None

def format_alert(feature: dict) -> str:
    """Format an alert feature into a readable string."""
    props = feature["properties"]
    return f"""
Event: {props.get('event', 'Unknown')}
Area: {props.get('areaDesc', 'Unknown')}
Severity: {props.get('severity', 'Unknown')}
Description: {props.get('description', 'No description available')}
Instructions: {props.get('instruction', 'No specific instructions provided')}
"""
		

5) Ultimo passo: definizione dei tool e inizializzazione del server:

@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get weather alerts for a US state.

    Args:
        state: Two-letter US state code (e.g. CA, NY)
    """
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data = await make_nws_request(url)

    if not data or "features" not in data:
        return "Unable to fetch alerts or no alerts found."

    if not data["features"]:
        return "No active alerts for this state."

    alerts = [format_alert(feature) for feature in data["features"]]
    return "\n---\n".join(alerts)

@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """Get weather forecast for a location.

    Args:
        latitude: Latitude of the location
        longitude: Longitude of the location
    """
    # First get the forecast grid endpoint
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return "Unable to fetch forecast data for this location."

    # Get the forecast URL from the points response
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)

    if not forecast_data:
        return "Unable to fetch detailed forecast."

    # Format the periods into a readable forecast
    periods = forecast_data["properties"]["periods"]
    forecasts = []
    for period in periods[:5]:  # Only show next 5 periods
        forecast = f"""
{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""
        forecasts.append(forecast)

    return "\n---\n".join(forecasts)


if __name__ == "__main__":
    # Initialize and run the server
    mcp.run(transport='stdio')		

Il tuo server è completo! Esegui uv run weather.py per verificare che tutto funzioni. Il tuo MCP server è ora pronto per essere integrato in Claude Desktop o altri client compatibili. Assicurati di mantenere sempre aggiornata la documentazione delle tue funzioni e di implementare un sistema di logging adeguato per il debugging. Man mano che il protocollo evolve, potrai facilmente estendere il tuo server con nuove funzionalità seguendo gli stessi principi architetturali che hai imparato in questo tutorial.