Wie man einen MCP-Server aufbaut: Schritt für Schritt mit Code-Beispielen

Das Model Context Protocol (MCP) entwickelt sich zum neuen Standard für die Verbindung von KI-Modellen mit realen Tools und Diensten. Der Aufbau eines MCP-Servers ermöglicht es Ihnen, Daten, Aktionen und Ressourcen einem LLM wie Claude über eine einfache, standardisierte Schnittstelle zur Verfügung zu stellen.

In dieser Anleitung lernen Sie Schritt für Schritt, wie Sie einen grundlegenden MCP-Server in Python einrichten, Ressourcen und Tools definieren und ihn mit einem MCP-Client verbinden.

Was ist ein MCP-Server

Ein MCP-Server fungiert als Brücke zwischen einem LLM und einem externen System wie einer Datenbank, einem Dateispeicher oder einer API. Er definiert Ressourcen (lesbare Daten), Tools (Aktionen) und Prompts (Anweisungen) so, dass das LLM sie während seiner Aufgaben sicher nutzen kann.

Anstatt für jedes Modell oder Tool eine benutzerdefinierte Integration zu schreiben, bietet MCP einen universellen Standard, der mit der Protokollversion 0.1 (aktuell Stand April 2025) funktioniert.

Was Sie vor dem Start benötigen

• Python 3.8 oder höher
• Grundlegende Erfahrung mit Python-Scripting
• MCP SDK für Python (verfügbar über pip)
• Ein MCP-kompatibler Client wie Claude Desktop oder Cursor (optional zum Testen)
• Git für die Versionskontrolle (empfohlen)
• Ein Texteditor oder eine IDE (Visual Studio Code empfohlen)

Verständnis der Grundstruktur

In MCP:

• Server: Stellt dem LLM Ressourcen und Tools zur Verfügung.
• Client: Verbindet das LLM mit Ihrem Server.
• Protokoll: Verwaltet die Kommunikation zwischen Client und Server.

Sie werden zwei wichtige Primitive definieren:

• Ressource: Statische oder dynamische Informationen, die das LLM lesen kann.
• Tool: Eine aufrufbare Funktion, die das LLM ausführen kann.

Der Kommunikationsfluss funktioniert wie folgt:

1. Das LLM (über einen Client) fordert Daten oder Aktionen von Ihrem Server an
2. Ihr Server verarbeitet diese Anfragen und gibt standardisierte Antworten zurück
3. Das LLM kann diese Informationen dann in seinen Überlegungen und Antworten verwenden

1. Richten Sie Ihr Python-Projekt ein

Beginnen Sie mit der Erstellung eines Projektverzeichnisses und einer Python-Virtualumgebung.

mkdir my_mcp_server
cd my_mcp_server
python -m venv venv
source venv/bin/activate  # Linux/Mac
venvScriptsactivate     # Windows

Erstellen Sie eine grundlegende Projektstruktur:

mkdir -p src/resources src/tools tests
touch src/__init__.py src/resources/__init__.py src/tools/__init__.py
touch requirements.txt README.md

Fügen Sie Folgendes zu Ihrer requirements.txt hinzu:

mcp-server>=0.1.0
pydantic>=2.0.0
pytest>=7.0.0

2. Installieren Sie das MCP SDK

Installieren Sie das MCP-Server-SDK für Python und andere Abhängigkeiten:

pip install -r requirements.txt

Wenn das offizielle SDK noch nicht veröffentlicht ist, müssen Sie möglicherweise von einem GitHub-Repository installieren:

pip install git+https://github.com/anthropic/mcp-server-python.git

3. Erstellen Sie einen einfachen MCP-Server

Erstellen Sie eine Datei namens src/server.py:

from typing import Dict, Any
from mcp_server import MCPServer
import logging

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

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Resources and tools will be added here
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

if __name__ == "__main__":
    main()

Dies richtet einen grundlegenden MCP-Server mit ordnungsgemäßem Logging ein.

4. Definieren Sie eine Ressource

Ressourcen stellen Daten bereit, die das Modell lesen kann. Erstellen Sie eine Datei src/resources/user_profiles.py:

from typing import List, Dict, Any
from pydantic import BaseModel
import logging

logger = logging.getLogger("mcp_server.resources")

class UserProfile(BaseModel):
    """Data model for user profiles."""
    name: str
    role: str
    department: str = "General"
    years_experience: int = 0

