From REST API to MCP Server

REST-API zu MCP-Server in Python: Schritt für Schritt

event

Einführung in MCP-APIs

Stell dir vor: du hast Monate damit verbracht, solide REST-APIs zu bauen. Deine Endpunkte funktionieren einwandfrei, deine Dokumentation ist gründlich und alles läuft reibungslos. Und plötzlich reden alle über KI-Agenten und LLMs, und du fragst dich: "Wie bringe ich meine APIs dazu, mit diesen KI-Anwendungen zusammenzuarbeiten, ohne von vorne anzufangen?"

Wenn du zustimmend nickst, bist du nicht allein. Genau vor dieser Herausforderung stehen gerade Tausende von Entwicklern.

Die gute Nachricht: du musst deine Arbeit nicht wegwerfen. Das Model Context Protocol (MCP) ist wie ein universeller Übersetzer, der deine bestehenden REST-APIs fließend "KI" sprechen lässt. Betrachte es als einen intelligenten Adapter für deine APIs – sie tun weiterhin das, was sie am besten können, aber jetzt können KI-Anwendungen sie mühelos verstehen und nutzen.

In dieser Anleitung bauen wir Schritt für Schritt einen MCP-Server auf, beginnen einfach und fügen nach und nach Funktionen hinzu. Du siehst genau, wie jedes Teil ineinandergreift, und am Ende hast du einen funktionierenden MCP-Server, der deine REST-API umhüllt und sich nahtlos mit KI-Clients wie Claude Desktop verbindet.

REST-APIs vs. MCP-Server

Beginnen wir mit dem, was du bereits kennst. Eine REST-API ist eine Menge von HTTP-Endpunkten, die strukturierte Daten annehmen und zurückgeben. Jeder Endpunkt hat einen Pfad, eine Methode (GET, POST usw.) und ein Request-/Response-Schema. Deine Clients senden HTTP-Anfragen und erhalten JSON-Antworten zurück.

Ein MCP-Server ist anders. Statt HTTP-Endpunkten stellt er "Tools" und "Ressourcen" bereit, die KI-Anwendungen nutzen können. Tools dienen für Aktionen, die Daten verändern (etwa das Anlegen eines Benutzers), während Ressourcen einen schreibgeschützten Zugriff auf Informationen bieten (etwa das Abrufen von Benutzerdetails).

Die zentrale Erkenntnis ist folgende: deine REST-API wird zum Motor, und der MCP-Server wird zum Übersetzer. Der MCP-Server empfängt Anfragen von KI-Anwendungen, übersetzt sie in REST-API-Aufrufe und formatiert die Antworten anschließend so, dass KI-Anwendungen sie verstehen können.

So werden die Komponenten einander zugeordnet:

REST-API-Komponente MCP-Server-Entsprechung Zweck
POST/PUT/DELETE-Endpunkte Tools Aktionen, die Daten verändern
GET-Endpunkte Ressourcen Schreibgeschützter Datenzugriff
Query-Parameter Tool-Parameter Eingabefelder für Tools
Authentifizierungs-Header Umgebungsvariablen Sichere Handhabung von Zugangsdaten

Voraussetzungen

Bevor wir mit dem Aufbau beginnen, stell sicher, dass du Folgendes hast:

  1. Eine funktionierende REST-API mit dokumentierten Endpunkten (wir verwenden als Beispiel eine einfache Benutzerverwaltungs-API)
  2. Python 3.8+ auf deinem System installiert
  3. Grundkenntnisse in JSON und API-Konzepten
  4. Einen API-Schlüssel oder eine Authentifizierungsmethode für deine REST-API

Für unsere Beispiele gehen wir davon aus, dass du eine REST-API mit diesen Endpunkten hast:

  • GET /users/{id} - Benutzerdetails abrufen
  • POST /users - Einen neuen Benutzer anlegen
  • PUT /users/{id} - Benutzerinformationen aktualisieren
  • DELETE /users/{id} - Einen Benutzer löschen

Schritt 1: Einrichten deiner Entwicklungsumgebung

Beginnen wir damit, ein neues Projekt zu erstellen und die notwendigen Abhängigkeiten zu installieren.

# Create a new directory for your MCP server
mkdir my-mcp-server
cd my-mcp-server

# Create a virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install the MCP SDK and dependencies
pip install mcp httpx python-dotenv

Erstelle eine .env-Datei, um deine API-Zugangsdaten zu speichern:

# .env
API_BASE_URL=https://your-api.example.com
API_KEY=your-api-key-here

Schritt 2: Erstellen deines ersten MCP-Tools

