Создание серверов MCP на Python: руководство по веб-поиску и парсингу

Создавайте серверы MCP для ИИ-ассистентов с примерами на Python

Содержимое страницы

Протокол Контекста Модели (MCP) революционизирует способ взаимодействия ИИ-ассистентов с внешними источниками данных и инструментами. В этом руководстве мы исследуем, как строить MCP-серверы на Python, с примерами, сосредоточенными на возможностях веб-поиска и парсинга.

MCP роботы

Что такое Протокол Контекста Модели?

Протокол Контекста Модели (MCP) — это открытый протокол, представленный компанией Anthropic, чтобы стандартизировать способ подключения ИИ-ассистентов к внешним системам. Вместо создания пользовательских интеграций для каждого источника данных, MCP предоставляет унифицированный интерфейс, который позволяет:

  • ИИ-ассистентам (таким как Claude, ChatGPT или пользовательские приложения LLM) обнаруживать и использовать инструменты
  • Разработчикам предоставлять источники данных, инструменты и запросы через стандартизированный протокол
  • Плавную интеграцию без необходимости изобретать колесо для каждого случая использования

Протокол работает на архитектуре клиент-сервер, где:

  • MCP Клиенты (ИИ-ассистенты) обнаруживают и используют возможности
  • MCP Серверы предоставляют ресурсы, инструменты и запросы
  • Общение происходит через JSON-RPC по stdio или HTTP/SSE

Если вас интересует реализация MCP-серверов на других языках, ознакомьтесь с нашим руководством по реализации MCP-сервера на Go, которое подробно рассматривает спецификации протокола и структуру сообщений.

Почему стоит строить MCP-серверы на Python?

Python — отличный выбор для разработки MCP-серверов, потому что:

  1. Богатая экосистема: Библиотеки, такие как requests, beautifulsoup4, selenium и playwright, делают веб-парсинг простым
  2. SDK MCP: Официальный Python SDK (mcp) предоставляет надежную поддержку реализации сервера
  3. Быстрая разработка: Простота Python позволяет быстро создавать прототипы и итерации
  4. Интеграция с ИИ/ML: Легкая интеграция с библиотеками ИИ, такими как langchain, openai, и инструментами обработки данных
  5. Поддержка сообщества: Большое сообщество с обширной документацией и примерами

Настройка среды разработки

Сначала создайте виртуальную среду и установите необходимые зависимости. Использование виртуальных сред важно для изоляции проектов Python — если вам нужна подсказка, ознакомьтесь с нашим Cheatsheet по venv для подробных команд и лучших практик.

# Создание и активация виртуальной среды
python -m venv mcp-env
source mcp-env/bin/activate  # На Windows: mcp-env\Scripts\activate

# Установка SDK MCP и библиотек для веб-парсинга
pip install mcp requests beautifulsoup4 playwright lxml
playwright install  # Установка драйверов браузеров для Playwright

Современная альтернатива: Если вы предпочитаете более быструю установку зависимостей, рассмотрите использование uv - современного менеджера пакетов и сред Python, который может быть значительно быстрее pip для крупных проектов.

Построение базового MCP-сервера

Начнем с минимальной структуры MCP-сервера. Если вы новичок в Python или вам нужна быстрая справка по синтаксису и общим паттернам, наш Cheatsheet по Python предоставляет всесторонний обзор основ Python.

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

# Создание экземпляра сервера
app = Server("websearch-scraper")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Определение доступных инструментов"""
    return [
        Tool(
            name="search_web",
            description="Поиск информации в интернете",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Поисковый запрос"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Максимальное количество результатов",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Обработка выполнения инструмента"""
    if name == "search_web":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)

        # Реализация логики поиска здесь
        results = await perform_web_search(query, max_results)

        return [TextContent(
            type="text",
            text=f"Результаты поиска для '{query}':\n\n{results}"
        )]

    raise ValueError(f"Неизвестный инструмент: {name}")

async def perform_web_search(query: str, max_results: int) -> str:
    """Заглушка для реальной реализации поиска"""
    return f"Найдено {max_results} результатов для: {query}"

async def main():
    """Запуск сервера"""
    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())

Реализация функциональности веб-поиска

Теперь давайте реализуем реальный инструмент веб-поиска с использованием DuckDuckGo (который не требует 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]:
    """Поиск в DuckDuckGo и парсинг результатов"""

    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"Поиск не удался: {str(e)}")

