MCP-servers bouwen in Python: WebSearch & Scrape gids

Maak MCP-servers voor AI-assistenten met Python-voorbeelden

Inhoud

De Model Context Protocol (MCP) is revolutionair voor de manier waarop AI-assistenten met externe gegevensbronnen en tools interacteren. In deze gids bespreken we hoe je MCP-servers in Python kunt bouwen, met voorbeelden gericht op webzoekfuncties en web scraping.

MCP robots

Wat is het Model Context Protocol?

Model Context Protocol (MCP) is een open protocol dat is ingevoerd door Anthropic om te standaardiseren hoe AI-assistenten verbinding maken met externe systemen. In plaats van aangepaste integraties te bouwen voor elke gegevensbron, biedt MCP een geïntegreerde interface die toelaat:

  • AI-assistenten (zoals Claude, ChatGPT of aangepaste LLM-toepassingen) om tools te ontdekken en te gebruiken
  • Ontwikkelaars om gegevensbronnen, tools en prompts te tonen via een gestandaardiseerd protocol
  • Naadloze integratie zonder het wiel opnieuw uit te vinden voor elke toepassing

Het protocol werkt op een client-serverarchitectuur waarbij:

  • MCP-clients (AI-assistenten) ontdekken en gebruiken functionaliteiten
  • MCP-servers tonen bronnen, tools en prompts
  • Communicatie gebeurt via JSON-RPC over stdio of HTTP/SSE

Als je geïnteresseerd bent in het implementeren van MCP-servers in andere talen, raadpleeg dan onze gids over het implementeren van een MCP-server in Go, die de protocolespecificaties en berichtstructuur in detail bespreekt.

Waarom MCP-servers bouwen in Python?

Python is een uitstekende keuze voor het bouwen van MCP-servers omdat:

  1. Rijke Ecosystem: Bibliotheken zoals requests, beautifulsoup4, selenium en playwright maken web scraping eenvoudig
  2. MCP SDK: Officiële Python SDK (mcp) biedt robuuste ondersteuning voor serverimplementatie
  3. Snelle ontwikkeling: De eenvoud van Python maakt snelle prototyping en iteratie mogelijk
  4. AI/ML-integratie: Eenvoudige integratie met AI-bibliotheken zoals langchain, openai en dataprocessing-tools
  5. Communityondersteuning: Grote gemeenschap met uitgebreide documentatie en voorbeelden

Instellen van je ontwikkelomgeving

Maak eerst een virtuele omgeving aan en installeer de benodigde afhankelijkheden. Het gebruik van virtuele omgevingen is essentieel voor isolatie van Python-projecten – als je een herhaling nodig hebt, raadpleeg dan onze venv Cheatsheet voor gedetailleerde opdrachten en best practices.

# Maak en activeer virtuele omgeving
python -m venv mcp-env
source mcp-env/bin/activate  # Op Windows: mcp-env\Scripts\activate

# Installeer MCP SDK en web scraping-bibliotheken
pip install mcp requests beautifulsoup4 playwright lxml
playwright install  # Installeer browserdrivers voor Playwright

Moderne alternatief: Als je snellere afhankelijkheidsoptimalisatie en installatie prefereert, overweeg dan het gebruik van uv - de moderne Python-pakket- en omgevingsbeheerder die aanzienlijk sneller kan zijn dan pip voor grote projecten.

Bouwen van een basis MCP-server

Laten we beginnen met een minimale structuur van een MCP-server. Als je nieuw bent in Python of een snelle verwijzing nodig hebt voor syntaxis en veelvoorkomende patronen, biedt onze Python Cheatsheet een uitgebreid overzicht van Python-fundamenten.

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

# Maak serverinstantie
app = Server("websearch-scraper")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Definieer beschikbare tools"""
    return [
        Tool(
            name="search_web",
            description="Zoek op het internet naar informatie",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Zoekquery"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Maximaal aantal resultaten",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Behandel tooluitvoering"""
    if name == "search_web":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)
        
        # Implementeer zoeklogica hier
        results = await perform_web_search(query, max_results)
        
        return [TextContent(
            type="text",
            text=f"Zoekresultaten voor '{query}':\n\n{results}"
        )]
    
    raise ValueError(f"Onbekende tool: {name}")

async def perform_web_search(query: str, max_results: int) -> str:
    """Tijdelijke implementatie voor zoeklogica"""
    return f"Gevonden {max_results} resultaten voor: {query}"

async def main():
    """Start de server"""
    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())

Implementeren van webzoekfunctionaliteit

Laten we nu een echte webzoektool implementeren met DuckDuckGo (die geen API-sleutels vereist):

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]:
    """Zoek op DuckDuckGo en parse resultaten"""
    
    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"Zoekfout: {str(e)}")

