Tworzenie serwerów MCP w Pythonie: przewodnik po wyszukiwaniu w sieci i skrapowaniu

Tworzenie serwerów MCP dla asystentów AI z przykładami w Pythonie

Page content

Protokół Kontekstu Modelu (MCP) rewolucjonizuje sposób, w jaki asystenci AI interagują z zewnętrznymi źródłami danych i narzędziami. W tym przewodniku omówimy, jak zbudować serwery MCP w Pythonie, z przykładami skupionymi na możliwościach wyszukiwania w sieci i skrapowania.

MCP robots

Co to jest Protokół Kontekstu Modelu?

Protokół Kontekstu Modelu (MCP) to otwarty protokół wprowadzony przez Anthropic, który standardizuje sposób, w jaki asystenci AI łączą się z zewnętrznymi systemami. Zamiast tworzyć niestandardowe integracje dla każdego źródła danych, MCP oferuje jednolity interfejs, który umożliwia:

  • Asystentom AI (takim jak Claude, ChatGPT lub niestandardowe aplikacje LLM) do odkrywania i korzystania z narzędzi
  • Programistom do udostępniania źródeł danych, narzędzi i wskazówek za pomocą standardowego protokołu
  • Bezproblemową integrację bez ponownego tworzenia koła dla każdego przypadku użycia

Protokół działa na architekturze klient-serwer, gdzie:

  • Klienci MCP (asystenci AI) odkrywają i korzystają z możliwości
  • Serwery MCP udostępniają zasoby, narzędzia i wskazówki
  • Komunikacja odbywa się za pomocą JSON-RPC przez stdio lub HTTP/SSE

Jeśli chcesz zaimplementować serwery MCP w innych językach, sprawdź nasz przewodnik po implementacji serwera MCP w Go, który szczegółowo omawia specyfikacje protokołu i strukturę wiadomości.

Dlaczego budować serwery MCP w Pythonie?

Python to świetny wybór do tworzenia serwerów MCP, ponieważ:

  1. Bogata ekosystema: Biblioteki takie jak requests, beautifulsoup4, selenium i playwright ułatwiają skrapowanie sieci
  2. SDK MCP: Oficjalny SDK Pythona (mcp) oferuje solidną obsługę implementacji serwera
  3. Szybka rozwijalność: Prosta struktura Pythona umożliwia szybkie prototypowanie i iterację
  4. Integracja z AI/ML: Łatwe łączenie się z bibliotekami AI takimi jak langchain, openai i narzędziami do przetwarzania danych
  5. Wsparcie społeczności: Duża społeczność z rozszerzonymi dokumentacjami i przykładami

Konfiguracja środowiska programistycznego

Najpierw utwórz środowisko wirtualne i zainstaluj wymagane zależności. Używanie środowisk wirtualnych jest kluczowe dla izolacji projektów w Pythonie – jeśli potrzebujesz odświeżenia wiedzy, sprawdź nasz cheatsheet do venv dla szczegółowych poleceń i najlepszych praktyk.

# Utwórz i aktywuj środowisko wirtualne
python -m venv mcp-env
source mcp-env/bin/activate  # Na Windows: mcp-env\Scripts\activate

# Zainstaluj SDK MCP i biblioteki do skrapowania sieci
pip install mcp requests beautifulsoup4 playwright lxml
playwright install  # Zainstaluj sterowniki przeglądarki dla Playwright

Modernizowany alternatywa: Jeśli preferujesz szybsze rozwiązywanie zależności i instalację, rozważ użycie uv - nowego zarządcy pakietów i środowisk Pythona, który może być znacznie szybszy niż pip dla dużych projektów.

Budowanie podstawowego serwera MCP

Zaczniemy od minimalnej struktury serwera MCP. Jeśli jesteś nowy w Pythonie lub potrzebujesz szybkiego odniesienia do składni i typowych wzorców, nasz cheatsheet Pythona oferuje kompleksowy przegląd podstaw Pythona.

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

# Utwórz instancję serwera
app = Server("websearch-scraper")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Zdefiniuj dostępne narzędzia"""
    return [
        Tool(
            name="search_web",
            description="Wyszukaj informacje w sieci",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Zapytanie wyszukiwania"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Maksymalna liczba wyników",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Obsłuż wykonanie narzędzia"""
    if name == "search_web":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)
        
        # Zaimplementuj logikę wyszukiwania tutaj
        results = await perform_web_search(query, max_results)
        
        return [TextContent(
            type="text",
            text=f"Wyniki wyszukiwania dla '{query}':\n\n{results}"
        )]
    
    raise ValueError(f"Nieznane narzędzie: {name}")

async def perform_web_search(query: str, max_results: int) -> str:
    """Zastępcza implementacja wyszukiwania"""
    return f"Znaleziono {max_results} wyników dla: {query}"

async def main():
    """Uruchom serwer"""
    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())

Implementacja funkcji wyszukiwania w sieci

Teraz zaimplementujemy rzeczywiste narzędzie do wyszukiwania w sieci za pomocą DuckDuckGo (które nie wymaga kluczy API):

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]:
    """Wyszukaj w DuckDuckGo i przeanalizuj wyniki"""
    
    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"Wyszukiwanie nie powiodło się: {str(e)}")