def format_search_results(results: list[dict]) -> str:
    """Форматирование результатов поиска для отображения"""
    if not results:
        return "Результаты не найдены."

    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)

Добавление возможностей веб-парсинга

Давайте добавим инструмент для парсинга и извлечения контента с веб-страниц. При парсинге HTML-контента для использования с LLM вам также может понадобиться преобразовать его в формат Markdown для лучшей обработки. Для этой цели ознакомьтесь с нашим всесторонним руководством по преобразованию HTML в Markdown с помощью Python, которое сравнивает 6 различных библиотек с бенчмарками и практическими рекомендациями.

from playwright.async_api import async_playwright
import asyncio

async def scrape_webpage(url: str, selector: str = None) -> dict:
    """Парсинг контента с веб-страницы с использованием 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')

            if selector:
                # Извлечение конкретного элемента
                element = await page.query_selector(selector)
                content = await element.inner_text() if element else "Селектор не найден"
            else:
                # Извлечение основного контента
                content = await page.inner_text('body')

            title = await page.title()

            return {
                "title": title,
                "content": content[:5000],  # Ограничение длины контента
                "url": url,
                "success": True
            }

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

# Добавление инструмента парсера в MCP-сервер
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_web",
            description="Поиск в интернете с использованием DuckDuckGo",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Поисковый запрос"},
                    "max_results": {"type": "number", "default": 5}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Парсинг контента с веб-страницы",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "URL для парсинга"},
                    "selector": {
                        "type": "string",
                        "description": "Необязательный CSS-селектор для конкретного контента"
                    }
                },
                "required": ["url"]
            }
        )
    ]

Полная реализация сервера MCP

Вот полная, готовая к производству реализация сервера MCP с возможностями поиска и сбора данных:

#!/usr/bin/env python3
"""
Сервер MCP для поиска в интернете и сбора данных
Предоставляет инструменты для поиска в интернете и извлечения контента с веб-страниц
"""

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

# Настройка логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("websearch-scraper")

# Создание сервера
app = Server("websearch-scraper")

# Реализация поиска
async def search_web(query: str, max_results: int = 5) -> str:
    """Поиск в DuckDuckGo и возврат отформатированных результатов"""
    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 ""
                })

        # Форматирование результатов
        if not results:
            return "Результаты не найдены."

        formatted = [f"Найдено {len(results)} результатов для '{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"Поиск не удался: {e}")
        return f"Ошибка поиска: {str(e)}")

# Реализация сбора данных
async def scrape_page(url: str, selector: str = None) -> str:
    """Сбор контента веб-страницы с использованием 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 "Селектор не найден"
            else:
                content = await page.inner_text('body')

            # Ограничение длины контента
            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"Сбор данных не удался: {e}")
            return f"Ошибка сбора данных: {str(e)}"
        finally:
            await browser.close()

# Определения инструментов MCP
@app.list_tools()
async def list_tools() -> list[Tool]:
    """Список доступных инструментов MCP"""
    return [
        Tool(
            name="search_web",
            description="Поиск в интернете с использованием DuckDuckGo. Возвращает заголовки, описания и URL.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Запрос для поиска"
                    },
                    "max_results": {
                        "type": "number",
                        "description": "Максимальное количество результатов (по умолчанию: 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="scrape_webpage",
            description="Извлечение контента с веб-страницы. Может нацеливать на конкретные элементы с помощью CSS-селекторов.",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "URL для сбора данных"
                    },
                    "selector": {
                        "type": "string",
                        "description": "Необязательный CSS-селектор для извлечения конкретного контента"
                    }
                },
                "required": ["url"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """Обработка выполнения инструмента"""
    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"Неизвестный инструмент: {name}")

    except Exception as e:
        logger.error(f"Выполнение инструмента не удалось: {e}")
        return [TextContent(
            type="text",
            text=f"Ошибка выполнения {name}: {str(e)}"
        )]

async def main():
    """Запуск сервера MCP"""
    logger.info("Запуск сервера MCP WebSearch-Scraper")

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

Настройка вашего сервера MCP

Для использования вашего сервера MCP с Claude Desktop или другими клиентами MCP создайте файл конфигурации:

Для Claude Desktop (claude_desktop_config.json):

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

Расположение:

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