Beginnen wir mit dem einfachsten möglichen MCP-Server. Wir erstellen ein einzelnes Tool, das Benutzerinformationen aus deiner REST-API abruft.

Erstelle eine Datei namens server.py:

import os
import httpx
from mcp.server.fastmcp import FastMCP

# Load environment variables
API_BASE_URL = os.getenv("API_BASE_URL")
API_KEY = os.getenv("API_KEY")

# Initialize the MCP server
mcp = FastMCP("user-api-server")

@mcp.tool()
async def get_user(user_id: str) -> dict:
    """Get user details by ID from the REST API."""

    # Make the REST API call
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{API_BASE_URL}/users/{user_id}",
            headers={"Authorization": f"Bearer {API_KEY}"}
        )

        if response.status_code == 200:
            return response.json()
        else:
            return {"error": f"Failed to get user: {response.status_code}"}

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

Das war's! Du hast soeben deinen ersten MCP-Server erstellt. Testen wir ihn:

python server.py

Der Server startet und wartet auf Eingaben. Du kannst ihn mit dem MCP Inspector testen (darauf gehen wir im Abschnitt zum Testen ein).

Schritt 3: Fehlerbehandlung und Validierung hinzufügen

Unsere erste Version funktioniert, ist aber nicht besonders robust. Fügen wir eine ordentliche Fehlerbehandlung und Eingabevalidierung hinzu:

async def make_api_request(method: str, endpoint: str, data: dict = None) -> Dict[str, Any]:
    """Helper function to make REST API requests with proper error handling."""

    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            if method == "GET":
                response = await client.get(f"{API_BASE_URL}{endpoint}", headers=headers)
            elif method == "POST":
                response = await client.post(f"{API_BASE_URL}{endpoint}", headers=headers, json=data)
            elif method == "PUT":
                response = await client.put(f"{API_BASE_URL}{endpoint}", headers=headers, json=data)
            elif method == "DELETE":
                response = await client.delete(f"{API_BASE_URL}{endpoint}", headers=headers)

            response.raise_for_status()

            # Handle empty responses (common for DELETE operations)
            if response.status_code == 204 or not response.content:
                return {"success": True, "message": "Operation completed successfully"}

            return response.json()

        except httpx.HTTPStatusError as e:
            return {"error": f"HTTP {e.response.status_code}: {e.response.text}"}
        except httpx.TimeoutException:
            return {"error": "Request timeout - API took too long to respond"}
        except Exception as e:
            return {"error": f"Request failed: {str(e)}"}

Jetzt behandelt unser Server Fehler elegant und validiert Eingaben ordnungsgemäß.

Schritt 4: Weitere Tools hinzufügen

Erweitern wir unseren Server, um das Anlegen und Aktualisieren von Benutzern zu ermöglichen. Wir fügen diese Tools nacheinander hinzu:

@mcp.tool()
async def create_user(name: str, email: str, role: str = "user") -> dict:
    """Create a new user account.

    Args:
        name: User's full name
        email: User's email address
        role: User role (defaults to 'user')
    """

    # Validate inputs
    if not name or not name.strip():
        return {"error": "Name is required"}

    if not email or "@" not in email:
        return {"error": "Valid email is required"}

    user_data = {
        "name": name.strip(),
        "email": email.lower().strip(),
        "role": role
    }

    result = await make_api_request("POST", "/users", data=user_data)

    if "error" in result:
        return {"error": f"Failed to create user: {result['error']}"}

    return {
        "success": True,
        "message": f"User '{name}' created successfully",
        "user": result
    }

@mcp.tool()
async def update_user(user_id: str, name: str = None, email: str = None, role: str = None) -> dict:
    """Update an existing user's information.

    Args:
        user_id: ID of the user to update
        name: New name (optional)
        email: New email (optional)
        role: New role (optional)
    """

    if not user_id or not user_id.strip():
        return {"error": "User ID is required"}

    if not any([name, email, role]):
        return {"error": "At least one field (name, email, or role) must be provided"}

    # Build update data
    update_data = {}
    if name:
        update_data["name"] = name.strip()
    if email:
        if "@" not in email:
            return {"error": "Valid email is required"}
        update_data["email"] = email.lower().strip()
    if role:
        update_data["role"] = role

    result = await make_api_request("PUT", f"/users/{user_id.strip()}", data=update_data)

    if "error" in result:
        return {"error": f"Failed to update user: {result['error']}"}

    return {
        "success": True,
        "message": f"User {user_id} updated successfully",
        "user": result
    }

Schritt 5: Ressourcen für schreibgeschützte Daten hinzufügen

