MCP Server 开发入门：用 Python 构建第一个 Model Context Protocol 服务

Der Model Context Protocol (MCP) revolutioniert die Art und Weise, wie KI-Anwendungen mit externen Datenquellen und Werkzeugen interagieren. In diesem umfassenden Tutorial erfahren Sie, wie Sie Ihren ersten MCP Server mit Python implementieren – inklusive praktischer Codebeispiele, Best Practices und einer detaillierten Anleitung zur Integration mit HolySheep AI.

Was ist der Model Context Protocol?

Der Model Context Protocol ist ein offenes Protokoll, das eine standardisierte Kommunikation zwischen KI-Modellen und externen Werkzeugen ermöglicht. Im Gegensatz zu traditionellen API-Integrationen bietet MCP eine einheitliche Schnittstelle für:

• Dateisystem-Zugriff: Lesen und Schreiben von Dateien über standardisierte Endpunkte
• Datenbank-Abfragen: Strukturierte Datenbankinteraktionen mit automatischer Typisierung
• Webhook-Integration: Echtzeit-Kommunikation mit externen Diensten
• Benutzerdefinierte Werkzeuge: Flexible Erweiterbarkeit für spezifische Anwendungsfälle

Plattformvergleich: HolySheep AI vs. Offizielle APIs

Bevor wir in die technische Implementierung einsteigen, lohnt sich ein Blick auf die verfügbaren Optionen für MCP-Server-Hosting und API-Zugang:

Feature	HolySheep AI	Offizielle API	Andere Relay-Dienste
Preis pro Million Tokens	$0.42 - $15 (modellabhängig)	$15 - $60+	$3 - $25	85%+ Ersparnis
Währung	¥1 ≈ $1	Nur USD	USD/EUR	Flexibel
Zahlungsmethoden	WeChat, Alipay, Kreditkarte	Nur Kreditkarte	Begrenzt	Lokal angepasst
Latenz	<50ms	100-300ms	80-200ms	Optimiert
Startguthaben	Kostenlose Credits	Keine	Variabel	Einladend
MCP-Kompatibilität	Vollständig	Nur proprietär	Teilweise	Offen

Voraussetzungen und Installation

Bevor wir mit der MCP-Server-Entwicklung beginnen, stellen Sie sicher, dass Sie über folgende Komponenten verfügen:

• Python 3.10 oder höher
• pip-Paketmanager
• Ein HolySheep AI API-Key
• Grundlegende Kenntnisse in Python-Asyncio

Installation der MCP-SDK

# MCP SDK Installation
pip install mcp

HolySheep AI Python-Client
pip install openai

Optional: Für erweiterte Funktionen
pip install aiohttp pydantic

Projektstruktur erstellen

Eine saubere Projektstruktur ist essentiell für wartbare MCP-Server. Wir erstellen folgende Verzeichnisstruktur:

my-mcp-server/
├── src/
│   ├── __init__.py
│   ├── server.py          # Hauptserver-Logik
│   ├── tools.py           # Werkzeugdefinitionen
│   └── config.py          # Konfiguration
├── tests/
│   └── test_server.py
├── requirements.txt
└── README.md

Grundlegender MCP Server mit Python

Jetzt implementieren wir unseren ersten funktionalen MCP-Server. Der folgende Code zeigt eine vollständige Implementierung mit HolySheep AI-Integration:

# src/server.py
import asyncio
from mcp.server import Server
from mcp.types import Tool, TextContent
from openai import AsyncOpenAI

HolySheep AI Konfiguration
client = AsyncOpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