def fetch_user_profiles() -> List[Dict[str, Any]]:
    """
    Fetch user profiles from the database.
    
    Returns:
        List[Dict[str, Any]]: A list of user profile dictionaries.
    """
    try:
        # In a real implementation, this would query a database
        # For this example, we'll return mock data
        users = [
            UserProfile(name="Alice", role="Engineer", department="Engineering", years_experience=5),
            UserProfile(name="Bob", role="Product Manager", department="Product", years_experience=3),
            UserProfile(name="Charlie", role="Designer", department="Design", years_experience=7)
        ]
        
        logger.info(f"Successfully fetched {len(users)} user profiles")
        return [user.model_dump() for user in users]
    except Exception as e:
        logger.error(f"Error fetching user profiles: {e}")
        # In production, you might want to return an empty list or raise
        # a specific exception depending on your error handling strategy
        return []

Aktualisieren Sie nun src/server.py, um diese Ressource einzubeziehen:

from typing import Dict, Any
from mcp_server import MCPServer, Resource
import logging
from src.resources.user_profiles import fetch_user_profiles

# ... existing code ...

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Add the user profiles resource
        user_profiles = Resource(
            name="user_profiles",
            description="List of user profiles from the company database.",
            fetch_fn=fetch_user_profiles
        )
        
        server.add_resource(user_profiles)
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

if __name__ == "__main__":
    main()

Das LLM kann nun über den MCP-Client user_profiles abfragen.

5. Definieren Sie ein Tool

Tools ermöglichen dem LLM, eine Aktion auszuführen. Erstellen Sie eine Datei src/tools/user_management.py:

from typing import Dict, Any, Optional
from pydantic import BaseModel, Field, ValidationError
import logging

logger = logging.getLogger("mcp_server.tools")

class CreateUserRequest(BaseModel):
    """Validation model for user creation requests."""
    name: str = Field(..., min_length=2, description="User's full name")
    role: str = Field(..., min_length=2, description="User's job role")
    department: Optional[str] = Field("General", description="User's department")
    years_experience: Optional[int] = Field(0, ge=0, description="Years of professional experience")