def format_search_results(results: list[dict]) -> str:
    """Formateer zoekresultaten voor weergave"""
    if not results:
        return "Geen resultaten gevonden."
    
    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)

Web scraping-functionaliteit toevoegen

Laten we een tool toevoegen om inhoud van webpagina’s te scrapen. Bij het scrapen van HTML-inhoud voor gebruik met LLMs, wilt u mogelijk ook converteren naar Markdown-formaat voor betere verwerking. Voor dit doel, raadpleeg dan onze uitgebreide gids over het converteren van HTML naar Markdown met Python, die 6 verschillende bibliotheken vergelijkt met benchmarks en praktische aanbevelingen.

from playwright.async_api import async_playwright
import asyncio

async def scrape_webpage(url: str, selector: str = None) -> dict:
    """Scrape inhoud van een webpagina met 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)
            
            # Wacht op inhoud om te laden
            await page.wait_for_load_state('networkidle')
            
            if selector:
                # Extraheer specifiek element
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "Selector niet gevonden"
            else:
                # Extraheer hoofdinhoud
                content = await page.inner_text('body')
            
            title = await page.title()
            
            return {
                "title": title,
                "content": content[:5000],  # Beperk inhoudslengte
                "url": url,
                "success": True
            }
            
        except Exception as e:
            return {
                "error": str(e),
                "url": url,
                "success": False
            }
        finally:
            await browser.close()

# Voeg scraper-tool toe aan de MCP-server
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="Zoek op het internet met DuckDuckGo",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Zoekquery"},
                    "max_results": {"type": "number", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Scrape inhoud van een webpagina",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "URL om te scrapeeren"},
                    "selector": {
                        "type": "string",
                        "description": "Optionele CSS-selector voor specifieke inhoud"
                    }
                },
                "required": ["url"]
            }
        )
    ]

Volledige MCP-serverimplementatie

Hieronder volgt een volledige, productie-klare MCP-server met zowel zoek- als scrapefunctionaliteit:

#!/usr/bin/env python3
"""
MCP-server voor webzoekfuncties en web scraping
Biedt tools voor het zoeken op het internet en het extraheren van inhoud van pagina's
"""

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

# Stel logregistratie in
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("websearch-scraper")

# Maak server
app = Server("websearch-scraper")

# Zoekimplementatie
async def search_web(query: str, max_results: int = 5) -> str:
    """Zoek op DuckDuckGo en retourneer geverifieerde resultaten"""
    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 ""
                })
        
        # Formateer resultaten
        if not results:
            return "Geen resultaten gevonden."
        
        formatted = [f"Gevonden {len(results)} resultaten voor '{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"Zoekfout: {e}")
        return f"Zoekfout: {str(e)}"

# Scraperimplementatie
async def scrape_page(url: str, selector: str = None) -> str:
    """Scrape webpagina-inhoud met 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 niet gevonden"
            else:
                content = await page.inner_text('body')
            
            # Beperk inhoudslengte
            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"Scrapingfout: {e}")
            return f"Scrapingfout: {str(e)}"
        finally:
            await browser.close()

# MCP-tooldefinities
@app.list_tools()
async def list_tools() -> list[Tool]:
    """Lijst beschikbare MCP-tools"""
    return [
        Tool(
            name="search_web",
            description="Zoek op het internet met DuckDuckGo. Retourneert titels, samenvattingen en URLs.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "De zoekquery"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Maximaal aantal resultaten (standaard: 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Extraheer inhoud van een webpagina. Kan specifieke elementen met CSS-selectors doelen.",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "De URL om te scrapeeren"
                    },
                    "selector": {
                        "type": "string",
                        "description": "Optionele CSS-selector om specifieke inhoud te extraheren"
                    }
                },
                "required": ["url"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """Behandel tooluitvoering"""
    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"Onbekende tool: {name}")
    
    except Exception as e:
        logger.error(f"Tooluitvoering gefaald: {e}")
        return [TextContent(
            type="text",
            text=f"Fout bij het uitvoeren van {name}: {str(e)}"
        )]

async def main():
    """Start de MCP-server"""
    logger.info("Starten van WebSearch-Scraper MCP-server")
    
    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())

Instellen van je MCP-server

Om je MCP-server te gebruiken met Claude Desktop of andere MCP-clients, maak een configuratiebestand aan:

Voor Claude Desktop (claude_desktop_config.json):

{
  "mcpServers": {
    "websearch-scraper": {
      "command": "python",
      "args": [
        "/pad/naar/je/mcp_server.py"
      ],
      "env": {}
    }
  }
}