Тестирование вашего сервера MCP

Создайте тестовый скрипт для проверки функциональности:

import asyncio
import json
import sys
from io import StringIO

async def test_mcp_server():
    """Тестирование сервера MCP локально"""

    # Тестирование поиска
    print("Тестирование поиска в интернете...")
    results = await search_web("Python MCP tutorial", 3)
    print(results)

    # Тестирование сбора данных
    print("\n\nТестирование веб-скрейпера...")
    content = await scrape_page("https://example.com")
    print(content[:500])

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

Дополнительные функции и лучшие практики

1. Ограничение скорости

Реализуйте ограничение скорости, чтобы не перегружать целевые серверы:

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("Превышен лимит запросов")

        self.requests[key].append(now)

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

2. Кэширование

Добавьте кэширование для улучшения производительности:

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. Обработка ошибок

Реализуйте надежную обработку ошибок:

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"Ошибка ({error_type.value}): {str(error)}"

4. Валидация входных данных

Проверяйте входные данные пользователя перед обработкой:

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

Рассмотрения развертывания

Использование транспорта SSE для веб-развертывания

Для веб-развертываний используйте транспорт SSE (Server-Sent Events). Если вы рассматриваете serverless-развертывание, возможно, вам будет интересно сравнить производительность AWS Lambda на JavaScript, Python и Golang для принятия обоснованного решения о вашем рантайме:

import mcp.server.sse

async def main_sse():
    """Запуск сервера с транспортом 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()

Развертывание в AWS Lambda

Серверы MCP также могут быть развернуты как функции AWS Lambda, особенно при использовании транспорта SSE. Для всесторонних руководств по развертыванию Lambda:

Развертывание в Docker

Создайте Dockerfile для контейнеризованного развертывания:

FROM python:3.11-slim

WORKDIR /app

# Установка системных зависимостей
RUN apt-get update && apt-get install -y \
    wget \
    && rm -rf /var/lib/apt/lists/*

# Установка зависимостей Python
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install chromium
RUN playwright install-deps

# Копирование приложения
COPY mcp_server.py .

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

Оптимизация производительности

Асинхронные операции

Используйте asyncio для одновременных операций:

async def search_multiple_queries(queries: list[str]) -> list[str]:
    """Поиск нескольких запросов одновременно"""
    tasks = [search_web(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return results

Пулинг соединений

Переиспользуйте соединения для лучшей производительности:

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

Лучшие практики безопасности

  1. Санитизация ввода: Всегда проверяйте и санитизируйте пользовательские входные данные
  2. Белый список URL: Рассмотрите возможность реализации белого списка URL для скрейпинга
  3. Контроль таймаутов: Установите подходящие таймауты, чтобы предотвратить истощение ресурсов
  4. Ограничение контента: Ограничьте размер скрейпинга контента
  5. Аутентификация: Реализуйте аутентификацию для продакшен-развертываний
  6. HTTPS: Используйте HTTPS для транспорта SSE в продакшене

Работа с разными поставщиками LLM

Хотя MCP был разработан Anthropic для Claude, протокол предназначен для работы с любым LLM. Если вы создаете серверы MCP, которые взаимодействуют с несколькими поставщиками ИИ и требуют структурированных выходных данных, вам стоит изучить наше сравнение структурированных выходных данных среди популярных поставщиков LLM включая OpenAI, Gemini, Anthropic, Mistral и AWS Bedrock.

Полезные ссылки и ресурсы

Связанные ресурсы

MCP и реализация протокола

Разработка на Python

Веб-скрейпинг и обработка контента

Ресурсы для serverless-развертывания

Интеграция LLM

Заключение

Создание серверов MCP на Python открывает мощные возможности для расширения ИИ-ассистентов с помощью пользовательских инструментов и источников данных. Демонстрируемые здесь возможности веб-поиска и скрейпинга - это только начало: вы можете расширить эту основу для интеграции баз данных, API, файловых систем и практически любой внешней системы.

Model Context Protocol все еще развивается, но его стандартизированный подход к интеграции ИИ-инструментов делает его захватывающей технологией для разработчиков, создающих следующее поколение приложений, работающих на основе ИИ. Будь то создание внутренних инструментов для вашей организации или разработка публичных серверов MCP для сообщества, Python предоставляет отличную основу для быстрой разработки и развертывания.