def create_user_profile(request_data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Create a new user profile in the database.
    
    Args:
        request_data (Dict[str, Any]): User data containing name, role, etc.
        
    Returns:
        Dict[str, Any]: Response with status and user info
    """
    try:
        # Validate the input data
        user_data = CreateUserRequest(**request_data)
        
        # In a real implementation, this would insert into a database
        # For this example, we'll just log the action
        logger.info(f"Creating new user: {user_data.name} - {user_data.role} in {user_data.department}")
        
        # Return success response with created user data
        return {
            "status": "success",
            "message": f"User {user_data.name} created successfully",
            "user": user_data.model_dump()
        }
    except ValidationError as e:
        # Handle validation errors
        logger.error(f"Validation error: {e}")
        return {
            "status": "error",
            "message": "Invalid user data provided",
            "details": str(e)
        }
    except Exception as e:
        # Handle other errors
        logger.error(f"Error creating user: {e}")
        return {
            "status": "error",
            "message": "Failed to create user",
            "details": str(e)
        }

Aktualisieren Sie nun src/server.py, um dieses Tool einzubeziehen:

from typing import Dict, Any
from mcp_server import MCPServer, Resource, Tool
import logging
from src.resources.user_profiles import fetch_user_profiles
from src.tools.user_management import create_user_profile

# ... existing code ...

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Add the user profiles resource
        user_profiles = Resource(
            name="user_profiles",
            description="List of user profiles from the company database.",
            fetch_fn=fetch_user_profiles
        )
        
        # Add the create user tool
        create_user = Tool(
            name="create_user_profile",
            description="Create a new user profile in the database.",
            parameters={
                "name": {"type": "string", "description": "User's full name"},
                "role": {"type": "string", "description": "User's job role"},
                "department": {"type": "string", "description": "User's department (optional)"},
                "years_experience": {"type": "integer", "description": "Years of experience (optional)"}
            },
            execute_fn=create_user_profile
        )
        
        server.add_resource(user_profiles)
        server.add_tool(create_user)
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

6. Fehlerbehandlung und Validierung

Erstellen Sie eine Datei src/utils/validation.py, um Ihre Validierungslogik zu zentralisieren:

from typing import Dict, Any, List, Optional, Type
from pydantic import BaseModel, ValidationError
import logging

logger = logging.getLogger("mcp_server.validation")

def validate_request(
    data: Dict[str, Any],
    model_class: Type[BaseModel]
) -> tuple[Optional[BaseModel], Optional[Dict[str, Any]]]:
    """
    Validate request data against a Pydantic model.
    
    Args:
        data: The input data to validate
        model_class: The Pydantic model class to use for validation
        
    Returns:
        tuple: (validated_model, error_dict)
            - If valid: (model instance, None)
            - If invalid: (None, error dictionary)
    """
    try:
        validated_data = model_class(**data)
        return validated_data, None
    except ValidationError as e:
        errors = e.errors()
        error_dict = {
            "status": "error",
            "message": "Validation failed",
            "errors": errors
        }
        logger.error(f"Validation error: {errors}")
        return None, error_dict

Diese Hilfsfunktion kann in all Ihren Tools verwendet werden, um Eingabedaten konsistent zu validieren.

7. Ausführen und Testen des MCP-Servers

Erstellen Sie ein einfaches Testskript test_server.py, um zu überprüfen, ob Ihr Server funktioniert:

import requests
import json
import time
import subprocess
import sys
from pathlib import Path

def test_server():
    """Simple test to verify the MCP server is running correctly."""
    # Start the server in a separate process
    server_process = subprocess.Popen([sys.executable, "src/server.py"])
    
    try:
        # Wait for server to start
        time.sleep(2)
        
        # Test the server using the MCP client
        # In a real test, you would use the MCP client SDK
        # For this example, we'll simulate a client using HTTP requests
        
        # Assuming the server is running on localhost:8000
        base_url = "http://localhost:8000"
        
        # Test fetching resources
        response = requests.get(f"{base_url}/resources/user_profiles")
        assert response.status_code == 200
        data = response.json()
        print("Resource response:", json.dumps(data, indent=2))
        
        # Test executing a tool
        tool_data = {
            "name": "Test User",
            "role": "Tester",
            "department": "QA"
        }
        response = requests.post(
            f"{base_url}/tools/create_user_profile",
            json=tool_data
        )
        assert response.status_code == 200
        data = response.json()
        print("Tool response:", json.dumps(data, indent=2))
        
        print("All tests passed!")
        
    finally:
        # Clean up: terminate the server process
        server_process.terminate()
        server_process.wait()

if __name__ == "__main__":
    test_server()

Starten Sie Ihren Server:

python src/server.py

In einem separaten Terminal könnten Sie eine Ausgabe wie diese sehen, wenn der Server läuft:

2025-04-28 10:15:23 - mcp_server - INFO - Starting MCP server...
2025-04-28 10:15:23 - mcp_server - INFO - Server listening on 0.0.0.0:8000
2025-04-28 10:15:30 - mcp_server.resources - INFO - Successfully fetched 3 user profiles
2025-04-28 10:15:35 - mcp_server.tools - INFO - Creating new user: Test User - Tester in QA

Konfigurieren Sie dann Ihren MCP-Client (wie Claude Desktop), um eine Verbindung zu Ihrem lokalen MCP-Server herzustellen, indem Sie die Server-URL oder den Befehl zum Starten des Servers angeben.

Sicherheitsüberlegungen

Beachten Sie bei der Bereitstellung eines MCP-Servers diese Sicherheitsempfehlungen:

1. Authentifizierung: Implementieren Sie API-Schlüssel oder OAuth zur Authentifizierung von Clients.

def authenticate_request(request):
    api_key = request.headers.get("X-API-Key")
    if not api_key or api_key != os.environ.get("MCP_API_KEY"):
        raise ValueError("Invalid API key")

2. Eingabevalidierung: Validieren Sie immer alle Eingaben mit Pydantic-Modellen.
3. Ratenbegrenzung: Implementieren Sie Ratenbegrenzungen, um Missbrauch zu verhindern.
4. HTTPS: Verwenden Sie in der Produktion immer HTTPS.
5. Eingeschränkte Aktionen: Definieren Sie klare Grenzen für die Funktionen von Tools.

Leistungsoptimierung

1. Caching: Zwischenspeichern von teuren Ressourcenabrufen:

from functools import lru_cache

@lru_cache(maxsize=128, ttl=300)  # Cache for 5 minutes
def fetch_user_profiles():
    # Expensive database query
    pass

2. Asynchrone Verarbeitung: Verwenden Sie Async für I/O-gebundene Operationen:

async def fetch_user_profiles():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com/users") as response:
            data = await response.json()
            return data

3. Verbindungspooling: Verwenden Sie Verbindungspools für Datenbankzugriffe.

Bereitstellung

Lokale Entwicklung

Für die lokale Entwicklung führen Sie aus:

python src/server.py

Docker-Bereitstellung

Erstellen Sie ein Dockerfile:

FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["python", "src/server.py"]

Bauen und ausführen:

docker build -t mcp-server .
docker run -p 8000:8000 mcp-server

Cloud-Bereitstellung (AWS)

1. Erstellen Sie eine EC2-Instanz oder verwenden Sie AWS App Runner
2. Stellen Sie Ihren Docker-Container bereit
3. Richten Sie einen Application Load Balancer ein
4. Konfigurieren Sie Sicherheitsgruppen, um den Zugriff zu beschränken

Testen Ihres MCP-Servers

Erstellen Sie eine Testdatei tests/test_resources.py:

import pytest
from src.resources.user_profiles import fetch_user_profiles

def test_fetch_user_profiles():
    """Test that user profiles can be fetched successfully."""
    profiles = fetch_user_profiles()
    
    # Check structure
    assert isinstance(profiles, list)
    assert len(profiles) > 0
    
    # Check content
    first_profile = profiles[0]
    assert "name" in first_profile
    assert "role" in first_profile
    assert isinstance(first_profile["name"], str)

Führen Sie Tests mit aus:

pytest

Häufige Fehler und Fehlerbehebung

Problem Lösung Beispiel Keine Verbindung zum MCP-Server möglich Überprüfen Sie, ob Ihr Server läuft und der Port korrekt ist netstat -tulpn | grep 8000 LLM kann Ressourcen nicht finden Überprüfen Sie, ob die Felder name und description richtig gesetzt sind Überprüfen Sie Ihre Resource-Initialisierung Fehler bei der Tool-Ausführung Validieren Sie, dass Eingabeparameter den erwarteten Typen entsprechen Verwenden Sie Pydantic zur Validierung Client kann Ausgabe nicht parsen Stellen Sie sicher, dass Ihre Funktionen JSON-serialisierbare Daten zurückgeben Verwenden Sie .model_dump() anstelle von benutzerdefinierten Objekten Server stürzt beim Start ab Überprüfen Sie Ihre Importe und Umgebungsvariablen Setzen Sie DEBUG=True für ausführliches Logging Tool-Timeout Fügen Sie Timeout-Behandlung für externe API-Aufrufe hinzu Verwenden Sie asyncio.wait_for() mit einem Timeout Authentifizierungsfehler Überprüfen Sie API-Schlüssel und Berechtigungen Überprüfen Sie Request-Header XML/JSON-Parsing-Fehler Verwenden Sie korrekte Content-Type-Header Setzen Sie Content-Type: application/json

Nächste Schritte

Nach dem Aufbau Ihres grundlegenden MCP-Servers sollten Sie diese erweiterten Erweiterungen in Betracht ziehen:

1. Datenbankintegration: Verbindung zu PostgreSQL, MongoDB oder anderen Datenbanken.
2. Dateioperationen: Hinzufügen von Tools zum Lesen, Schreiben und Transformieren von Dateien.
3. Externe APIs: Integration mit beliebten Diensten wie GitHub, Slack oder Google Drive.
4. Webhooks: Ermöglichen Sie dem LLM, Ereignisse in anderen Systemen auszulösen.
5. Streaming-Ressourcen: Unterstützung für das Streaming großer Datensätze.
6. Kontextbewusste Aktionen: Hinzufügen von Tools, die den aktuellen Kontext des LLM verstehen.

Beispiel: Hinzufügen einer Datenbankverbindung

import psycopg2
from contextlib import contextmanager

@contextmanager
def get_db_connection():
    """Create a database connection context manager."""
    conn = None
    try:
        conn = psycopg2.connect(
            host=os.environ.get("DB_HOST"),
            database=os.environ.get("DB_NAME"),
            user=os.environ.get("DB_USER"),
            password=os.environ.get("DB_PASSWORD")
        )
        yield conn
    finally:
        if conn is not None:
            conn.close()

def fetch_user_profiles_from_db():
    """Fetch user profiles from a PostgreSQL database."""
    with get_db_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT name, role, department FROM users")
            columns = [desc[0] for desc in cur.description]
            return [dict(zip(columns, row)) for row in cur.fetchall()]

Fazit

Der Aufbau eines einfachen MCP-Servers in Python öffnet die Tür, um LLMs viel leistungsfähiger zu machen. Indem Sie Daten und Aktionen über ein sauberes, standardisiertes Protokoll bereitstellen, erleichtern Sie es KI-Systemen, sicher und sinnvoll mit externen Diensten zu interagieren.

Beginnen Sie klein, konzentrieren Sie sich auf eine Ressource und ein Tool, und Sie können im Laufe der Zeit zu fortgeschritteneren Anwendungsfällen wie Datenbanken, Cloud-Speicher oder internen APIs erweitern.

Das MCP-Ökosystem wächst schnell, und die Implementierung dieser Standards jetzt wird Ihre Anwendungen in die Lage versetzen, von Verbesserungen sowohl im Protokoll als auch in den LLMs, die es verwenden, zu profitieren.

FAQs