Tools eignen sich hervorragend für Aktionen, aber manchmal müssen KI-Anwendungen einfach nur Daten lesen. Genau dafür sind Ressourcen da. Fügen wir eine Ressource zum Abrufen von Benutzerinformationen hinzu:

@mcp.resource("user://{user_id}")
async def get_user_resource(user_id: str) -> str:
    """Get detailed user information as a resource.

    This provides read-only access to user data in a format
    that's easy for AI applications to understand.
    """

    result = await make_api_request("GET", f"/users/{user_id}")

    if "error" in result:
        return f"Error retrieving user {user_id}: {result['error']}"

    user = result

    # Format the data for AI consumption
    formatted_output = f"""
User Profile for ID {user_id}:
- Name: {user.get('name', 'Not specified')}
- Email: {user.get('email', 'Not specified')}
- Role: {user.get('role', 'Not specified')}
- Status: {user.get('status', 'Active')}
- Created: {user.get('created_at', 'Unknown')}
- Last Updated: {user.get('updated_at', 'Unknown')}
"""

    return formatted_output.strip()

@mcp.resource("users://list")
async def list_users_resource() -> str:
    """Get a list of all users in the system."""

    result = await make_api_request("GET", "/users")

    if "error" in result:
        return f"Error retrieving users: {result['error']}"

    users = result.get("users", []) if isinstance(result, dict) else result

    if not users:
        return "No users found in the system."

    # Format the user list for AI consumption
    formatted_output = f"User Directory ({len(users)} users):\n\n"

    for user in users:
        formatted_output += f"• {user.get('name', 'Unknown')} ({user.get('email', 'No email')})\n"
        formatted_output += f"  ID: {user.get('id', 'Unknown')}, Role: {user.get('role', 'Unknown')}\n\n"

    return formatted_output.strip()

Vollständiger Code für den MCP-Server

Nachdem wir unseren MCP-Server nun Schritt für Schritt aufgebaut haben, folgt hier der vollständige, produktionsreife Code, der alle besprochenen Bausteine vereint. Du kannst diese gesamte Datei kopieren und sofort ausführen:

import os
import httpx
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv
from typing import Dict, Any, Optional

# Load environment variables from .env file
load_dotenv()

# Load environment variables
API_BASE_URL = os.getenv("API_BASE_URL")
API_KEY = os.getenv("API_KEY")

if not API_BASE_URL or not API_KEY:
    raise ValueError("API_BASE_URL and API_KEY environment variables are required")

# Initialize the MCP server
mcp = FastMCP("user-api-server")

async def make_api_request(method: str, endpoint: str, data: dict = None) -> Dict[str, Any]:
    """Helper function to make REST API requests with proper error handling."""

    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            if method == "GET":
                response = await client.get(f"{API_BASE_URL}{endpoint}", headers=headers)
            elif method == "POST":
                response = await client.post(f"{API_BASE_URL}{endpoint}", headers=headers, json=data)
            elif method == "PUT":
                response = await client.put(f"{API_BASE_URL}{endpoint}", headers=headers, json=data)
            elif method == "DELETE":
                response = await client.delete(f"{API_BASE_URL}{endpoint}", headers=headers)

            response.raise_for_status()

            # Handle empty responses (common for DELETE operations)
            if response.status_code == 204 or not response.content:
                return {"success": True, "message": "Operation completed successfully"}

            return response.json()

        except httpx.HTTPStatusError as e:
            return {"error": f"HTTP {e.response.status_code}: {e.response.text}"}
        except httpx.TimeoutException:
            return {"error": "Request timeout - API took too long to respond"}
        except Exception as e:
            return {"error": f"Request failed: {str(e)}"}

# TOOLS - for actions that modify data
@mcp.tool()
async def get_user(user_id: str) -> dict:
    """Get user details by ID from the REST API.

    Args:
        user_id: The ID of the user to retrieve
    """

    # Validate input
    if not user_id or not user_id.strip():
        return {"error": "User ID is required"}

    result = await make_api_request("GET", f"/users/{user_id.strip()}")

    if "error" in result:
        return {"error": f"Failed to get user: {result['error']}"}

    return {
        "success": True,
        "user": result
    }

@mcp.tool()
async def create_user(name: str, email: str, role: str = "user") -> dict:
    """Create a new user account.

    Args:
        name: User's full name
        email: User's email address
        role: User role (defaults to 'user')
    """

    # Validate inputs
    if not name or not name.strip():
        return {"error": "Name is required"}

    if not email or "@" not in email:
        return {"error": "Valid email is required"}

    user_data = {
        "name": name.strip(),
        "email": email.lower().strip(),
        "role": role
    }

    result = await make_api_request("POST", "/users", data=user_data)

    if "error" in result:
        return {"error": f"Failed to create user: {result['error']}"}

    return {
        "success": True,
        "message": f"User '{name}' created successfully",
        "user": result
    }