def format_search_results(results: list[dict]) -> str:
    """Formatuj wyniki wyszukiwania do wyświetlenia"""
    if not results:
        return "Nie znaleziono wyników."
    
    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)

Dodanie możliwości skrapowania sieci

Dodajmy narzędzie do skrapowania i ekstrakcji treści z stron internetowych. Gdy skrapujesz zawartość HTML do użycia z LLM, możesz również chcieć przekonwertować ją na format Markdown dla lepszego przetwarzania. W tym celu sprawdź nasz kompleksowy przewodnik po konwertowaniu HTML na Markdown w Pythonie, który porównuje 6 różnych bibliotek z testami wydajnościowymi i praktycznymi rekomendacjami.

from playwright.async_api import async_playwright
import asyncio

async def scrape_webpage(url: str, selector: str = None) -> dict:
    """Skrapuj zawartość z strony internetowej za pomocą 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)
            
            # Poczekaj, aż zawartość zostanie załadowana
            await page.wait_for_load_state('networkidle')
            
            if selector:
                # Wyodrębnij konkretny element
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "Nie znaleziono selektora"
            else:
                # Wyodrębnij zawartość główną
                content = await page.inner_text('body')
            
            title = await page.title()
            
            return {
                "title": title,
                "content": content[:5000],  # Ogranicz długość zawartości
                "url": url,
                "success": True
            }
            
        except Exception as e:
            return {
                "error": str(e),
                "url": url,
                "success": False
            }
        finally:
            await browser.close()

# Dodaj narzędzie do skrapowania do serwera MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="Wyszukaj w sieci za pomocą DuckDuckGo",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Zapytanie wyszukiwania"},
                    "max_results": {"type": "number", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Skrapuj zawartość z strony internetowej",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "URL do skrapowania"},
                    "selector": {
                        "type": "string",
                        "description": "Opcjonalny selektor CSS dla konkretnego zawartości"
                    }
                },
                "required": ["url"]
            }
        )
    ]

Pełna implementacja serwera MCP

Oto kompletna, gotowa do produkcji implementacja serwera MCP z możliwością wyszukiwania i skrapowania:

#!/usr/bin/env python3
"""
Serwer MCP do wyszukiwania w sieci i skrapowania
Oferuje narzędzia do wyszukiwania w sieci i ekstrakcji treści z stron
"""

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

# Skonfiguruj logowanie
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("websearch-scraper")

# Utwórz serwer
app = Server("websearch-scraper")

# Implementacja wyszukiwania
async def search_web(query: str, max_results: int = 5) -> str:
    """Wyszukaj w DuckDuckGo i zwróć sformatowane wyniki"""
    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 ""
                })
        
        # Sformatuj wyniki
        if not results:
            return "Nie znaleziono wyników."
        
        formatted = [f"Znaleziono {len(results)} wyników dla '{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"Wyszukiwanie nie powiodło się: {e}")
        return f"Błąd wyszukiwania: {str(e)}"

# Implementacja skrapowania
async def scrape_page(url: str, selector: str = None) -> str:
    """Skrapuj zawartość strony internetowej za pomocą 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 "Nie znaleziono selektora"
            else:
                content = await page.inner_text('body')
            
            # Ogranicz długość zawartości
            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"Błąd skrapowania: {e}")
            return f"Błąd skrapowania: {str(e)}"
        finally:
            await browser.close()

# Definicje narzędzi MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    """Wyświetl dostępne narzędzia MCP"""
    return [
        Tool(
            name="search_web",
            description="Wyszukaj w sieci za pomocą DuckDuckGo. Zwraca tytuły, fragmenty i URL-y.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Zapytanie wyszukiwania"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Maksymalna liczba wyników (domyślnie: 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Ekstrahuj zawartość z strony internetowej. Można celować w konkretne elementy za pomocą selektorów CSS.",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "URL do skrapowania"
                    },
                    "selector": {
                        "type": "string",
                        "description": "Opcjonalny selektor CSS do ekstrakcji konkretnego zawartości"
                    }
                },
                "required": ["url"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """Obsłuż wykonanie narzędzia"""
    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"Nieznane narzędzie: {name}")
    
    except Exception as e:
        logger.error(f"Błąd wykonania narzędzia: {e}")
        return [TextContent(
            type="text",
            text=f"Błąd wykonania {name}: {str(e)}"
        )]

async def main():
    """Uruchom serwer MCP"""
    logger.info("Uruchamianie serwera WebSearch-Scraper MCP")
    
    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())

Konfiguracja serwera MCP

Aby użyć swojego serwera MCP z Claude Desktop lub innymi klientami MCP, utwórz plik konfiguracyjny:

Dla Claude Desktop (claude_desktop_config.json):

{
  "mcpServers": {
    "websearch-scraper": {
      "command": "python",
      "args": [
        "/ścieżka/do/Twojego/mcp_server.py"
      ],
      "env": {}
    }
  }
}

