MCP-Server in Python aufbauen: WebSearch & Scrape-Anleitung

Bauen Sie MCP-Server für KI-Assistenten mit Python-Beispielen

Inhaltsverzeichnis

Das Model Context Protocol (MCP) revolutioniert die Art und Weise, wie KI-Assistenten mit externen Datenquellen und Tools interagieren. In diesem Leitfaden erkunden wir, wie man MCP-Server in Python aufbaut, mit Beispielen, die sich auf Websuche und Scraping-Funktionen konzentrieren.

MCP-Roboter

Was ist das Model Context Protocol?

Das Model Context Protocol (MCP) ist ein offenes Protokoll, das von Anthropic eingeführt wurde, um zu standardisieren, wie KI-Assistenten mit externen Systemen verbunden werden. Anstatt für jede Datenquelle individuelle Integrationen zu erstellen, bietet MCP eine einheitliche Schnittstelle, die es ermöglicht:

  • KI-Assistenten (wie Claude, ChatGPT oder maßgeschneiderte LLM-Anwendungen) Tools zu entdecken und zu nutzen
  • Entwicklern Datenquellen, Tools und Prompts über ein standardisiertes Protokoll bereitzustellen
  • Nahtlose Integration ohne für jeden Anwendungsfall das Rad neu zu erfinden

Das Protokoll arbeitet auf einer Client-Server-Architektur, bei der:

  • MCP-Clients (KI-Assistenten) Fähigkeiten entdecken und nutzen
  • MCP-Server Ressourcen, Tools und Prompts bereitstellen
  • Die Kommunikation erfolgt über JSON-RPC über stdio oder HTTP/SSE

Wenn Sie daran interessiert sind, MCP-Server in anderen Sprachen zu implementieren, werfen Sie einen Blick auf unseren Leitfaden zur Implementierung eines MCP-Servers in Go, der die Protokollspezifikationen und Nachrichtenstruktur detailliert behandelt.

Warum MCP-Server in Python aufbauen?

Python ist eine hervorragende Wahl für die MCP-Serverentwicklung, weil:

  1. Reiche Ökosysteme: Bibliotheken wie requests, beautifulsoup4, selenium und playwright machen das Web-Scraping einfach
  2. MCP-SDK: Das offizielle Python-SDK (mcp) bietet robuste Unterstützung für die Serverimplementierung
  3. Schnelle Entwicklung: Die Einfachheit von Python ermöglicht schnelles Prototyping und Iteration
  4. KI/ML-Integration: Einfache Integration mit KI-Bibliotheken wie langchain, openai und Datenverarbeitungs-Tools
  5. Community-Unterstützung: Große Community mit umfangreicher Dokumentation und Beispielen

Einrichtung Ihrer Entwicklungsumgebung

Zuerst erstellen Sie eine virtuelle Umgebung und installieren die erforderlichen Abhängigkeiten. Die Verwendung virtueller Umgebungen ist für die Isolierung von Python-Projekten unerlässlich - wenn Sie eine Auffrischung benötigen, werfen Sie einen Blick auf unseren venv Cheatsheet für detaillierte Befehle und Best Practices.

# Erstellen und Aktivieren der virtuellen Umgebung
python -m venv mcp-env
source mcp-env/bin/activate  # Auf Windows: mcp-env\Scripts\activate

# Installieren des MCP-SDK und der Web-Scraping-Bibliotheken
pip install mcp requests beautifulsoup4 playwright lxml
playwright install  # Installieren der Browser-Treiber für Playwright

Moderne Alternative: Wenn Sie schnellere Abhängigkeitsauflösung und Installation bevorzugen, sollten Sie uv - den modernen Python-Paket- und Umgebungsmanager in Betracht ziehen, der für große Projekte deutlich schneller sein kann als pip.

Aufbau eines grundlegenden MCP-Servers

Lassen Sie uns mit einer minimalen MCP-Server-Struktur beginnen. Wenn Sie neu in Python sind oder eine schnelle Referenz für Syntax und häufige Muster benötigen, bietet unser Python Cheatsheet einen umfassenden Überblick über die Python-Grundlagen.

