Bygg MCP-servrar i Python: Guide för webbsökning och skrapning

Bygg MCP-server för AI-assistenter med Python-exempel

Sidinnehåll

Model Context Protocol (MCP) revolutionerar hur AI-assistenter interagerar med externa datorkällor och verktyg. I den här guiden kommer vi att utforska hur man bygger MCP servrar i Python, med exempel som fokuserar på webbsökning och skrapning.

MCP robotar

Vad är Model Context Protocol?

Model Context Protocol (MCP) är ett öppet protokoll som introducerats av Anthropic för att standardisera hur AI-assistenter ansluter till externa system. Istället för att bygga anpassade integreringar för varje datorkälla, ger MCP en enhetlig gränssnitt som möjliggör:

  • AI-assistenter (som Claude, ChatGPT eller anpassade LLM-program) att upptäcka och använda verktyg
  • Utvecklare att exponera datorkällor, verktyg och frågor genom ett standardiserat protokoll
  • Enkel integration utan att behöva omfärda hjulet för varje användningss fall

Protokollet fungerar enligt en klient-serverarkitektur där:

  • MCP-klienter (AI-assistenter) upptäcker och använder funktioner
  • MCP-servrar exponerar resurser, verktyg och frågor
  • Kommunikation sker via JSON-RPC över stdio eller HTTP/SSE

Om du är intresserad av att implementera MCP-servrar i andra språk, se vår guide om implementering av MCP-server i Go, som detaljerar protokollspecifikationerna och meddelandestrukturen.

Varför bygga MCP-servrar i Python?

Python är ett utmärkt val för utveckling av MCP-servrar eftersom:

  1. Riktigt ekosystem: Bibliotek som requests, beautifulsoup4, selenium och playwright gör webbskrapning enkel
  2. MCP SDK: Officiell Python SDK (mcp) ger robust stöd för serverimplementering
  3. Snabb utveckling: Pythons enkelhet möjliggör snabb prototypering och iteration
  4. AI/ML-integration: Enkel integration med AI-bibliotek som langchain, openai och dataverktyg
  5. Komponentstöd: Stor gemenskap med omfattande dokumentation och exempel

Att konfigurera din utvecklingsmiljö

Först, skapa en virtuell miljö och installera de nödvändiga beroendena. Att använda virtuella miljöer är viktigt för att isolera Python-projekt – om du behöver en påminnelse, se vår venv CheatSheet för detaljerade kommandon och bästa praxis.

# Skapa och aktivera virtuell miljö
python -m venv mcp-env
source mcp-env/bin/activate  # På Windows: mcp-env\Scripts\activate

# Installera MCP SDK och webbskrapningsbibliotek
pip install mcp requests beautifulsoup4 playwright lxml
playwright install  # Installera webbläsardrivare för Playwright

Modern alternativ: Om du föredrar snabbare beroendelösning och installation, överväg att använda uv - den moderna Python-paket- och miljöhanteraren som kan vara mycket snabbare än pip för stora projekt.

Bygg en grundläggande MCP-server

Låt oss börja med en minimal MCP-serverstruktur. Om du är ny i Python eller behöver en snabb referens för syntax och vanliga mönster, ger vår Python CheatSheet en omfattande översikt över Python-fundament.

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

# Skapa serverinstans
app = Server("websearch-scraper")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Definiera tillgängliga verktyg"""
    return [
        Tool(
            name="search_web",
            description="Sök på nätet efter information",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Sökkommando"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Maximalt antal resultat",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Hantera verktygsutförande"""
    if name == "search_web":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)
        
        # Implementera söklogik här
        results = await perform_web_search(query, max_results)
        
        return [TextContent(
            type="text",
            text=f"Sökrésultat för '{query}':\n\n{results}"
        )]
    
    raise ValueError(f"Okänt verktyg: {name}")

async def perform_web_search(query: str, max_results: int) -> str:
    """Platshållare för faktisk sökimplementation"""
    return f"Funnen {max_results} resultat för: {query}"

async def main():
    """Kör servern"""
    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())

Implementering av webbsökningsfunktioner