@mcp.tool()
async def update_user(user_id: str, name: Optional[str] = None, email: Optional[str] = None, role: Optional[str] = None) -> dict:
    """Update an existing user's information.

    Args:
        user_id: ID of the user to update
        name: New name (optional)
        email: New email (optional)
        role: New role (optional)
    """

    if not user_id or not user_id.strip():
        return {"error": "User ID is required"}

    if not any([name, email, role]):
        return {"error": "At least one field (name, email, or role) must be provided"}

    # Build update data
    update_data = {}
    if name:
        update_data["name"] = name.strip()
    if email:
        if "@" not in email:
            return {"error": "Valid email is required"}
        update_data["email"] = email.lower().strip()
    if role:
        update_data["role"] = role

    result = await make_api_request("PUT", f"/users/{user_id.strip()}", data=update_data)

    if "error" in result:
        return {"error": f"Failed to update user: {result['error']}"}

    return {
        "success": True,
        "message": f"User {user_id} updated successfully",
        "user": result
    }

@mcp.tool()
async def delete_user(user_id: str) -> dict:
    """Delete a user account.

    Args:
        user_id: ID of the user to delete
    """

    if not user_id or not user_id.strip():
        return {"error": "User ID is required"}

    result = await make_api_request("DELETE", f"/users/{user_id.strip()}")

    if "error" in result:
        return {"error": f"Failed to delete user: {result['error']}"}

    return {
        "success": True,
        "message": f"User {user_id} deleted successfully"
    }

# RESOURCES - for read-only data access optimized for AI consumption
@mcp.resource("user://{user_id}")
async def get_user_resource(user_id: str) -> str:
    """Get detailed user information as a resource.

    This provides read-only access to user data in a format
    that's easy for AI applications to understand.
    """

    result = await make_api_request("GET", f"/users/{user_id}")

    if "error" in result:
        return f"Error retrieving user {user_id}: {result['error']}"

    user = result

    # Format the data for AI consumption
    formatted_output = f"""
User Profile for ID {user_id}:
- Name: {user.get('name', 'Not specified')}
- Email: {user.get('email', 'Not specified')}
- Role: {user.get('role', 'Not specified')}
- Status: {user.get('status', 'Active')}
- Created: {user.get('created_at', 'Unknown')}
- Last Updated: {user.get('updated_at', 'Unknown')}
"""

    return formatted_output.strip()

@mcp.resource("users://list")
async def list_users_resource() -> str:
    """Get a list of all users in the system."""

    result = await make_api_request("GET", "/users")

    if "error" in result:
        return f"Error retrieving users: {result['error']}"

    users = result.get("users", []) if isinstance(result, dict) else result

    if not users:
        return "No users found in the system."

    # Format the user list for AI consumption
    formatted_output = f"User Directory ({len(users)} users):\n\n"

    for user in users:
        formatted_output += f"• {user.get('name', 'Unknown')} ({user.get('email', 'No email')})\n"
        formatted_output += f"  ID: {user.get('id', 'Unknown')}, Role: {user.get('role', 'Unknown')}\n\n"

    return formatted_output.strip()

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

Speichere dies als server.py, stell sicher, dass deine .env-Datei mit deinen API-Zugangsdaten konfiguriert ist, und führe aus:

python server.py

Schritt 6: Testen deines MCP-Servers

Bevor wir uns mit KI-Clients verbinden, testen wir unseren Server mit dem MCP Inspector. Installiere ihn zunächst:

npm install -g @modelcontextprotocol/inspector

Teste dann deinen Server:

mcp-inspector python server.py

Dadurch öffnet sich eine Weboberfläche, auf der du Folgendes kannst:

  • Alle deine Tools und Ressourcen ansehen
  • Jedes Tool mit verschiedenen Eingaben testen
  • Die Antworten betrachten
  • Probleme debuggen

Versuche, das Tool get_user mit einer Benutzer-ID aufzurufen oder das Tool create_user mit Name- und E-Mail-Parametern. Du solltest sehen, wie die tatsächlichen REST-API-Aufrufe erfolgen und die Antworten für die KI-Nutzung formatiert werden.

Schritt 7: Verbindung mit Claude Desktop herstellen

Nun zum spannenden Teil – der Verbindung deines MCP-Servers mit Claude Desktop, sodass du ihn tatsächlich mit einer KI-Anwendung nutzen kannst.

Installation von Claude Desktop