import asyncio
from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio

# Erstellen einer Serverinstanz
app = Server("websearch-scraper")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Definieren der verfügbaren Tools"""
    return [
        Tool(
            name="search_web",
            description="Suche im Web nach Informationen",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Suchbegriff"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Maximale Anzahl der Ergebnisse",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Toolausführung verarbeiten"""
    if name == "search_web":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)

        # Suchlogik hier implementieren
        results = await perform_web_search(query, max_results)

        return [TextContent(
            type="text",
            text=f"Suchergebnisse für '{query}':\n\n{results}"
        )]

    raise ValueError(f"Unbekanntes Tool: {name}")

async def perform_web_search(query: str, max_results: int) -> str:
    """Platzhalter für die tatsächliche Suchimplementierung"""
    return f"Gefunden {max_results} Ergebnisse für: {query}"

async def main():
    """Server ausführen"""
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

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

Implementierung der Web-Suchfunktion

Jetzt implementieren wir ein echtes Web-Such-Tool mit DuckDuckGo (das keine API-Schlüssel benötigt):

import asyncio
import requests
from bs4 import BeautifulSoup
from urllib.parse import quote_plus

async def search_duckduckgo(query: str, max_results: int = 5) -> list[dict]:
    """DuckDuckGo durchsuchen und Ergebnisse parsen"""

    url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }

    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()

        soup = BeautifulSoup(response.text, 'html.parser')
        results = []

        for result in soup.select('.result')[:max_results]:
            title_elem = result.select_one('.result__title')
            snippet_elem = result.select_one('.result__snippet')
            url_elem = result.select_one('.result__url')

            if title_elem and snippet_elem:
                results.append({
                    "title": title_elem.get_text(strip=True),
                    "snippet": snippet_elem.get_text(strip=True),
                    "url": url_elem.get_text(strip=True) if url_elem else "N/A"
                })

        return results

    except Exception as e:
        raise Exception(f"Suche fehlgeschlagen: {str(e)}")

def format_search_results(results: list[dict]) -> str:
    """Suchergebnisse für die Anzeige formatieren"""
    if not results:
        return "Keine Ergebnisse gefunden."

    formatted = []
    for i, result in enumerate(results, 1):
        formatted.append(
            f"{i}. {result['title']}\n"
            f"   {result['snippet']}\n"
            f"   URL: {result['url']}\n"
        )

    return "\n".join(formatted)

Hinzufügen von Web-Scraping-Funktionen

Lassen Sie uns ein Tool hinzufügen, um Inhalte von Webseiten zu scrapen. Beim Scrapen von HTML-Inhalten für die Verwendung mit LLMs möchten Sie diese möglicherweise in Markdown-Format umwandeln, um eine bessere Verarbeitung zu ermöglichen. Zu diesem Zweck werfen Sie einen Blick auf unseren umfassenden Leitfaden zur Umwandlung von HTML in Markdown mit Python, der 6 verschiedene Bibliotheken mit Benchmarks und praktischen Empfehlungen vergleicht.

from playwright.async_api import async_playwright
import asyncio

async def scrape_webpage(url: str, selector: str = None) -> dict:
    """Inhalte von einer Webseite mit Playwright scrapen"""

    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()

        try:
            await page.goto(url, timeout=30000)

            # Warten, bis der Inhalt geladen ist
            await page.wait_for_load_state('networkidle')

            if selector:
                # Spezifisches Element extrahieren
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "Selector nicht gefunden"
            else:
                # Hauptinhalt extrahieren
                content = await page.inner_text('body')

            title = await page.title()

            return {
                "title": title,
                "content": content[:5000],  # Inhaltlänge begrenzen
                "url": url,
                "success": True
            }

        except Exception as e:
            return {
                "error": str(e),
                "url": url,
                "success": False
            }
        finally:
            await browser.close()

# Scraper-Tool zum MCP-Server hinzufügen
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="Suche im Web mit DuckDuckGo",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Suchbegriff"},
                    "max_results": {"type": "number", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Inhalte von einer Webseite scrapen",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "URL zum Scrapen"},
                    "selector": {
                        "type": "string",
                        "description": "Optionaler CSS-Selector für spezifische Inhalte"
                    }
                },
                "required": ["url"]
            }
        )
    ]