Låt oss nu implementera ett verkligt webbsökningsverktyg med hjälp av DuckDuckGo (som inte kräver API-nycklar):

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]:
    """Sök på DuckDuckGo och tolka resultat"""
    
    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"Sökning misslyckades: {str(e)}")

def format_search_results(results: list[dict]) -> str:
    """Formatera sökrésultat för visning"""
    if not results:
        return "Inga resultat hittades."
    
    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)

Lägga till webbskrapningsfunktioner

Låt oss lägga till ett verktyg för att skrapa och extrahera innehåll från webbsidor. När du skrapar HTML-innehåll för användning med LLM:er kan du också vilja konvertera det till Markdown-format för bättre bearbetning. För detta ändamål, se vår omfattande guide om konvertera HTML till Markdown med Python, som jämför 6 olika bibliotek med prestandamått och praktiska rekommendationer.

from playwright.async_api import async_playwright
import asyncio

async def scrape_webpage(url: str, selector: str = None) -> dict:
    """Skrapa innehåll från en webbsida med hjälp av 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)
            
            # Vänta på att innehåll ska laddas
            await page.wait_for_load_state('networkidle')
            
            if selector:
                # Extrahera specifikt element
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "Selector hittades inte"
            else:
                # Extrahera huvudinnehåll
                content = await page.inner_text('body')
            
            title = await page.title()
            
            return {
                "title": title,
                "content": content[:5000],  # Begränsa innehållslängd
                "url": url,
                "success": True
            }
            
        except Exception as e:
            return {
                "error": str(e),
                "url": url,
                "success": False
            }
        finally:
            await browser.close()

# Lägg till skrapningsverktyg i MCP-servern
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="Sök på nätet med hjälp av DuckDuckGo",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Sökkommando"},
                    "max_results": {"type": "number", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Skrapa innehåll från en webbsida",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "URL att skrapa"},
                    "selector": {
                        "type": "string",
                        "description": "Valfri CSS-utväljare för specifikt innehåll"
                    }
                },
                "required": ["url"]
            }
        )
    ]

Komplett MCP-serverimplementering

Här är en komplett, produktionsklar MCP-server med både sök- och skrapningsfunktioner:

#!/usr/bin/env python3
"""
MCP-server för webbsökning och skrapning
Ger verktyg för att söka på nätet och extrahera innehåll från sidor
"""

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

# Konfigurera loggning
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("websearch-scraper")

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

# Sökimplementering
async def search_web(query: str, max_results: int = 5) -> str:
    """Sök på DuckDuckGo och returnera formaterade resultat"""
    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 ""
                })
        
        # Formatera resultat
        if not results:
            return "Inga resultat hittades."
        
        formatted = [f"Funnen {len(results)} resultat 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"Sökning misslyckades: {e}")
        return f"Sökfel: {str(e)}"

# Skraparimplementering
async def scrape_page(url: str, selector: str = None) -> str:
    """Skrapa sidainnehåll med hjälp av 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 hittades inte"
            else:
                content = await page.inner_text('body')
            
            # Begränsa innehållslängd
            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"Skrapning misslyckades: {e}")
            return f"Skrapningsfel: {str(e)}"
        finally:
            await browser.close()

# MCP-verktygsdefinitioner
@app.list_tools()
async def list_tools() -> list[Tool]:
    """Lista tillgängliga MCP-verktyg"""
    return [
        Tool(
            name="search_web",
            description="Sök på nätet med hjälp av DuckDuckGo. Returnerar rubriker, sammanfattningar och URL:er.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Sökkommando"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Maximalt antal resultat (standard: 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Extrahera innehåll från en webbsida. Kan målvara specifika element med CSS-utväljare.",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "URL att skrapa"
                    },
                    "selector": {
                        "type": "string",
                        "description": "Valfri CSS-utväljare för att extrahera specifikt innehåll"
                    }
                },
                "required": ["url"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """Hantera verktygsutförande"""
    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"Okänt verktyg: {name}")
    
    except Exception as e:
        logger.error(f"Verktygsutförande misslyckades: {e}")
        return [TextContent(
            type="text",
            text=f"Fel vid utförande av {name}: {str(e)}"
        )]

async def main():
    """Kör MCP-servern"""
    logger.info("Startar WebSearch-Scraper MCP-servern")
    
    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())