Lade zunächst Claude Desktop herunter und installiere es, falls du das noch nicht getan hast.

Konfiguration deines MCP-Servers

Claude Desktop sucht in einer bestimmten Datei nach MCP-Server-Konfigurationen. Der Speicherort hängt von deinem Betriebssystem ab:

Unter macOS:

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

Unter Windows:

%APPDATA%\Claude\claude_desktop_config.json

Unter Linux:

~/.config/Claude/claude_desktop_config.json

Erstelle diese Datei, falls sie nicht existiert, und füge deine MCP-Server-Konfiguration hinzu:

{
  "mcpServers": {
    "user-api-server": {
      "command": "python",
      "args": ["/path/to/your/server.py"],
      "env": {
        "API_BASE_URL": "https://your-api.example.com",
        "API_KEY": "your-api-key-here"
      }
    }
  }
}

Wichtige Hinweise:

  • Ersetze /path/to/your/server.py durch den tatsächlichen absoluten Pfad zu deiner Serverdatei
  • Ersetze die Umgebungsvariablen durch deine tatsächliche API-URL und deinen Schlüssel
  • Stell sicher, dass Python in deinem System-PATH liegt, oder verwende den vollständigen Pfad zu deiner Python-Ausführbaren

Testen der Verbindung

  1. Starte Claude Desktop neu, nachdem du die Konfigurationsdatei gespeichert hast
  2. Öffne eine neue Unterhaltung in Claude Desktop
  3. Achte auf die MCP-Anzeige – du solltest ein kleines Tool-Symbol oder einen Indikator sehen, der anzeigt, dass dein MCP-Server verbunden ist
  4. Teste deine Tools, indem du Claude bittest, mit deiner API zu interagieren:

Probiere diese Beispiel-Prompts aus:

  • "Kannst du Informationen über die Benutzer-ID 123 abrufen?"
  • "Lege einen neuen Benutzer namens John Doe mit der E-Mail john@example.com an"
  • "Aktualisiere Benutzer 456, um dessen Rolle auf admin zu ändern"

Claude sollte nun in der Lage sein, deine MCP-Tools aufzurufen und mit den Daten deiner REST-API zu arbeiten!

Fehlerbehebung bei Verbindungsproblemen

Falls Claude Desktop keine Verbindung zu deinem Server herstellt:

  1. Prüfe die Logs – Claude Desktop zeigt Verbindungsfehler üblicherweise in seiner Entwicklerkonsole an
  2. Überprüfe die Dateipfade – Stell sicher, dass alle Pfade in der Konfiguration absolut und korrekt sind
  3. Teste deinen Server unabhängig – Führe python server.py direkt aus, um sicherzustellen, dass er fehlerfrei startet
  4. Prüfe die Umgebungsvariablen – Stell sicher, dass deine API-Zugangsdaten korrekt gesetzt sind
  5. Starte Claude Desktop neu – Konfigurationsänderungen erfordern einen Neustart

Schritt 8: Erweiterte Funktionen und bewährte Praktiken

Authentifizierungs-Handling hinzufügen

Für den Produktiveinsatz benötigst du ein ausgefeilteres Authentifizierungs-Handling:

import os
from datetime import datetime, timedelta

class APIAuthenticator:
    def __init__(self):
        self.api_key = os.getenv("API_KEY")
        self.token = None
        self.token_expires = None

    async def get_headers(self):
        """Get authentication headers, refreshing token if needed."""

        # For simple API key auth
        if self.api_key:
            return {"Authorization": f"Bearer {self.api_key}"}

        # For OAuth or token-based auth (implement token refresh logic here)
        if self.token and self.token_expires > datetime.now():
            return {"Authorization": f"Bearer {self.token}"}

        # Refresh token logic would go here
        await self.refresh_token()
        return {"Authorization": f"Bearer {self.token}"}

    async def refresh_token(self):
        """Implement your token refresh logic here."""
        pass

# Use the authenticator in your API requests
auth = APIAuthenticator()

async def make_authenticated_request(method: str, endpoint: str, data: dict = None):
    headers = await auth.get_headers()
    headers["Content-Type"] = "application/json"

    # Rest of your request logic...

Logging und Monitoring hinzufügen

Füge umfassendes Logging hinzu, um Debugging und Monitoring zu erleichtern:

import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

async def make_api_request(method: str, endpoint: str, data: dict = None) -> Dict[str, Any]:
    """Enhanced API request function with logging."""

    logger.info(f"Making {method} request to {endpoint}")

    try:
        # Your existing request logic...

        logger.info(f"Request successful: {method} {endpoint}")
        return response.json()

    except Exception as e:
        logger.error(f"Request failed: {method} {endpoint} - {str(e)}")
        return {"error": f"Request failed: {str(e)}"}

