MCP-Server in Python bauen: Anleitung + Inspector

Das Versprechen von KI-Assistenten war immer ihre Fähigkeit, bei realen Aufgaben zu helfen, doch es gab eine grundlegende Einschränkung: Die meisten KI-Systeme arbeiten isoliert und können nicht auf die spezifischen Datenquellen, APIs und Werkzeuge zugreifen, die sie für deinen konkreten Anwendungsfall wirklich nützlich machen. Ob du eine KI brauchst, die die Datenbank deines Unternehmens abfragt, mit deinem Projektmanagementsystem interagiert oder auf Echtzeitdaten aus spezialisierten APIs zugreift – der herkömmliche Ansatz erforderte den Aufbau individueller Integrationen für jede KI-Plattform, ein zeitaufwendiger und fragmentierter Prozess.

Genau hier verändert das Model Context Protocol (MCP) alles. Statt dass KI-Assistenten auf ihre Trainingsdaten beschränkt sind oder komplexe individuelle Integrationen benötigen, bietet MCP eine standardisierte Möglichkeit, KI-Systeme mit deinen spezifischen Datenquellen und Werkzeugen zu verbinden. Soll deine KI auf deine Kundendatenbank zugreifen? Baue einen MCP-Server. Möchtest du, dass sie mit deiner Bestandsverwaltungs-API interagiert? Erstelle einen MCP-Server. Derselbe Server funktioniert über verschiedene KI-Plattformen hinweg und macht es überflüssig, Integrationen für jedes System neu aufzubauen.

Die Verbreitung von MCP beschleunigt sich rasant im gesamten KI-Ökosystem. Große Plattformen setzen das Protokoll als Standard für KI-Integration ein. Claude Desktop und Claude for Code bieten bereits native MCP-Unterstützung, sodass Nutzer nahtlos eigene Datenquellen und Werkzeuge anbinden können. Die großen Anbieter von KI-APIs – OpenAI, Anthropic und Google – fügen ihren Completion-APIs MCP-Kompatibilität hinzu, was es Entwicklern ermöglicht, KI-Anwendungen zu bauen, die über standardisierte Schnittstellen auf externe Systeme zugreifen können. Dieses wachsende Ökosystem bedeutet, dass die MCP-Server, die du heute baust, mit einem ständig wachsenden Spektrum an KI-Plattformen und Anwendungen funktionieren werden.

In diesem umfassenden Tutorial lernst du, wie du produktionsreife MCP-Server mit Python entwickelst – durch einen iterativen, praxisnahen Ansatz. Wir beginnen mit einem minimalen Server und fügen schrittweise Funktionen hinzu, sodass du genau nachvollziehen kannst, wie du deinen Server Schritt für Schritt entwickelst, testest und erweiterst. Am Ende dieser Anleitung hast du einen vollständigen Wetterdienst-MCP-Server mit fortgeschrittenen Funktionen gebaut, darunter KI-gestütztes Sampling, OAuth-Authentifizierung und Produktionsbereitstellungsfähigkeiten.

MCP verstehen

Das Model Context Protocol stellt einen Paradigmenwechsel darin dar, wie wir über die Architektur der KI-Integration denken. Im Kern ist MCP ein offenes Protokoll, das standardisiert, wie Anwendungen großen Sprachmodellen Kontext bereitstellen, doch seine Auswirkungen reichen weit über das einfache Teilen von Daten hinaus.

Um zu verstehen, warum MCP wichtig ist, betrachte den aktuellen Stand der KI-Integrationen. Die meisten KI-Anwendungen arbeiten heute isoliert und haben nur begrenzte Möglichkeiten, auf Echtzeitdaten zuzugreifen oder mit externen Systemen zu interagieren. Wenn Entwickler einem KI-Assistenten Zugriff auf eine Datenbank, ein Dateisystem oder einen Webdienst geben wollen, müssen sie in der Regel individuelle Lösungen bauen, die eng an bestimmte KI-Plattformen gekoppelt sind. Dieser Ansatz schafft mehrere Probleme: Anbieterabhängigkeit, doppelter Aufwand über verschiedene KI-Plattformen hinweg, Sicherheitsbedenken durch direkten API-Zugriff und Wartungsaufwand, wenn sich APIs weiterentwickeln.

MCP begegnet diesen Herausforderungen durch eine Client-Server-Architektur, die eine standardisierte Vermittlungsschicht einführt. Statt dass KI-Anwendungen direkt auf externe Systeme zugreifen, kommunizieren sie über MCP-Server, die als sichere, standardisierte Gateways fungieren. Diese Architektur bietet mehrere zentrale Vorteile, die sie besonders für Unternehmens- und Produktionsbereitstellungen leistungsfähig machen.

Grundlegende MCP-Architektur

Im Herzen der MCP-Architektur stehen drei zentrale Beteiligte, die zusammenwirken, um eine nahtlose KI-Integration zu ermöglichen:

MCP Host: Die KI-Anwendung, die Verbindungen zu mehreren MCP-Servern koordiniert und verwaltet. Beliebte KI-Anwendungen wie Claude Desktop fungieren als MCP-Hosts, wenn sie das Protokoll unterstützen.

MCP Client: Eine Komponente innerhalb des Hosts, die dedizierte Verbindungen zu einzelnen MCP-Servern aufrechterhält und die Kommunikation auf Protokollebene sowie den Verbindungslebenszyklus abwickelt.

MCP Server: Die Komponente, die MCP-Clients Daten und Funktionalität auf standardisierte Weise bereitstellt. Server können lokal (über STDIO-Transport) oder entfernt (über HTTP-Transport) laufen.

Die drei Säulen von MCP

Das Protokoll definiert drei grundlegende Primitive, die Server bereitstellen können:

Tools sind ausführbare Funktionen, die KI-Anwendungen aufrufen können, um Aktionen durchzuführen. Dazu können Operationen wie das Abfragen einer Datenbank, das Versenden einer E-Mail oder der Aufruf einer externen API gehören.

Resources stellen KI-Anwendungen kontextbezogene Informationen bereit, ohne Aktionen auszuführen. Sie repräsentieren Daten, die von der KI gelesen und verstanden werden können, etwa Dateiinhalte oder Datenbankeinträge.

Prompts sind wiederverwendbare Vorlagen, die dabei helfen, Interaktionen mit Sprachmodellen zu strukturieren, und bieten eine Möglichkeit, Domänenwissen und Best Practices zu kapseln.

Einrichten deiner Entwicklungsumgebung

Bevor wir mit dem Aufbau unseres MCP-Servers beginnen, richten wir eine geeignete Entwicklungsumgebung ein, die sowohl schnelle Iteration als auch Produktionsbereitstellung unterstützt. Wir werden in diesem Tutorial durchgängig das Model Context Protocol Python SDK verwenden.

Voraussetzungen

Stell sicher, dass Python 3.10 oder höher auf deinem System installiert ist. Deine Python-Version kannst du so prüfen:

python --version
# or
python3 --version

dein Projekt erstellen

Erstelle ein neues Projektverzeichnis und richte eine virtuelle Umgebung ein:

# Create project directory
mkdir weather-mcp-server
cd weather-mcp-server

# Create a virtual environment
python -m venv .venv

# Activate the virtual environment
# On macOS/Linux:
source .venv/bin/activate

# On Windows:
.venv\Scripts\activate

Abhängigkeiten installieren

Installiere das MCP Python SDK und zusätzliche Abhängigkeiten mit pip:

# Upgrade pip to the latest version
pip install --upgrade pip

# Install MCP SDK with CLI tools
pip install "mcp[cli]"

# Install HTTP client for API requests
pip install httpx

# Install JWT library for authentication (we'll use this later)
pip install PyJWT

# Install development dependencies
pip install pytest black isort mypy

Eine Requirements-Datei erstellen

Erstelle eine requirements.txt-Datei, um deine Abhängigkeiten zu verfolgen:

# Generate requirements file
pip freeze > requirements.txt

Deine requirements.txt sollte Einträge wie diese enthalten:

mcp[cli]
httpx
PyJWT
pytest
black
isort
mypy

deinen ersten MCP-Server bauen: Schritt für Schritt

Nun bauen wir unseren Wetter-MCP-Server iterativ auf, beginnend mit der einfachsten möglichen Implementierung, und fügen Schritt für Schritt Funktionen hinzu. Dieser Ansatz hilft dir, jede Komponente zu verstehen, und erleichtert die Fehlersuche.

Schritt 1: Einen minimalen MCP-Server erstellen

Beginnen wir mit dem absoluten Minimum – einem Server, der nichts weiter tut, als auf grundlegende MCP-Protokollnachrichten zu antworten. Erstelle eine Datei namens weather_server.py:

"""
Step 1: Minimal MCP server that responds to protocol messages
"""
from mcp.server.fastmcp import FastMCP

# Create the MCP server instance
mcp = FastMCP("weather-server")

if __name__ == "__main__":
    # Run the server using STDIO transport
    mcp.run(transport='stdio')

Teste diesen minimalen Server:

# Start the MCP Inspector to test your server
python -m mcp dev weather_server.py