Vollständige MCP-Server-Implementierung

Hier ist ein vollständiger, produktionsreifer MCP-Server mit sowohl Such- als auch Scraping-Funktionalität:

#!/usr/bin/env python3
"""
MCP-Server für Websuche und Scraping
Bietet Tools zum Durchsuchen des Webs und zum Extrahieren von Inhalten aus Seiten
"""

import asyncio
import logging
from typing import Any
import requests
from bs4 import BeautifulSoup
from urllib.parse import quote_plus
from playwright.async_api import async_playwright

from mcp.server import Server
from mcp.types import Tool, TextContent, ImageContent, EmbeddedResource
import mcp.server.stdio

# Konfigurieren des Loggings
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("websearch-scraper")

# Server erstellen
app = Server("websearch-scraper")

# Suchimplementierung
async def search_web(query: str, max_results: int = 5) -> str:
    """Durchsuche DuckDuckGo und gebe formatierte Ergebnisse zurück"""
    url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
    headers = {"User-Agent": "Mozilla/5.0"}

    try:
        response = requests.get(url, headers=headers, timeout=10)
        soup = BeautifulSoup(response.text, 'html.parser')
        results = []

        for result in soup.select('.result')[:max_results]:
            title = result.select_one('.result__title')
            snippet = result.select_one('.result__snippet')
            link = result.select_one('.result__url')

            if title and snippet:
                results.append({
                    "title": title.get_text(strip=True),
                    "snippet": snippet.get_text(strip=True),
                    "url": link.get_text(strip=True) if link else ""
                })

        # Ergebnisse formatieren
        if not results:
            return "Keine Ergebnisse gefunden."

        formatted = [f"Gefunden {len(results)} Ergebnisse für '{query}':\n"]
        for i, r in enumerate(results, 1):
            formatted.append(f"\n{i}. **{r['title']}**")
            formatted.append(f"   {r['snippet']}")
            formatted.append(f"   {r['url']}")

        return "\n".join(formatted)

    except Exception as e:
        logger.error(f"Suche fehlgeschlagen: {e}")
        return f"Suchfehler: {str(e)}")

# Scraper-Implementierung
async def scrape_page(url: str, selector: str = None) -> str:
    """Extrahiere Webseiteninhalt mit Playwright"""
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()

        try:
            await page.goto(url, timeout=30000)
            await page.wait_for_load_state('networkidle')

            title = await page.title()

            if selector:
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "Selector nicht gefunden"
            else:
                content = await page.inner_text('body')

            # Begrenzung der Inhaltslänge
            content = content[:8000] + "..." if len(content) > 8000 else content

            result = f"**{title}**\n\nURL: {url}\n\n{content}"
            return result

        except Exception as e:
            logger.error(f"Scraping fehlgeschlagen: {e}")
            return f"Scraping-Fehler: {str(e)}"
        finally:
            await browser.close()