Rate Limits handhaben

Füge intelligentes Rate Limiting und Retry-Logik hinzu:

import asyncio
from typing import Optional

class RateLimitHandler:
    def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay

    async def make_request_with_retry(self, request_func, *args, **kwargs):
        """Make a request with exponential backoff retry logic."""

        for attempt in range(self.max_retries + 1):
            try:
                response = await request_func(*args, **kwargs)

                # Check for rate limiting
                if hasattr(response, 'status_code') and response.status_code == 429:
                    if attempt < self.max_retries:
                        delay = self.base_delay * (2 ** attempt)
                        logger.warning(f"Rate limited, retrying in {delay} seconds...")
                        await asyncio.sleep(delay)
                        continue

                return response

            except Exception as e:
                if attempt < self.max_retries:
                    delay = self.base_delay * (2 ** attempt)
                    logger.warning(f"Request failed, retrying in {delay} seconds: {str(e)}")
                    await asyncio.sleep(delay)
                    continue
                raise e

        raise Exception("Max retries exceeded")

Produktionsreifer, vollständiger Code: Alle erweiterten Funktionen enthalten

Hier ist der vollständige, produktionsreife MCP-Server, der alle erweiterten Funktionen aus Schritt 8 integriert. Diese Version enthält Authentifizierungs-Handling, umfassendes Logging und intelligentes Rate Limiting:

import os
import httpx
import asyncio
import logging
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv
from typing import Dict, Any, Optional
from datetime import datetime, timedelta

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Load environment variables from .env file
load_dotenv()

# Load environment variables
API_BASE_URL = os.getenv("API_BASE_URL")
API_KEY = os.getenv("API_KEY")

if not API_BASE_URL or not API_KEY:
    raise ValueError("API_BASE_URL and API_KEY environment variables are required")

# Initialize the MCP server
mcp = FastMCP("user-api-server")

class APIAuthenticator:
    """Handle API authentication with token refresh capabilities."""

    def __init__(self):
        self.api_key = os.getenv("API_KEY")
        self.token = None
        self.token_expires = None

    async def get_headers(self):
        """Get authentication headers, refreshing token if needed."""

        # For simple API key auth
        if self.api_key:
            return {"Authorization": f"Bearer {self.api_key}"}

        # For OAuth or token-based auth (implement token refresh logic here)
        if self.token and self.token_expires > datetime.now():
            return {"Authorization": f"Bearer {self.token}"}

        # Refresh token logic would go here
        await self.refresh_token()
        return {"Authorization": f"Bearer {self.token}"}

    async def refresh_token(self):
        """Implement your token refresh logic here."""
        # This is where you'd implement OAuth token refresh
        # For now, we'll just use the API key
        pass

class RateLimitHandler:
    """Handle rate limiting with exponential backoff retry logic."""

    def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay

    async def make_request_with_retry(self, request_func, *args, **kwargs):
        """Make a request with exponential backoff retry logic."""

        for attempt in range(self.max_retries + 1):
            try:
                response = await request_func(*args, **kwargs)

                # Check for rate limiting
                if hasattr(response, 'status_code') and response.status_code == 429:
                    if attempt < self.max_retries:
                        delay = self.base_delay * (2 ** attempt)
                        logger.warning(f"Rate limited, retrying in {delay} seconds...")
                        await asyncio.sleep(delay)
                        continue

                return response

            except Exception as e:
                if attempt < self.max_retries:
                    delay = self.base_delay * (2 ** attempt)
                    logger.warning(f"Request failed, retrying in {delay} seconds: {str(e)}")
                    await asyncio.sleep(delay)
                    continue
                raise e

        raise Exception("Max retries exceeded")

# Initialize authentication and rate limiting
auth = APIAuthenticator()
rate_limiter = RateLimitHandler()