Locatie:

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

Testen van je MCP-server

Maak een testscript aan om de functionaliteit te verifiëren:

import asyncio
import json
import sys
from io import StringIO

async def test_mcp_server():
    """Test MCP-server lokaal"""
    
    # Test zoekfunctie
    print("Testen van webzoekfunctie...")
    results = await search_web("Python MCP tutorial", 3)
    print(results)
    
    # Test scraper
    print("\n\nTesten van webpagina-scraper...")
    content = await scrape_page("https://example.com")
    print(content[:500])

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

Geavanceerde functies en best practices

1. Beperking van aanvragen

Implementeer beperking van aanvragen om te voorkomen dat doelservers overbelast raken:

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("Aanvraaglimiet overschreden")
        
        self.requests[key].append(now)

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

2. Caching

Voeg caching toe om prestaties te verbeteren:

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. Foutafhandeling

Implementeer robuuste foutafhandeling:

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"Fout ({error_type.value}): {str(error)}"

4. Invoervalidatie

Valideer gebruikersinvoer voordat deze wordt verwerkt:

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

Overwegingen bij implementatie

Gebruik van SSE-transport voor webimplementatie

Voor webimplementaties, gebruik SSE (Server-Sent Events) transport. Als je overweegt serverloze implementatie, zou je mogelijk geïnteresseerd zijn in het vergelijken van AWS Lambda-prestaties over JavaScript, Python en Golang om een weloverwogen beslissing te nemen over je runtime:

import mcp.server.sse

async def main_sse():
    """Start server met SSE-transport"""
    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-implementatie

MCP-servers kunnen ook als AWS Lambda-functies worden geïmplementeerd, vooral wanneer SSE-transport wordt gebruikt. Voor uitgebreide gidsen over Lambda-implementatie:

Docker-implementatie

Maak een Dockerfile voor containerimplementatie:

FROM python:3.11-slim

WORKDIR /app

# Installeer systeemafhankelijkheden
RUN apt-get update && apt-get install -y \
    wget \
    && rm -rf /var/lib/apt/lists/*

# Installeer Python-afhankelijkheden
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install chromium
RUN playwright install-deps

# Kopieer toepassing
COPY mcp_server.py .

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

Prestatiesoptimalisatie

Async-acties

Gebruik asyncio voor concurrente acties:

async def search_multiple_queries(queries: list[str]) -> list[str]:
    """Zoek meerdere queries tegelijk"""
    tasks = [search_web(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return results

Verbindingsherbruik

Herbruik verbindingen voor betere prestaties:

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()

Beveiligingsbest practices

  1. Invoerontsmetting: Valideer en ontmetsel altijd gebruikersinvoer
  2. URL-witlijst: Overweeg het implementeren van een URL-witlijst voor web scraping
  3. Time-outbeheer: Stel geschikte time-outs in om resourceverbruik te voorkomen
  4. Inhoudslimieten: Beperk de grootte van gescrapte inhoud
  5. Authenticatie: Implementeer authenticatie voor productieimplementaties
  6. HTTPS: Gebruik HTTPS voor SSE-transport in productie

Werken met verschillende LLM-aanbieders

Hoewel MCP is ontwikkeld door Anthropic voor Claude, is het protocol ontworpen om te werken met elke LLM. Als je MCP-servers bouwt die interactie hebben met meerdere AI-aanbieders en gestructureerde uitvoer nodig hebt, wil je onze vergelijking van gestructureerde uitvoer bij populaire LLM-aanbieders bekijken, inclusief OpenAI, Gemini, Anthropic, Mistral en AWS Bedrock.

Gerelateerde bronnen

MCP en protocolimplementatie

Pythonontwikkeling

Web scraping en inhoudsverwerking

Serverless-implementatiebronnen

LLM-integratie

Conclusie

Het bouwen van MCP-servers in Python opent krachtige mogelijkheden voor het uitbreiden van AI-assistenten met aangepaste tools en gegevensbronnen. De webzoek- en scrapingsfunctionaliteiten die hier worden getoond, zijn slechts het begin—je kunt deze basis uitbreiden om databases, APIs, bestandssystemen en vrijwel elk extern systeem te integreren.

Het Model Context Protocol evolueert nog steeds, maar zijn gestandaardiseerde aanpak voor AI-toolintegratie maakt het een opwindende technologie voor ontwikkelaars die de volgende generatie AI-gemotoriseerde toepassingen bouwen. Of je nu internale tools voor je organisatie maakt of openbare MCP-servers bouwt voor de gemeenschap, biedt Python een uitstekende basis voor snelle ontwikkeling en implementatie.