# MCP-Tool-Definitionen
@app.list_tools()
async def list_tools() -> list[Tool]:
    """Liste der verfügbaren MCP-Tools"""
    return [
        Tool(
            name="search_web",
            description="Durchsuche das Web mit DuckDuckGo. Gibt Titel, Auszüge und URLs zurück.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Die Suchanfrage"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Maximale Anzahl der Ergebnisse (Standard: 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Extrahiere Inhalte von einer Webseite. Kann spezifische Elemente mit CSS-Selektoren ansteuern.",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "Die URL zum Scrapen"
                    },
                    "selector": {
                        "type": "string",
                        "description": "Optionaler CSS-Selektor zum Extrahieren spezifischer Inhalte"
                    }
                },
                "required": ["url"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """Verarbeitung der Tool-Ausführung"""
    try:
        if name == "search_web":
            query = arguments["query"]
            max_results = arguments.get("max_results", 5)
            result = await search_web(query, max_results)
            return [TextContent(type="text", text=result)]

        elif name == "scrape_webpage":
            url = arguments["url"]
            selector = arguments.get("selector")
            result = await scrape_page(url, selector)
            return [TextContent(type="text", text=result)]

        else:
            raise ValueError(f"Unbekanntes Tool: {name}")

    except Exception as e:
        logger.error(f"Tool-Ausführung fehlgeschlagen: {e}")
        return [TextContent(
            type="text",
            text=f"Fehler bei der Ausführung von {name}: {str(e)}"
        )]

async def main():
    """Führe den MCP-Server aus"""
    logger.info("Starten des WebSearch-Scraper MCP-Servers")

    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

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

Konfigurieren Ihres MCP-Servers

Um Ihren MCP-Server mit Claude Desktop oder anderen MCP-Clients zu verwenden, erstellen Sie eine Konfigurationsdatei:

Für Claude Desktop (claude_desktop_config.json):

{
  "mcpServers": {
    "websearch-scraper": {
      "command": "python",
      "args": [
        "/path/to/your/mcp_server.py"
      ],
      "env": {}
    }
  }
}

Speicherort:

  • macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
  • Windows: %APPDATA%\Claude\claude_desktop_config.json
  • Linux: ~/.config/Claude/claude_desktop_config.json

Testen Ihres MCP-Servers

Erstellen Sie ein Testskript, um die Funktionalität zu überprüfen:

import asyncio
import json
import sys
from io import StringIO

async def test_mcp_server():
    """Testen Sie den MCP-Server lokal"""

    # Testen der Suche
    print("Testen der Websuche...")
    results = await search_web("Python MCP-Tutorial", 3)
    print(results)

    # Testen des Scrapers
    print("\n\nTesten des Webseiten-Scrapers...")
    content = await scrape_page("https://example.com")
    print(content[:500])

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

Erweitete Funktionen und Best Practices

1. Rate Limiting

Implementieren Sie Rate Limiting, um Zielserver nicht zu überlasten:

import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_requests: int = 10, time_window: int = 60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = defaultdict(list)

    async def acquire(self, key: str):
        now = time.time()
        self.requests[key] = [
            t for t in self.requests[key]
            if now - t < self.time_window
        ]

        if len(self.requests[key]) >= self.max_requests:
            raise Exception("Rate Limit überschritten")

        self.requests[key].append(now)

limiter = RateLimiter(max_requests=10, time_window=60)

2. Caching

Fügen Sie Caching hinzu, um die Leistung zu verbessern:

from functools import lru_cache
import hashlib

@lru_cache(maxsize=100)
async def cached_search(query: str, max_results: int):
    return await search_web(query, max_results)

3. Fehlerbehandlung

Implementieren Sie robuste Fehlerbehandlung:

from enum import Enum

class ErrorType(Enum):
    NETWORK_ERROR = "network_error"
    PARSE_ERROR = "parse_error"
    RATE_LIMIT = "rate_limit_exceeded"
    INVALID_INPUT = "invalid_input"

def handle_error(error: Exception, error_type: ErrorType) -> str:
    logger.error(f"{error_type.value}: {str(error)}")
    return f"Fehler ({error_type.value}): {str(error)}"

4. Eingabeverifikation

Verifizieren Sie Benutzereingaben vor der Verarbeitung:

from urllib.parse import urlparse

def validate_url(url: str) -> bool:
    try:
        result = urlparse(url)
        return all([result.scheme, result.netloc])
    except:
        return False

def validate_query(query: str) -> bool:
    return len(query.strip()) > 0 and len(query) < 500

Bereitstellungsüberlegungen

Verwendung von SSE-Transport für Web-Bereitstellung

Für webbasierte Bereitstellungen verwenden Sie den SSE (Server-Sent Events) Transport. Wenn Sie eine serverlose Bereitstellung in Betracht ziehen, könnten Sie daran interessiert sein, die Leistung von AWS Lambda mit JavaScript, Python und Golang zu vergleichen, um eine fundierte Entscheidung über Ihre Laufzeitumgebung zu treffen:

import mcp.server.sse

async def main_sse():
    """Führen Sie den Server mit SSE-Transport aus"""
    from starlette.applications import Starlette
    from starlette.routing import Mount

    sse = mcp.server.sse.SseServerTransport("/messages")

    starlette_app = Starlette(
        routes=[
            Mount("/mcp", app=sse.get_server())
        ]
    )

    import uvicorn
    await uvicorn.Server(
        config=uvicorn.Config(starlette_app, host="0.0.0.0", port=8000)
    ).serve()

AWS Lambda-Bereitstellung

MCP-Server können auch als AWS Lambda-Funktionen bereitgestellt werden, insbesondere bei Verwendung des SSE-Transports. Für umfassende Anleitungen zur Lambda-Bereitstellung:

Docker-Bereitstellung

Erstellen Sie eine Dockerfile für die containerisierte Bereitstellung:

FROM python:3.11-slim

WORKDIR /app

# Installieren Sie Systemabhängigkeiten
RUN apt-get update && apt-get install -y \
    wget \
    && rm -rf /var/lib/apt/lists/*

# Installieren Sie Python-Abhängigkeiten
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install chromium
RUN playwright install-deps

# Kopieren Sie die Anwendung
COPY mcp_server.py .

CMD ["python", "mcp_server.py"]

Leistungsoptimierung

Asynchrone Operationen

Verwenden Sie asyncio für gleichzeitige Operationen:

async def search_multiple_queries(queries: list[str]) -> list[str]:
    """Suchen Sie mehrere Abfragen gleichzeitig"""
    tasks = [search_web(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return results

Verbindungspooling

Wiederverwenden Sie Verbindungen für eine bessere Leistung:

import aiohttp

session = None

async def get_session():
    global session
    if session is None:
        session = aiohttp.ClientSession()
    return session

async def fetch_url(url: str) -> str:
    session = await get_session()
    async with session.get(url) as response:
        return await response.text()

Sicherheits-Best Practices

  1. Eingabereinigung: Validieren und bereinigen Sie immer Benutzereingaben
  2. URL-Whitelisting: Überlegen Sie die Implementierung von URL-Whitelisting für das Scraping
  3. Timeout-Kontrollen: Legen Sie geeignete Timeouts fest, um Ressourcenerschöpfung zu verhindern
  4. Inhaltsbegrenzungen: Begrenzen Sie die Größe der gescrapten Inhalte
  5. Authentifizierung: Implementieren Sie Authentifizierung für Produktionsbereitstellungen
  6. HTTPS: Verwenden Sie HTTPS für den SSE-Transport in der Produktion

Arbeit mit verschiedenen LLM-Anbietern

Während MCP von Anthropic für Claude entwickelt wurde, ist das Protokoll so gestaltet, dass es mit jedem LLM funktioniert. Wenn Sie MCP-Server erstellen, die mit mehreren KI-Anbietern interagieren und strukturierte Ausgaben benötigen, möchten Sie unsere Vergleich von strukturierten Ausgaben bei beliebten LLM-Anbietern einschließlich OpenAI, Gemini, Anthropic, Mistral und AWS Bedrock überprüfen.

Verwandte Ressourcen

MCP und Protokollimplementierung

Python-Entwicklung

Web-Scraping und Inhaltsverarbeitung

Serverlose Bereitstellungsressourcen

LLM-Integration

Fazit

Der Aufbau von MCP-Servern in Python eröffnet mächtige Möglichkeiten zur Erweiterung von KI-Assistenten mit benutzerdefinierten Tools und Datenquellen. Die hier demonstrierten Web-Such- und Scraping-Funktionen sind erst der Anfang - Sie können diese Grundlage erweitern, um Datenbanken, APIs, Dateisysteme und virtually jedes externe System zu integrieren.

Das Model Context Protocol entwickelt sich noch weiter, aber sein standardisierter Ansatz zur KI-Tool-Integration macht es zu einer spannenden Technologie für Entwickler, die die nächste Generation von KI-gestützten Anwendungen erstellen. Ob Sie interne Tools für Ihre Organisation erstellen oder öffentliche MCP-Server für die Community, Python bietet eine hervorragende Grundlage für schnelle Entwicklung und Bereitstellung.