async def make_api_request(method: str, endpoint: str, data: dict = None) -> Dict[str, Any]:
    """Enhanced API request function with authentication, logging, and rate limiting."""

    logger.info(f"Making {method} request to {API_BASE_URL}{endpoint}")

    async def _make_request():
        headers = await auth.get_headers()
        headers["Content-Type"] = "application/json"

        async with httpx.AsyncClient(timeout=30.0) as client:
            if method == "GET":
                return await client.get(f"{API_BASE_URL}{endpoint}", headers=headers)
            elif method == "POST":
                return await client.post(f"{API_BASE_URL}{endpoint}", headers=headers, json=data)
            elif method == "PUT":
                return await client.put(f"{API_BASE_URL}{endpoint}", headers=headers, json=data)
            elif method == "DELETE":
                return await client.delete(f"{API_BASE_URL}{endpoint}", headers=headers)

    try:
        response = await rate_limiter.make_request_with_retry(_make_request)
        response.raise_for_status()

        # Handle empty responses (common for DELETE operations)
        if response.status_code == 204 or not response.content:
            logger.info(f"Request successful: {method} {endpoint}")
            return {"success": True, "message": "Operation completed successfully"}

        logger.info(f"Request successful: {method} {endpoint}")
        return response.json()

    except httpx.HTTPStatusError as e:
        logger.error(f"HTTP error: {e.response.status_code} for {method} {endpoint}")
        return {"error": f"HTTP {e.response.status_code}: {e.response.text}"}
    except httpx.TimeoutException:
        logger.error(f"Timeout error for {method} {endpoint}")
        return {"error": "Request timeout - API took too long to respond"}
    except Exception as e:
        logger.error(f"Unexpected error for {method} {endpoint}: {str(e)}")
        return {"error": f"Request failed: {str(e)}"}

# TOOLS - for actions that modify data
@mcp.tool()
async def get_user(user_id: str) -> dict:
    """Get user details by ID from the REST API.

    Args:
        user_id: The ID of the user to retrieve
    """

    # Validate input
    if not user_id or not user_id.strip():
        return {"error": "User ID is required"}

    result = await make_api_request("GET", f"/users/{user_id.strip()}")

    if "error" in result:
        return {"error": f"Failed to get user: {result['error']}"}

    return {
        "success": True,
        "user": result
    }

@mcp.tool()
async def create_user(name: str, email: str, role: str = "user") -> dict:
    """Create a new user account.

    Args:
        name: User's full name
        email: User's email address
        role: User role (defaults to 'user')
    """

    # Validate inputs
    if not name or not name.strip():
        return {"error": "Name is required"}

    if not email or "@" not in email:
        return {"error": "Valid email is required"}

    user_data = {
        "name": name.strip(),
        "email": email.lower().strip(),
        "role": role
    }

    result = await make_api_request("POST", "/users", data=user_data)

    if "error" in result:
        return {"error": f"Failed to create user: {result['error']}"}

    return {
        "success": True,
        "message": f"User '{name}' created successfully",
        "user": result
    }

@mcp.tool()
async def update_user(user_id: str, name: Optional[str] = None, email: Optional[str] = None, role: Optional[str] = None) -> dict:
    """Update an existing user's information.

    Args:
        user_id: ID of the user to update
        name: New name (optional)
        email: New email (optional)
        role: New role (optional)
    """

    if not user_id or not user_id.strip():
        return {"error": "User ID is required"}

    if not any([name, email, role]):
        return {"error": "At least one field (name, email, or role) must be provided"}

    # Build update data
    update_data = {}
    if name:
        update_data["name"] = name.strip()
    if email:
        if "@" not in email:
            return {"error": "Valid email is required"}
        update_data["email"] = email.lower().strip()
    if role:
        update_data["role"] = role

    result = await make_api_request("PUT", f"/users/{user_id.strip()}", data=update_data)

    if "error" in result:
        return {"error": f"Failed to update user: {result['error']}"}

    return {
        "success": True,
        "message": f"User {user_id} updated successfully",
        "user": result
    }

@mcp.tool()
async def delete_user(user_id: str) -> dict:
    """Delete a user account.

    Args:
        user_id: ID of the user to delete
    """

    if not user_id or not user_id.strip():
        return {"error": "User ID is required"}

    result = await make_api_request("DELETE", f"/users/{user_id.strip()}")

    if "error" in result:
        return {"error": f"Failed to delete user: {result['error']}"}

    return {
        "success": True,
        "message": f"User {user_id} deleted successfully"
    }

# RESOURCES - for read-only data access optimized for AI consumption
@mcp.resource("user://{user_id}")
async def get_user_resource(user_id: str) -> str:
    """Get detailed user information as a resource.

    This provides read-only access to user data in a format
    that's easy for AI applications to understand.
    """

    result = await make_api_request("GET", f"/users/{user_id}")

    if "error" in result:
        return f"Error retrieving user {user_id}: {result['error']}"

    user = result

    # Format the data for AI consumption
    formatted_output = f"""
User Profile for ID {user_id}:
- Name: {user.get('name', 'Not specified')}
- Email: {user.get('email', 'Not specified')}
- Role: {user.get('role', 'Not specified')}
- Status: {user.get('status', 'Active')}
- Created: {user.get('created_at', 'Unknown')}
- Last Updated: {user.get('updated_at', 'Unknown')}
"""

    return formatted_output.strip()