Server-Instanz erstellen
app = Server("mein-mcp-server")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Verfügbare Werkzeuge registrieren"""
    return [
        Tool(
            name="chat_completion",
            description="Erstellt eine Chat-Vervollständigung mit HolySheep AI",
            inputSchema={
                "type": "object",
                "properties": {
                    "model": {
                        "type": "string",
                        "enum": ["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"],
                        "description": "Das zu verwendende KI-Modell"
                    },
                    "message": {
                        "type": "string",
                        "description": "Die Benutzernachricht"
                    },
                    "temperature": {
                        "type": "number",
                        "default": 0.7,
                        "description": "Kreativitätsgrad (0-2)"
                    }
                },
                "required": ["model", "message"]
            }
        ),
        Tool(
            name="text_analysis",
            description="Analysiert Text mit KI-Unterstützung",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "Zu analysierender Text"},
                    "analysis_type": {
                        "type": "string",
                        "enum": ["sentiment", "summary", "keywords"],
                        "description": "Art der Analyse"
                    }
                },
                "required": ["text", "analysis_type"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Werkzeug-Aufrufe verarbeiten"""
    
    if name == "chat_completion":
        response = await client.chat.completions.create(
            model=arguments["model"],
            messages=[{"role": "user", "content": arguments["message"]}],
            temperature=arguments.get("temperature", 0.7)
        )
        return [TextContent(type="text", text=response.choices[0].message.content)]
    
    elif name == "text_analysis":
        prompt = f"Führe eine {arguments['analysis_type']}-Analyse durch: {arguments['text']}"
        response = await client.chat.completions.create(
            model="deepseek-v3.2",  # Kostengünstigste Option
            messages=[{"role": "user", "content": prompt}]
        )
        return [TextContent(type="text", text=response.choices[0].message.content)]
    
    else:
        raise ValueError(f"Unbekanntes Werkzeug: {name}")

async def main():
    """Server starten"""
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

if __name__ == "__main__":
    asyncio.run(main())

Erweiterte Werkzeugdefinitionen

Für komplexere Anwendungsfälle können Sie benutzerdefinierte Werkzeuge mit verschachtelten Eingabestrukturen definieren:

# src/tools.py
from mcp.types import Tool, Resource
from typing import Optional
import json

class ToolRegistry:
    """Zentrales Register für MCP-Werkzeuge"""
    
    @staticmethod
    def create_data_processing_tool() -> Tool:
        return Tool(
            name="process_data",
            description="Verarbeitet und transformiert strukturierte Daten",
            inputSchema={
                "type": "object",
                "properties": {
                    "data": {"type": "array", "description": "Eingabedaten als JSON-Array"},
                    "operations": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "type": {"type": "string", "enum": ["filter", "map", "reduce", "sort"]},
                                "config": {"type": "object"}
                            }
                        }
                    },
                    "llm_assisted": {
                        "type": "boolean",
                        "default": False,
                        "description": "KI-gestützte Verarbeitung aktivieren"
                    }
                },
                "required": ["data", "operations"]
            }
        )
    
    @staticmethod
    def create_batch_processing_tool() -> Tool:
        return Tool(
            name="batch_chat",
            description="Verarbeitet mehrere Chat-Anfragen parallel",
            inputSchema={
                "type": "object",
                "properties": {
                    "requests": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "model": {"type": "string"},
                                "message": {"type": "string"},
                                "priority": {"type": "integer", "default": 0}
                            }
                        },
                        "maxItems": 100
                    },
                    "concurrency": {
                        "type": "integer",
                        "default": 5,
                        "minimum": 1,
                        "maximum": 50
                    }
                },
                "required": ["requests"]
            }
        )

Werkzeug-Registry exportieren
tool_registry = ToolRegistry()

Konfiguration und Umgebungsvariablen

# src/config.py
import os
from dataclasses import dataclass
from typing import Optional

@dataclass
class ServerConfig:
    """Server-Konfiguration mit HolySheep AI"""
    
    # API-Konfiguration
    api_key: str
    base_url: str = "https://api.holysheep.ai/v1"
    
    # Server-Einstellungen
    server_name: str = "mcp-server"
    server_version: str = "1.0.0"
    
    # Modell-Standardwerte
    default_model: str = "deepseek-v3.2"
    fallback_model: str = "gemini-2.5-flash"
    
    # Performance-Einstellungen
    max_concurrent_requests: int = 50
    request_timeout: int = 30
    
    # Preisoptimierung
    use_cost_optimization: bool = True
    
    @classmethod
    def from_env(cls) -> "ServerConfig":
        """Konfiguration aus Umgebungsvariablen laden"""
        return cls(
            api_key=os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
            base_url=os.getenv("HOLYSHEEP_BASE_URL", "https://api.holysheep.ai/v1"),
            server_name=os.getenv("SERVER_NAME", "mcp-server"),
            server_version=os.getenv("SERVER_VERSION", "1.0.0"),
            default_model=os.getenv("DEFAULT_MODEL", "deepseek-v3.2"),
            max_concurrent_requests=int(os.getenv("MAX_CONCURRENT", "50"))
        )