Konfigurera din MCP-server

För att använda din MCP-server med Claude Desktop eller andra MCP-klienter, skapa en konfigurationsfil:

För Claude Desktop (claude_desktop_config.json):

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

Placering:

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

Testa din MCP-server

Skapa ett testskript för att bekräfta funktionen:

import asyncio
import json
import sys
from io import StringIO

async def test_mcp_server():
    """Testa MCP-server lokalt"""
    
    # Testa sökning
    print("Testar webbsökning...")
    results = await search_web("Python MCP tutorial", 3)
    print(results)
    
    # Testa skrapning
    print("\n\nTestar webbsida-skrapning...")
    content = await scrape_page("https://example.com")
    print(content[:500])

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

Avancerade funktioner och bästa praxis

1. Hastighetsbegränsning

Implementera hastighetsbegränsning för att undvika att överväldiga målserverna:

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("Hastighetsbegränsning överskriden")
        
        self.requests[key].append(now)

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

2. Cachning

Lägg till cachning för att förbättra prestanda:

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

Implementera robust felhantering:

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

4. Validering av indata

Validera användarindata före bearbetning:

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

Överväganden vid distribution

Använda SSE-transport för webbdistribution

För webbaserad distribution, använd SSE (Server-Sent Events) transport. Om du överväger serverlöslig distribution, kan du vara intresserad av att jämföra AWS Lambda-prestanda över JavaScript, Python och Golang för att göra ett välgrundat beslut om din körningsmiljö:

import mcp.server.sse

async def main_sse():
    """Kör server med 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-distribution

MCP-servrar kan också distribueras som AWS Lambda-funktioner, särskilt när man använder SSE-transport. För omfattande guider om Lambda-distribution:

Docker-distribution

Skapa en Dockerfile för containeriserad distribution:

FROM python:3.11-slim

WORKDIR /app

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

# Installera Pythonberoenden
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install chromium
RUN playwright install-deps

# Kopiera program
COPY mcp_server.py .

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

Prestandaoptimering

Asynkrona operationer

Använd asyncio för parallella operationer:

async def search_multiple_queries(queries: list[str]) -> list[str]:
    """Sök flera frågor parallellt"""
    tasks = [search_web(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return results

Användning av anslutningspool

Återanvänd anslutningar för bättre prestanda:

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

Säkerhetsrekommendationer

  1. Inmatningsrening: Validera och rensa alltid användarinmatning
  2. URL-vitlista: Tänk på att implementera en URL-vitlista vid skrapning
  3. Tidsgränskontroller: Ställ in lämpliga tidsgränser för att förhindra resursernas uttömning
  4. Innehållsgränser: Begränsa storleken på skrapat innehåll
  5. Autentisering: Implementera autentisering vid produktionsdistributioner
  6. HTTPS: Använd HTTPS för SSE-transport i produktionsmiljöer

Arbeta med olika LLM-leverantörer

Även om MCP utvecklades av Anthropic för Claude är protokollet utformat för att fungera med vilken LLM som helst. Om du bygger MCP-servrar som interagerar med flera AI-leverantörer och behöver strukturerade utdata bör du granska vår jämförelse av strukturerade utdata hos populära LLM-leverantörer inklusive OpenAI, Gemini, Anthropic, Mistral och AWS Bedrock.

Några användbara länkar och resurser

Relaterade resurser

MCP och protokollierimplementering

Pythonutveckling

Webbskrapning och innehållshantering

Resurser för serverlös distribution

LLM-integration

Slutsats

Att bygga MCP-servrar i Python öppnar upp kraftfulla möjligheter att utöka AI-assistenter med anpassade verktyg och datorkällor. De webbsök- och skrapningsförmågorna som visas här är bara början – du kan utöka denna grund för att integrera databaser, API:er, filsystem och nästan vilket extern system som helst.

Model Context Protocol utvecklas fortfarande, men dess standardiserade metod för AI-verktygsintegration gör det till en spännande teknik för utvecklare som bygger nästa generations AI-driven program. Oavsett om du skapar interna verktyg för din organisation eller bygger publika MCP-servrar för gemenskapen ger Python en utmärkt grund för snabb utveckling och distribution.