@mcp.resource("users://list")
async def list_users_resource() -> str:
    """Get a list of all users in the system."""

    result = await make_api_request("GET", "/users")

    if "error" in result:
        return f"Error retrieving users: {result['error']}"

    users = result.get("users", []) if isinstance(result, dict) else result

    if not users:
        return "No users found in the system."

    # Format the user list for AI consumption
    formatted_output = f"User Directory ({len(users)} users):\n\n"

    for user in users:
        formatted_output += f"• {user.get('name', 'Unknown')} ({user.get('email', 'No email')})\n"
        formatted_output += f"  ID: {user.get('id', 'Unknown')}, Role: {user.get('role', 'Unknown')}\n\n"

    return formatted_output.strip()

if __name__ == "__main__":
    logger.info("Starting production-ready MCP server with advanced features...")
    mcp.run(transport="stdio")

Speichere dies als server_production.py und führe es aus mit:

python server_production.py

Häufige Fallstricke und wie du sie vermeidest

Schema-Diskrepanzen

Problem: deine MCP-Tool-Schemata stimmen nicht mit dem überein, was deine REST-API erwartet.

Lösung: Teste deine Schemata stets gegen echte API-Aufrufe. Verwende Tools wie Postman, um das Verhalten deiner API zuerst zu überprüfen, und stell dann sicher, dass deine MCP-Schemata exakt übereinstimmen.

Komplexität der Authentifizierung

Problem: OAuth-Abläufe und Token-Erneuerung können in lang laufenden MCP-Servern komplex werden.

Lösung: Baue eine modulare Authentifizierung, die die Token-Erneuerung automatisch handhabt. Teste deine Authentifizierungslogik getrennt von deinen MCP-Tools.

Annahmen zur Performance

Problem: KI-Anwendungen können ganz andere Verkehrsmuster erzeugen als herkömmliche Clients.

Lösung: Implementiere ordentliches Rate Limiting, Connection Pooling und Monitoring. Teste mit realistischen KI-Nutzungsmustern.

Qualität der Fehlermeldungen

Problem: Technische Fehlermeldungen deiner REST-API sind für KI-Anwendungen oder Endnutzer nicht hilfreich.

Lösung: Wandle technische Fehler in benutzerfreundliche Erklärungen um, die KI-Anwendungen nutzen können, um Nutzern zu helfen, Probleme zu verstehen und zu beheben.

Fazit

Du hast soeben einen vollständigen MCP-Server gebaut, der deine REST-API mit KI-Anwendungen verbindet. Du hast mit einem einfachen Tool begonnen, Fehlerbehandlung hinzugefügt, die Funktionalität erweitert und ihn mit Claude Desktop verbunden. Deine APIs sind jetzt KI-bereit!

Der schrittweise Ansatz, den wir hier verwendet haben – einfach anfangen und die Funktionalität ausbauen – ist der Schlüssel zur erfolgreichen Entwicklung von MCP-Servern. Du kannst dieselben Muster auf jede REST-API anwenden, ob es sich um eine einfache CRUD-Schnittstelle oder ein komplexes Unternehmenssystem handelt.

Denk an die wichtigsten Prinzipien:

  • Fang einfach an mit einem Tool und bau schrittweise aus
  • Behandle Fehler elegant und gib aussagekräftiges Feedback
  • Teste gründlich mit dem MCP Inspector, bevor du dich mit KI-Clients verbindest
  • Denk aus der Perspektive der KI, wenn du Tool-Schnittstellen entwirfst

Deine um MCP-Fähigkeiten erweiterte REST-API ist nicht mehr nur eine Datenquelle – sie ist ein aktiver Teilnehmer an KI-gestützten Workflows. Die Brücke, die du zwischen herkömmlichen APIs und KI-Anwendungen gebaut hast, eröffnet Möglichkeiten, die wir erst zu erkunden beginnen.

Referenzen

[1] Anthropic. "Introducing the Model Context Protocol." https://www.anthropic.com/news/model-context-protocol

[2] Model Context Protocol Official Documentation. https://modelcontextprotocol.io/

[3] Model Context Protocol Python SDK. https://modelcontextprotocol.io/quickstart/server

[4] Claude Desktop Download. https://claude.ai/download

Teste deinen MCP-Server im Browser

MCP-Server-URL einfügen und alle Tools, Resources und Prompts sehen, mit vollständigen Schemas und rohem Request-Log. Kostenlos, ohne Installation, ohne Anmeldung.

MCP Inspector öffnen