Modell-Preisübersicht (2026)
MODEL_PRICING = {
    "gpt-4.1": {"input": 8.00, "output": 8.00, "currency": "USD"},
    "claude-sonnet-4.5": {"input": 15.00, "output": 15.00, "currency": "USD"},
    "gemini-2.5-flash": {"input": 2.50, "output": 2.50, "currency": "USD"},
    "deepseek-v3.2": {"input": 0.42, "output": 0.42, "currency": "USD"}
}

def get_cheapest_model() -> str:
    """Gibt das günstigste Modell zurück"""
    return min(MODEL_PRICING.items(), key=lambda x: x[1]["input"])[0]

Testing und Debugging

# tests/test_server.py
import pytest
import asyncio
from src.server import app, client
from src.config import ServerConfig

@pytest.fixture
def config():
    return ServerConfig.from_env()

@pytest.mark.asyncio
async def test_chat_completion(config):
    """Test der Chat-Completion-Funktion"""
    response = await client.chat.completions.create(
        model="deepseek-v3.2",
        messages=[{"role": "user", "content": "Sag 'Hallo Welt' auf Deutsch"}]
    )
    assert response.choices[0].message.content
    assert "Hallo" in response.choices[0].message.content

@pytest.mark.asyncio
async def test_model_listing():
    """Verifiziere verfügbare Modelle"""
    from openai import AsyncOpenAI
    
    test_client = AsyncOpenAI(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1"
    )
    
    models = await test_client.models.list()
    model_ids = [m.id for m in models.data]
    
    expected_models = ["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"]
    
    for model in expected_models:
        assert model in model_ids, f"Modell {model} nicht verfügbar"

@pytest.mark.asyncio
async def test_connection_latency():
    """Test der HolySheep AI Latenz"""
    import time
    
    start = time.time()
    response = await client.chat.completions.create(
        model="deepseek-v3.2",
        messages=[{"role": "user", "content": "Ping"}]
    )
    latency = (time.time() - start) * 1000
    
    assert latency < 5000, f"Latenz zu hoch: {latency}ms"
    print(f"Latenz: {latency:.2f}ms")

if __name__ == "__main__":
    pytest.main([__file__, "-v"])

Häufige Fehler und Lösungen

1. AuthenticationError: Invalid API Key

Symptom: Die Anfrage wird mit einem 401 Unauthorized-Fehler abgelehnt.

Lösungen:

• Überprüfen Sie, ob der API-Key korrekt als YOUR_HOLYSHEEP_API_KEY gesetzt ist
• Vergewissern Sie sich, dass keine führenden oder nachfolgenden Leerzeichen vorhanden sind
• Erstellen Sie einen neuen API-Key in Ihrem HolySheep AI Dashboard

# Korrekte Authentifizierung
client = AsyncOpenAI(
    api_key=os.environ.get("HOLYSHEEP_API_KEY"),  # NICHT hardcodieren!
    base_url="https://api.holysheep.ai/v1"  # Korrekter Endpunkt
)

2. RateLimitError: Überschreitung der Anfragenlimite

Symptom: Fehlermeldung "Rate limit exceeded" bei mehreren gleichzeitigen Anfragen.

Lösungen:

Verwandte Ressourcen

• 📚 KI API Tutorials
• 💰 Preise ansehen
• 📖 Entwickler-Dokumentation
• 🚀 Kostenlos registrieren

Verwandte Artikel

• de openai embeddings api xianglianghuajierushijian 2026 04 0
• de gpt 41 api zhongguoqufangwentongguo holysheep zhon 2026 0