Der MCP Inspector startet eine Weboberfläche (typischerweise unter http://localhost:3000), in der du sehen kannst, dass dein Server läuft und auf MCP-Protokollnachrichten antwortet, auch wenn er noch keine Tools bereitstellt.

Schritt 2: dein erstes Tool hinzufügen

Fügen wir nun ein einfaches Tool hinzu, das statische Wetterinformationen zurückgibt:

"""
Step 2: Add a simple weather tool with static data
"""
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather-server")

@mcp.tool()
def get_weather(city: str) -> str:
    """Get current weather for a city (demo with static data)"""
    # For now, return static data to test the tool mechanism
    return f"Weather in {city}: Sunny, 22°C (This is demo data)"

if __name__ == "__main__":
    mcp.run(transport='stdio')

Teste das neue Tool im MCP Inspector. Du solltest nun ein get_weather-Tool sehen, das du mit verschiedenen Städtenamen aufrufen kannst.

Schritt 3: Echte API-Integration hinzufügen

Verbinden wir uns nun mit einer echten Wetter-API. Wir verwenden die API des National Weather Service, die kostenlos ist und keine Authentifizierung erfordert:

"""
Step 3: Connect to real weather API
"""
import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather-server")

# Configuration
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-mcp-server/1.0"

async def make_nws_request(url: str) -> dict | None:
    """Make a request to the National Weather Service API"""
    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 as e:
            print(f"API request failed: {e}", file=sys.stderr)
            return None

@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """Get weather forecast for a specific location using coordinates"""
    # Step 1: Get the forecast grid endpoint for this location
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return f"Unable to fetch forecast data for coordinates ({latitude}, {longitude})"

    # Step 2: Get the actual forecast data
    properties = points_data.get("properties", {})
    forecast_url = properties.get("forecast")

    if not forecast_url:
        return "Error: Unable to determine forecast URL for this location"

    forecast_data = await make_nws_request(forecast_url)

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

    # Format the first few periods
    periods = forecast_data.get("properties", {}).get("periods", [])

    if not periods:
        return "No forecast periods available for this location"

    # Format the first 3 periods for display
    forecasts = []
    for period in periods[:3]:
        forecast_text = f"""
{period.get('name', 'Unknown Period')}:
Temperature: {period.get('temperature', 'Unknown')}°{period.get('temperatureUnit', 'F')}
Wind: {period.get('windSpeed', 'Unknown')} {period.get('windDirection', '')}
Forecast: {period.get('detailedForecast', 'No detailed forecast available')}
"""
        forecasts.append(forecast_text.strip())

    return f"Forecast for {latitude}, {longitude}:\n" + "\n---\n".join(forecasts)

if __name__ == "__main__":
    import sys
    mcp.run(transport='stdio')

Teste diese Version mit echten Koordinaten (z. B. New York City: 40.7128, -74.0060). Du solltest nun echte Wettervorhersagedaten erhalten!

Schritt 4: Eingabevalidierung und Fehlerbehandlung hinzufügen

Machen wir unseren Server robuster, indem wir eine ordentliche Eingabevalidierung und Fehlerbehandlung hinzufügen:

"""
Step 4: Add input validation and better error handling
"""
import sys
import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather-server")

# Configuration
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-mcp-server/1.0"

async def make_nws_request(url: str) -> dict | None:
    """Make a request to the National Weather Service 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 httpx.TimeoutException:
            print(f"Request timeout for URL: {url}", file=sys.stderr)
            return None
        except httpx.HTTPStatusError as e:
            print(f"HTTP error {e.response.status_code} for URL: {url}", file=sys.stderr)
            return None
        except Exception as e:
            print(f"Unexpected error for URL {url}: {e}", file=sys.stderr)
            return None

@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """Get weather forecast for a specific location using coordinates"""
    # Validate coordinate ranges
    if not (-90 <= latitude <= 90):
        return "Error: Latitude must be between -90 and 90 degrees"

    if not (-180 <= longitude <= 180):
        return "Error: Longitude must be between -180 and 180 degrees"

    # Step 1: Get the forecast grid endpoint for this location
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return f"Unable to fetch forecast data for coordinates ({latitude}, {longitude}). This location may be outside the US or the service may be unavailable."

    # Extract the forecast URL from the points response
    properties = points_data.get("properties", {})
    forecast_url = properties.get("forecast")

    if not forecast_url:
        return "Error: Unable to determine forecast URL for this location"

    # Step 2: Get the actual forecast data
    forecast_data = await make_nws_request(forecast_url)

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

    # Extract and format forecast periods
    forecast_properties = forecast_data.get("properties", {})
    periods = forecast_properties.get("periods", [])

    if not periods:
        return "No forecast periods available for this location"

    # Format the first 3 periods for display
    forecasts = []
    for period in periods[:3]:
        forecast_text = f"""
{period.get('name', 'Unknown Period')}:
Temperature: {period.get('temperature', 'Unknown')}°{period.get('temperatureUnit', 'F')}
Wind: {period.get('windSpeed', 'Unknown')} {period.get('windDirection', '')}
Forecast: {period.get('detailedForecast', 'No detailed forecast available')}
"""
        forecasts.append(forecast_text.strip())

    location_info = f"Forecast for {latitude}, {longitude}:\n"
    return location_info + "\n---\n".join(forecasts)

@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get active weather alerts for a US state"""
    # Validate state code format
    if not state or len(state) != 2:
        return "Error: Please provide a valid two-letter US state code (e.g., 'CA', 'NY', 'TX')"

    state = state.upper()
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"

    data = await make_nws_request(url)

    if not data:
        return f"Unable to fetch weather alerts for {state}. The service may be temporarily unavailable."

    features = data.get("features", [])

    if not features:
        return f"No active weather alerts for {state}."

    # Format alerts for display
    alerts = []
    for feature in features:
        props = feature.get("properties", {})
        event = props.get('event', 'Unknown Event')
        area = props.get('areaDesc', 'Unknown Area')
        severity = props.get('severity', 'Unknown Severity')
        description = props.get('description', 'No description available')

        alert_text = f"""
Event: {event}
Area: {area}
Severity: {severity}
Description: {description[:200]}{'...' if len(description) > 200 else ''}
"""
        alerts.append(alert_text.strip())

    alert_count = len(alerts)
    header = f"Found {alert_count} active weather alert{'s' if alert_count != 1 else ''} for {state}:\n"
    return header + "\n---\n".join(alerts)

if __name__ == "__main__":
    mcp.run(transport='stdio')

Teste nun beide Tools mit verschiedenen Eingaben, auch mit ungültigen, um zu sehen, wie die Fehlerbehandlung funktioniert.

Schritt 5: Resources für kontextbezogene Informationen hinzufügen

Fügen wir Resources hinzu, die kontextbezogene Informationen über Wetterstationen und -zonen bereitstellen:

"""
Step 5: Add resources for contextual weather information
"""
import sys
import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather-server")

# Configuration
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-mcp-server/1.0"

async def make_nws_request(url: str) -> dict | None:
    """Make a request to the National Weather Service 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 httpx.TimeoutException:
            print(f"Request timeout for URL: {url}", file=sys.stderr)
            return None
        except httpx.HTTPStatusError as e:
            print(f"HTTP error {e.response.status_code} for URL: {url}", file=sys.stderr)
            return None
        except Exception as e:
            print(f"Unexpected error for URL {url}: {e}", file=sys.stderr)
            return None

# Tools (same as Step 4)
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """Get weather forecast for a specific location using coordinates"""
    # Validate coordinate ranges
    if not (-90 <= latitude <= 90):
        return "Error: Latitude must be between -90 and 90 degrees"

    if not (-180 <= longitude <= 180):
        return "Error: Longitude must be between -180 and 180 degrees"

    # Step 1: Get the forecast grid endpoint for this location
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)

    if not points_data:
        return f"Unable to fetch forecast data for coordinates ({latitude}, {longitude}). This location may be outside the US or the service may be unavailable."

    # Extract the forecast URL from the points response
    properties = points_data.get("properties", {})
    forecast_url = properties.get("forecast")

    if not forecast_url:
        return "Error: Unable to determine forecast URL for this location"

    # Step 2: Get the actual forecast data
    forecast_data = await make_nws_request(forecast_url)

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

    # Extract and format forecast periods
    forecast_properties = forecast_data.get("properties", {})
    periods = forecast_properties.get("periods", [])

    if not periods:
        return "No forecast periods available for this location"

    # Format the first 3 periods for display
    forecasts = []
    for period in periods[:3]:
        forecast_text = f"""
{period.get('name', 'Unknown Period')}:
Temperature: {period.get('temperature', 'Unknown')}°{period.get('temperatureUnit', 'F')}
Wind: {period.get('windSpeed', 'Unknown')} {period.get('windDirection', '')}
Forecast: {period.get('detailedForecast', 'No detailed forecast available')}
"""
        forecasts.append(forecast_text.strip())

    location_info = f"Forecast for {latitude}, {longitude}:\n"
    return location_info + "\n---\n".join(forecasts)

@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get active weather alerts for a US state"""
    # Validate state code format
    if not state or len(state) != 2:
        return "Error: Please provide a valid two-letter US state code (e.g., 'CA', 'NY', 'TX')"

    state = state.upper()
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"

    data = await make_nws_request(url)

    if not data:
        return f"Unable to fetch weather alerts for {state}. The service may be temporarily unavailable."

    features = data.get("features", [])

    if not features:
        return f"No active weather alerts for {state}."

    # Format alerts for display
    alerts = []
    for feature in features:
        props = feature.get("properties", {})
        event = props.get('event', 'Unknown Event')
        area = props.get('areaDesc', 'Unknown Area')
        severity = props.get('severity', 'Unknown Severity')
        description = props.get('description', 'No description available')

        alert_text = f"""
Event: {event}
Area: {area}
Severity: {severity}
Description: {description[:200]}{'...' if len(description) > 200 else ''}
"""
        alerts.append(alert_text.strip())

    alert_count = len(alerts)
    header = f"Found {alert_count} active weather alert{'s' if alert_count != 1 else ''} for {state}:\n"
    return header + "\n---\n".join(alerts)

# Resources for contextual information
@mcp.resource("weather://stations/{state}")
async def get_weather_stations(state: str) -> str:
    """Get information about weather observation stations in a state"""
    if not state or len(state) != 2:
        return "Error: Please provide a valid two-letter US state code"

    state = state.upper()
    url = f"{NWS_API_BASE}/stations?state={state}"

    data = await make_nws_request(url)

    if not data:
        return f"Unable to fetch weather station information for {state}"

    features = data.get("features", [])

    if not features:
        return f"No weather stations found for {state}"

    stations = []
    for feature in features[:10]:  # Limit to first 10 stations
        props = feature.get("properties", {})
        name = props.get("name", "Unknown Station")
        identifier = props.get("stationIdentifier", "Unknown ID")
        elevation = props.get("elevation", {}).get("value", "Unknown")

        stations.append(f"- {name} ({identifier}) - Elevation: {elevation}m")

    station_count = len(features)
    header = f"Weather stations in {state} (showing first 10 of {station_count}):\n"
    return header + "\n".join(stations)

if __name__ == "__main__":
    mcp.run(transport='stdio')

Im MCP Inspector solltest du nun sowohl Tools als auch Resources sehen. Resources erscheinen in einem eigenen Abschnitt und stellen kontextbezogene Informationen bereit, die KI-Anwendungen nutzen können, um die Wetterdaten besser zu verstehen.

deinen Server mit dem MCP Inspector testen

Bevor du deinen Server in KI-Assistenten wie Claude Desktop integrierst, ist es unerlässlich, ihn mit dem MCP Inspector gründlich zu testen. Der Inspector bietet eine webbasierte Oberfläche zum Testen von MCP-Servern, mit der du überprüfen kannst, ob alle Tools und Resources korrekt funktionieren.

Den MCP Inspector starten

Um deinen Wetterserver mit dem MCP Inspector zu testen, führe den folgenden Befehl aus deinem Projektverzeichnis aus:

# Make sure you're in your project directory and virtual environment is activated
cd weather-mcp-server
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Start the MCP Inspector with your server
python -m mcp dev weather_server.py

Dieser Befehl wird:

1. Deinen Wetterserver im Entwicklungsmodus starten
2. Die Weboberfläche des MCP Inspector öffnen
3. Den Inspector automatisch mit deinem Server verbinden

Du solltest eine Ausgabe ähnlich der folgenden sehen:

Starting MCP Inspector...
Server running at: http://localhost:3000
MCP server connected successfully

Die Oberfläche des MCP Inspector nutzen

Öffne deinen Webbrowser und navigiere zu http://localhost:3000. Die Oberfläche des MCP Inspector bietet mehrere Abschnitte zum Testen deines Servers:

Server-Informationspanel: Zeigt den Namen, die Version und den Verbindungsstatus deines Servers an. Du solltest "weather-server" als verbunden aufgeführt sehen.

Tools-Abschnitt: Listet alle verfügbaren Tools mit ihren Beschreibungen und Parameter-Schemata auf. Für deinen Wetterserver solltest du sehen:

• get_forecast - Wettervorhersage für Koordinaten abrufen
• get_alerts - Aktive Wetterwarnungen für einen US-Bundesstaat abrufen
• analyze_weather_trends - KI-gestützte Wetteranalyse (falls du Schritt 6 implementiert hast)

Resources-Abschnitt: Zeigt verfügbare Resources an, die kontextbezogene Informationen bereitstellen:

• weather://stations/{state} - Informationen zu Wetterstationen für Bundesstaaten

deine Tools testen

Testen wir jedes Tool systematisch:

Das Forecast-Tool testen:

1. Klicke im Inspector auf das Tool get_forecast
2. Gib Testkoordinaten ein:
   • Latitude: 40.7128 (New York City)
   • Longitude: -74.0060
3. Klicke auf "Execute Tool"
4. Überprüfe, ob du eine korrekt formatierte Wettervorhersage mit Temperatur, Wind und detaillierten Vorhersageinformationen erhältst

Das Alerts-Tool testen:

1. Klicke auf das Tool get_alerts
2. Gib einen Bundesstaat-Code ein: CA (Kalifornien)
3. Klicke auf "Execute Tool"
4. Prüfe, ob du entweder aktive Warnungen oder eine "No active alerts"-Meldung erhältst

Eingabevalidierung testen:

1. Versuche ungültige Koordinaten (z. B. latitude: 100, longitude: 200)
2. Versuche ungültige Bundesstaat-Codes (z. B. XYZ oder California)
3. Überprüfe, ob dein Server passende Fehlermeldungen zurückgibt

Resources testen

Die Wetterstationen-Resource testen:

1. Navigiere zum Resources-Abschnitt
2. Suche nach der Resource weather://stations/{state}
3. Klicke darauf und gib einen Bundesstaat-Code wie TX ein
4. Überprüfe, ob du eine Liste von Wetterstationen mit Namen, Kennungen und Höhenangaben erhältst

Server-Logs überwachen

Behalte während des Testens das Terminal im Auge, in dem du den Inspector gestartet hast. Du solltest Log-Meldungen sehen, die Folgendes zeigen:

• Erfolgreiche API-Anfragen an den National Weather Service
• Etwaige Fehlermeldungen oder Warnungen
• Bestätigungen der Tool-Ausführung

Beispielhafte Log-Ausgabe:

INFO: Tool 'get_forecast' called with params: {'latitude': 40.7128, 'longitude': -74.0060}
INFO: API request successful: https://api.weather.gov/points/40.7128,-74.0060
INFO: Forecast data retrieved successfully

Häufige Probleme beheben

Falls beim Testen Probleme auftreten:

Server startet nicht:

• Prüfe, ob alle Abhängigkeiten installiert sind: pip install -r requirements.txt
• Überprüfe, ob deine virtuelle Umgebung aktiviert ist
• Achte auf Syntaxfehler in deinem Code

Tools geben Fehler zurück:

• Überprüfe deine Internetverbindung (der Server muss auf weather.gov zugreifen)
• Stell sicher, dass die API des National Weather Service erreichbar ist
• Prüfe die Fehlermeldungen in den Server-Logs

Keine Daten zurückgegeben:

• Versuche andere Koordinaten (stell sicher, dass sie innerhalb der USA liegen)
• Prüfe, ob die Bundesstaat-Codes gültige zweibuchstabige Abkürzungen sind
• Überprüfe, ob API-Antworten nicht von Firewalls blockiert werden

Das Ausgabeformat validieren

Stell sicher, dass deine Tool-Ausgaben korrekt formatiert sind:

• Wettervorhersagen sollten menschenlesbar sein
• Warninformationen sollten alle relevanten Details enthalten
• Fehlermeldungen sollten klar und handlungsleitend sein
• Alle Antworten sollten gültige Zeichenketten sein (JSON-serialisierbar)

Sobald du deinen Server gründlich mit dem MCP Inspector getestet und bestätigt hast, dass alle Tools und Resources korrekt funktionieren, bist du bereit, ihn in KI-Assistenten wie Claude Desktop zu integrieren.

deinen Server in Claude Desktop registrieren

Nachdem du nun einen funktionierenden MCP-Server hast, konfigurieren wir ihn für die Zusammenarbeit mit Claude Desktop. Dazu bearbeiten wir die Konfigurationsdatei von Claude Desktop, um deinen Server zu registrieren.

Die Konfigurationsdatei finden

Die Konfigurationsdatei von Claude Desktop befindet sich je nach Betriebssystem an unterschiedlichen Pfaden:

macOS:

~/Library/Application Support/Claude/claude_desktop_config.json

Windows:

%APPDATA%\Claude\claude_desktop_config.json

Linux:

~/.config/Claude/claude_desktop_config.json

deinen Server konfigurieren

Erstelle oder bearbeite die Konfigurationsdatei, um deinen Wetterserver einzubinden. Hier die Basiskonfiguration:

{
  "mcpServers": {
    "weather-server": {
      "command": "python",
      "args": ["weather_server.py"],
      "cwd": "/path/to/your/weather-mcp-server"
    }
  }
}

Ersetze /path/to/your/weather-mcp-server durch den tatsächlichen Pfad zu deinem Projektverzeichnis.

Alternative Konfiguration mit virtueller Umgebung

Wenn du explizit den Python-Interpreter deiner virtuellen Umgebung verwenden möchtest:

{
  "mcpServers": {
    "weather-server": {
      "command": "/path/to/your/weather-mcp-server/.venv/bin/python",
      "args": ["weather_server.py"],
      "cwd": "/path/to/your/weather-mcp-server"
    }
  }
}

Unter Windows würde der Pfad so aussehen:

{
  "mcpServers": {
    "weather-server": {
      "command": "C:\\path\\to\\your\\weather-mcp-server\\.venv\\Scripts\\python.exe",
      "args": ["weather_server.py"],
      "cwd": "C:\\path\\to\\your\\weather-mcp-server"
    }
  }
}

Die Integration testen

1. Speichere die Konfigurationsdatei
2. Starte Claude Desktop vollständig neu (beenden und erneut öffnen)
3. Beginne eine neue Unterhaltung
4. Bitte Claude, Wetterinformationen für einen Ort abzurufen

Du solltest sehen, wie Claude deine Wetter-Tools nutzt, um Echtzeit-Wetterinformationen bereitzustellen!

Fehlerbehebung bei der Claude-Desktop-Integration

Falls dein Server nicht in Claude Desktop erscheint:

1. Prüfe die Syntax der Konfigurationsdatei - Verwende einen JSON-Validator, um eine korrekte Formatierung sicherzustellen
2. Überprüfe die Dateipfade - Stell sicher, dass alle Pfade in der Konfiguration absolut und korrekt sind
3. Prüfe die Berechtigungen - Stell sicher, dass Claude Desktop deine Python-Umgebung ausführen kann
4. Prüfe die Logs - Claude Desktop kann Fehlermeldungen in seiner Oberfläche anzeigen
5. Teste zuerst mit dem MCP Inspector - Überprüfe stets, ob dein Server mit dem Inspector funktioniert, bevor du Claude Desktop konfigurierst

Fortgeschrittene Funktionen hinzufügen

Fügen wir nun einige fortgeschrittene Funktionen hinzu, um unseren Server leistungsfähiger und produktionsreifer zu machen.

Schritt 6: MCP-Sampling hinzufügen

MCP-Sampling ermöglicht es deinem Server, KI-Completions vom Client anzufordern, was eine intelligente Analyse von Wetterdaten ermöglicht:

"""
Step 6: Add MCP sampling
"""
import sys
import httpx
from mcp.server.fastmcp import FastMCP, Context
from mcp.server.session import ServerSession

mcp = FastMCP("weather-server")

# Configuration
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-mcp-server/1.0"

async def make_nws_request(url: str) -> dict | None:
    """Make a request to the National Weather Service 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 as e:
            print(f"API request failed: {e}", file=sys.stderr)
            return None

# Previous tools (get_forecast, get_alerts) - same as Step 5
@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """Get weather forecast for a specific location using coordinates"""
    # [Previous implementation from Step 5]
    # ... (keeping this concise for the tutorial)
    pass

@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get active weather alerts for a US state"""
    # [Previous implementation from Step 5]
    # ... (keeping this concise for the tutorial)
    pass

# New AI-powered tool using sampling
@mcp.tool()
async def analyze_weather_trends(
    state: str,
    ctx: Context[ServerSession, None]
) -> str:
    """Analyze weather alert trends for a state using AI-powered analysis"""
    # First, gather current weather alert data
    alerts_data = await get_alerts(state)

    if "No active weather alerts" in alerts_data or "Error:" in alerts_data:
        return f"No weather alerts available for analysis in {state}"

    # Use sampling to analyze the weather data
    analysis_prompt = f"""
    Analyze the following weather alert data for {state} and provide insights about:

    1. The types of weather events currently affecting the region
    2. The severity and geographic distribution of alerts
    3. Potential impacts on daily activities and safety
    4. Any notable patterns or unusual weather conditions

    Weather Alert Data:
    {alerts_data}

    Provide a concise but comprehensive analysis that would be helpful for emergency management and public safety planning.
    """

    try:
        # Request AI analysis through sampling
        response = await ctx.request_sampling(
            messages=[{
                "role": "user",
                "content": {
                    "type": "text",
                    "text": analysis_prompt
                }
            }],
            modelPreferences={
                "intelligencePriority": 0.8,  # High intelligence for analysis
                "speedPriority": 0.4,         # Moderate speed requirement
                "costPriority": 0.3           # Cost is less important for analysis
            },
            systemPrompt="You are a meteorological analyst with expertise in weather pattern analysis and emergency management.",
            maxTokens=500
        )

        return f"Weather Trend Analysis for {state}:\n\n{response.content.text}"

    except Exception as e:
        return f"Unable to generate weather analysis: {str(e)}"

if __name__ == "__main__":
    mcp.run(transport='stdio')

Schritt 7: Authentifizierung für die Produktionsbereitstellung hinzufügen

Für Produktionsbereitstellungen solltest du eine Authentifizierung hinzufügen. So fügst du eine einfache Bearer-Token-Authentifizierung hinzu:

"""
Step 7: Add authentication for production deployment
"""
import os
import sys
import httpx
import jwt
from datetime import datetime
from mcp.server.fastmcp import FastMCP, Context
from mcp.server.session import ServerSession

mcp = FastMCP("weather-server")

# Configuration
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-mcp-server/1.0"
JWT_SECRET = os.getenv("JWT_SECRET", "your-secret-key")

class AuthenticationError(Exception):
    """Custom exception for authentication failures"""
    pass

def verify_bearer_token(token: str) -> dict:
    """Verify and decode a JWT bearer token"""
    try:
        payload = jwt.decode(token, JWT_SECRET, algorithms=["HS256"])

        # Check token expiration
        if datetime.utcnow().timestamp() > payload.get("exp", 0):
            raise AuthenticationError("Token has expired")

        return payload

    except jwt.InvalidTokenError as e:
        raise AuthenticationError(f"Invalid token: {str(e)}")

def require_authentication(func):
    """Decorator for tools that require authentication"""
    async def wrapper(*args, **kwargs):
        # In a real implementation, you'd extract the auth header from the request context
        # For this tutorial, we'll simulate authentication

        # Check if running in authenticated mode
        if os.getenv("REQUIRE_AUTH", "false").lower() == "true":
            # In production, extract token from request headers
            token = os.getenv("AUTH_TOKEN")
            if not token:
                return "Authentication required: Please provide a valid bearer token"

            try:
                user_context = verify_bearer_token(token)
                kwargs['user_context'] = user_context
            except AuthenticationError as e:
                return f"Authentication failed: {str(e)}"

        return await func(*args, **kwargs)

    return wrapper

# Previous API helper function
async def make_nws_request(url: str) -> dict | None:
    """Make a request to the National Weather Service 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 as e:
            print(f"API request failed: {e}", file=sys.stderr)
            return None

# Authenticated tools
@mcp.tool()
@require_authentication
async def get_secure_forecast(
    latitude: float,
    longitude: float,
    user_context: dict = None
) -> str:
    """Get weather forecast with authentication and audit logging"""
    user_id = user_context.get("sub", "unknown") if user_context else "anonymous"
    print(f"Forecast request from user {user_id} for {latitude}, {longitude}", file=sys.stderr)

    # Use the same forecast logic as before
    # [Implementation details omitted for brevity]
    return f"Authenticated forecast for {latitude}, {longitude} (User: {user_id})"

if __name__ == "__main__":
    # Determine transport based on environment
    transport = os.getenv("MCP_TRANSPORT", "stdio")

    if transport == "http":
        # Production HTTP deployment with authentication
        port = int(os.getenv("PORT", 8000))
        host = os.getenv("HOST", "0.0.0.0")

        print(f"Starting secure MCP server on {host}:{port}", file=sys.stderr)
        mcp.run(transport="http", host=host, port=port)
    else:
        # Development STDIO deployment
        print("Starting MCP server in STDIO mode", file=sys.stderr)
        mcp.run(transport="stdio")

Überlegungen zur Produktionsbereitstellung

Wenn du deinen MCP-Server in Produktion bringst, solltest du diese wichtigen Faktoren berücksichtigen:

Umgebungskonfiguration

Verwende Umgebungsvariablen für die Konfiguration:

# .env file for production
MCP_TRANSPORT=http
HOST=0.0.0.0
PORT=8000
REQUIRE_AUTH=true
JWT_SECRET=your-production-secret-key
NWS_API_USER_AGENT=your-production-app/1.0

Docker-Bereitstellung

Erstelle ein Dockerfile für die containerisierte Bereitstellung:

FROM python:3.11-slim

WORKDIR /app

# Copy requirements and install dependencies
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt

# Copy project files
COPY weather_server.py ./

# Expose port
EXPOSE 8000

# Run the server
CMD ["python", "weather_server.py"]

Baue und starte den Docker-Container:

# Build the image
docker build -t weather-mcp-server .

# Run the container
docker run -p 8000:8000 -e MCP_TRANSPORT=http weather-mcp-server

Monitoring und Logging

Implementiere ein ordentliches Logging für die Produktion:

import logging
import sys

# Configure logging for production
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('/var/log/mcp-server.log'),
        logging.StreamHandler(sys.stderr)
    ]
)

logger = logging.getLogger(__name__)

# Use logger throughout your application
logger.info("Server starting up")
logger.error("API request failed", exc_info=True)

Häufige Probleme beheben

STDIO-Logging-Probleme

Das häufigste Problem ist das Schreiben nach stdout in STDIO-Servern:

# Wrong - breaks the protocol
print("Debug message")

# Correct - use stderr
print("Debug message", file=sys.stderr)

# Better - use logging
import logging
logging.basicConfig(stream=sys.stderr)
logger = logging.getLogger(__name__)
logger.info("Debug message")

JSON-Serialisierungsfehler

Stell sicher, dass alle Rückgabewerte von Tools JSON-serialisierbar sind:

# Wrong - returns complex object
@mcp.tool()
def bad_tool():
    return SomeComplexObject()

# Correct - returns string
@mcp.tool()
def good_tool():
    result = SomeComplexObject()
    return str(result)  # or result.to_dict() if available

Authentifizierungsprobleme

Überprüfe bei HTTP-Servern deine Authentifizierungskonfiguration:

# Debug authentication setup
def debug_auth():
    required_vars = ["JWT_SECRET", "AUTH_TOKEN"]
    for var in required_vars:
        if not os.getenv(var):
            print(f"Missing environment variable: {var}", file=sys.stderr)

Nächste Schritte und fortgeschrittene Themen

Da dein Wetter-MCP-Server nun fertig ist, bist du bereit, fortgeschrittenere Muster zu erkunden:

deinen Server erweitern

Erwäge das Hinzufügen dieser Funktionen:

• Historische Wetterdaten aus zusätzlichen APIs
• Wetterkarten-Integration mit Bild-Resources
• Echtzeit-Warnungen über WebSocket-Verbindungen
• Machine-Learning-Vorhersagen mit Sampling zur Analyse

Fortgeschrittene Architekturmuster

Erkunde diese Muster für komplexe Bereitstellungen:

• Multi-Server-Architekturen mit spezialisierten Domänen
• Server-Komposition durch Kombination mehrerer MCP-Server
• Verteilte Bereitstellungen über Cloud-Regionen hinweg
• Microservice-Integration in bestehende Systeme

Referenzen

[1] Anthropics Einführung in das Model Context Protocol

[2] Dokumentation des Model Context Protocol

[3] Model Context Protocol Python SDK

[4] Offizieller Leitfaden zum Bau von MCP-Servern

[5] Überblick über die MCP-Architektur

[6] Dokumentation zum MCP-Sampling

[7] MCP-Autorisierungsspezifikation