Lokalizacja:

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

Testowanie serwera MCP

Utwórz skrypt testowy, aby zweryfikować funkcjonalność:

import asyncio
import json
import sys
from io import StringIO

async def test_mcp_server():
    """Testuj serwer MCP lokalnie"""
    
    # Test wyszukiwania
    print("Testowanie wyszukiwania w sieci...")
    results = await search_web("Python MCP tutorial", 3)
    print(results)
    
    # Test skrapowania
    print("\n\nTestowanie skrapowania strony internetowej...")
    content = await scrape_page("https://example.com")
    print(content[:500])

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

Zaawansowane funkcje i najlepsze praktyki

1. Ograniczanie przepustowości

Zaimplementuj ograniczanie przepustowości, aby uniknąć przeciążania docelowych serwerów:

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("Przekroczono limit przepustowości")
        
        self.requests[key].append(now)

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

2. Buforowanie

Dodaj buforowanie, aby poprawić wydajność:

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. Obsługa błędów

Zaimplementuj solidną obsługę błędów:

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"Błąd ({error_type.value}): {str(error)}"

4. Walidacja danych wejściowych

Waliduj dane wejściowe przed przetwarzaniem:

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

Rozważania dotyczące wdrażania

Użycie transportu SSE dla wdrożeń w sieci

Dla wdrożeń opartych na sieci, użyj transportu SSE (Server-Sent Events). Jeśli rozważasz wdrożenie w środowisku bezserwerowym, możesz być zainteresowany porównaniem wydajności AWS Lambda w języku JavaScript, Python i Golang w celu podejmowania świadomego decyzji o wyborze środowiska uruchomieniowego:

import mcp.server.sse

async def main_sse():
    """Uruchom serwer z transportem SSE"""
    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()

Wdrażanie jako funkcje AWS Lambda

Serwery MCP mogą być również wdrażane jako funkcje AWS Lambda, szczególnie przy użyciu transportu SSE. Dla kompleksowych przewodników dotyczących wdrażania Lambda:

Wdrażanie w kontenerach Docker

Utwórz Dockerfile dla wdrożenia w kontenerach:

FROM python:3.11-slim

WORKDIR /app

# Zainstaluj zależności systemowe
RUN apt-get update && apt-get install -y \
    wget \
    && rm -rf /var/lib/apt/lists/*

# Zainstaluj zależności Pythona
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install chromium
RUN playwright install-deps

# Skopiuj aplikację
COPY mcp_server.py .

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

Optymalizacja wydajności

Operacje asynchroniczne

Użyj asyncio do operacji równoległych:

async def search_multiple_queries(queries: list[str]) -> list[str]:
    """Wyszukaj wiele zapytań równolegle"""
    tasks = [search_web(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return results

Pule połączeń

Odtwarzaj połączenia dla lepszej wydajności:

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

Najlepsze praktyki bezpieczeństwa

  1. Sanitacja danych wejściowych: Zawsze waliduj i sanituj dane wejściowe użytkownika
  2. Białe listy URL: Rozważ zaimplementowanie białej listy URL do scrapowania
  3. Kontrola czasu oczekiwania: Ustaw odpowiednie limity czasu, aby zapobiec wyczerpywaniu zasobów
  4. Limity treści: Ogranicz rozmiar scrapowanych treści
  5. Autoryzacja: Zaimplementuj autoryzację dla wdrożeń produkcyjnych
  6. HTTPS: Używaj HTTPS do transportu SSE w środowiskach produkcyjnych

Praca z różnymi dostawcami LLM

Choć MCP został opracowany przez Anthropic dla Claude, protokół został zaprojektowany tak, aby działać z dowolnym LLM. Jeśli tworzysz serwery MCP, które współpracują z wieloma dostawcami AI i potrzebujesz strukturalnych wyjść, warto przejrzeć nasze porównanie wyjść strukturalnych wśród popularnych dostawców LLM obejmujące OpenAI, Gemini, Anthropic, Mistral i AWS Bedrock.

Przydatne linki i zasoby

Powiązane zasoby

MCP i implementacja protokołu

Rozwój w Pythonie

Scrapowanie sieci i przetwarzanie treści

Zasoby dotyczące wdrożeń bezserwerowych

Integracja z LLM

Podsumowanie

Tworzenie serwerów MCP w Pythonie otwiera potężne możliwości rozszerzania asystentów AI o niestandardowe narzędzia i źródła danych. Możliwości wyszukiwania w sieci i scrapowania przedstawione tutaj to tylko początek – możesz rozszerzyć tę podstawę, aby zintegrować bazy danych, API, systemy plików i niemal każdy zewnętrzny system.

Model Context Protocol nadal ewoluuje, ale jego standaryzowany podejście do integracji narzędzi AI czyni z niego ekscytującą technologię dla programistów tworzących kolejne pokolenia aplikacji zasilanych AI. Niezależnie od tego, czy tworzysz wewnętrzne narzędzia dla swojej organizacji, czy budujesz publiczne serwery MCP dla społeczności, Python oferuje doskonałą podstawę do szybkiego rozwoju i wdrażania.