Как создать MCP-сервер: пошаговое руководство с примерами кода

Model Context Protocol (MCP) становится новым стандартом для подключения ИИ-моделей к реальным инструментам и сервисам. Создание MCP-сервера позволяет предоставить данные, действия и ресурсы для LLM, таких как Claude, через простой стандартизированный интерфейс.

В этом руководстве вы пошагово узнаете, как настроить базовый MCP-сервер на Python, определить ресурсы и инструменты, а также подключить его к MCP-клиенту.

Что такое MCP-сервер

MCP-сервер выступает в роли моста между LLM и внешней системой, такой как база данных, файловое хранилище или API. Он определяет ресурсы (читаемые данные), инструменты (действия) и подсказки (инструкции) таким образом, чтобы LLM мог безопасно использовать их во время выполнения своих задач.

Вместо написания пользовательской интеграции для каждой модели или инструмента, MCP предлагает универсальный стандарт, который работает с версией протокола 0.1 (актуальной на апрель 2025 года).

Что вам нужно перед началом работы

• Python 3.8 или новее
• Базовый опыт работы с Python-скриптами
• MCP SDK для Python (доступен через pip)
• MCP-совместимый клиент, такой как Claude Desktop или Cursor (опционально для тестирования)
• Git для контроля версий (рекомендуется)
• Текстовый редактор или IDE (рекомендуется Visual Studio Code)

Понимание основной структуры

В MCP:

• Сервер: Предоставляет ресурсы и инструменты для LLM.
• Клиент: Соединяет LLM с вашим сервером.
• Протокол: Управляет коммуникацией между клиентом и сервером.

Вы определите два важных примитива:

• Ресурс: Статическая или динамическая информация, которую LLM может читать.
• Инструмент: Вызываемая функция, которую LLM может выполнять.

Поток коммуникации работает следующим образом:

1. LLM (через клиент) запрашивает данные или действия с вашего сервера
2. Ваш сервер обрабатывает эти запросы и возвращает стандартизированные ответы
3. LLM может затем использовать эту информацию в своих рассуждениях и ответах

1. Настройка проекта Python

Начните с создания директории проекта и виртуального окружения Python.

mkdir my_mcp_server
cd my_mcp_server
python -m venv venv
source venv/bin/activate  # Linux/Mac
venvScriptsactivate     # Windows

Создайте базовую структуру проекта:

mkdir -p src/resources src/tools tests
touch src/__init__.py src/resources/__init__.py src/tools/__init__.py
touch requirements.txt README.md

Добавьте следующее в ваш requirements.txt:

mcp-server>=0.1.0
pydantic>=2.0.0
pytest>=7.0.0

2. Установка MCP SDK

Установите MCP SDK для Python и другие зависимости:

pip install -r requirements.txt

Если официальный SDK еще не опубликован, возможно, вам потребуется установить его из репозитория GitHub:

pip install git+https://github.com/anthropic/mcp-server-python.git

3. Создание базового MCP-сервера

Создайте файл src/server.py:

from typing import Dict, Any
from mcp_server import MCPServer
import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("mcp_server")

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Resources and tools will be added here
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

if __name__ == "__main__":
    main()

Это настраивает базовый MCP-сервер с правильным логированием.

4. Определение ресурса

Ресурсы предоставляют данные, которые модель может читать. Создадим файл src/resources/user_profiles.py:

from typing import List, Dict, Any
from pydantic import BaseModel
import logging

logger = logging.getLogger("mcp_server.resources")

class UserProfile(BaseModel):
    """Data model for user profiles."""
    name: str
    role: str
    department: str = "General"
    years_experience: int = 0

def fetch_user_profiles() -> List[Dict[str, Any]]:
    """
    Fetch user profiles from the database.
    
    Returns:
        List[Dict[str, Any]]: A list of user profile dictionaries.
    """
    try:
        # In a real implementation, this would query a database
        # For this example, we'll return mock data
        users = [
            UserProfile(name="Alice", role="Engineer", department="Engineering", years_experience=5),
            UserProfile(name="Bob", role="Product Manager", department="Product", years_experience=3),
            UserProfile(name="Charlie", role="Designer", department="Design", years_experience=7)
        ]
        
        logger.info(f"Successfully fetched {len(users)} user profiles")
        return [user.model_dump() for user in users]
    except Exception as e:
        logger.error(f"Error fetching user profiles: {e}")
        # In production, you might want to return an empty list or raise
        # a specific exception depending on your error handling strategy
        return []

Теперь обновите src/server.py, чтобы включить этот ресурс:

from typing import Dict, Any
from mcp_server import MCPServer, Resource
import logging
from src.resources.user_profiles import fetch_user_profiles

# ... existing code ...

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Add the user profiles resource
        user_profiles = Resource(
            name="user_profiles",
            description="List of user profiles from the company database.",
            fetch_fn=fetch_user_profiles
        )
        
        server.add_resource(user_profiles)
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

if __name__ == "__main__":
    main()

Теперь LLM может запрашивать user_profiles через MCP-клиент.

5. Определение инструмента

Инструменты позволяют LLM выполнять действия. Создайте файл src/tools/user_management.py:

from typing import Dict, Any, Optional
from pydantic import BaseModel, Field, ValidationError
import logging

logger = logging.getLogger("mcp_server.tools")

class CreateUserRequest(BaseModel):
    """Validation model for user creation requests."""
    name: str = Field(..., min_length=2, description="User's full name")
    role: str = Field(..., min_length=2, description="User's job role")
    department: Optional[str] = Field("General", description="User's department")
    years_experience: Optional[int] = Field(0, ge=0, description="Years of professional experience")

def create_user_profile(request_data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Create a new user profile in the database.
    
    Args:
        request_data (Dict[str, Any]): User data containing name, role, etc.
        
    Returns:
        Dict[str, Any]: Response with status and user info
    """
    try:
        # Validate the input data
        user_data = CreateUserRequest(**request_data)
        
        # In a real implementation, this would insert into a database
        # For this example, we'll just log the action
        logger.info(f"Creating new user: {user_data.name} - {user_data.role} in {user_data.department}")
        
        # Return success response with created user data
        return {
            "status": "success",
            "message": f"User {user_data.name} created successfully",
            "user": user_data.model_dump()
        }
    except ValidationError as e:
        # Handle validation errors
        logger.error(f"Validation error: {e}")
        return {
            "status": "error",
            "message": "Invalid user data provided",
            "details": str(e)
        }
    except Exception as e:
        # Handle other errors
        logger.error(f"Error creating user: {e}")
        return {
            "status": "error",
            "message": "Failed to create user",
            "details": str(e)
        }

Теперь обновите src/server.py, чтобы включить этот инструмент:

from typing import Dict, Any
from mcp_server import MCPServer, Resource, Tool
import logging
from src.resources.user_profiles import fetch_user_profiles
from src.tools.user_management import create_user_profile

# ... existing code ...

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Add the user profiles resource
        user_profiles = Resource(
            name="user_profiles",
            description="List of user profiles from the company database.",
            fetch_fn=fetch_user_profiles
        )
        
        # Add the create user tool
        create_user = Tool(
            name="create_user_profile",
            description="Create a new user profile in the database.",
            parameters={
                "name": {"type": "string", "description": "User's full name"},
                "role": {"type": "string", "description": "User's job role"},
                "department": {"type": "string", "description": "User's department (optional)"},
                "years_experience": {"type": "integer", "description": "Years of experience (optional)"}
            },
            execute_fn=create_user_profile
        )
        
        server.add_resource(user_profiles)
        server.add_tool(create_user)
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

6. Обработка ошибок и валидация

Создайте файл src/utils/validation.py для централизации логики валидации:

from typing import Dict, Any, List, Optional, Type
from pydantic import BaseModel, ValidationError
import logging

logger = logging.getLogger("mcp_server.validation")

def validate_request(
    data: Dict[str, Any],
    model_class: Type[BaseModel]
) -> tuple[Optional[BaseModel], Optional[Dict[str, Any]]]:
    """
    Validate request data against a Pydantic model.
    
    Args:
        data: The input data to validate
        model_class: The Pydantic model class to use for validation
        
    Returns:
        tuple: (validated_model, error_dict)
            - If valid: (model instance, None)
            - If invalid: (None, error dictionary)
    """
    try:
        validated_data = model_class(**data)
        return validated_data, None
    except ValidationError as e:
        errors = e.errors()
        error_dict = {
            "status": "error",
            "message": "Validation failed",
            "errors": errors
        }
        logger.error(f"Validation error: {errors}")
        return None, error_dict

Эта утилитарная функция может использоваться во всех ваших инструментах для последовательной валидации входных данных.

7. Запуск и тестирование MCP-сервера

Создайте простой тестовый скрипт test_server.py для проверки работы вашего сервера:

import requests
import json
import time
import subprocess
import sys
from pathlib import Path

def test_server():
    """Simple test to verify the MCP server is running correctly."""
    # Start the server in a separate process
    server_process = subprocess.Popen([sys.executable, "src/server.py"])
    
    try:
        # Wait for server to start
        time.sleep(2)
        
        # Test the server using the MCP client
        # In a real test, you would use the MCP client SDK
        # For this example, we'll simulate a client using HTTP requests
        
        # Assuming the server is running on localhost:8000
        base_url = "http://localhost:8000"
        
        # Test fetching resources
        response = requests.get(f"{base_url}/resources/user_profiles")
        assert response.status_code == 200
        data = response.json()
        print("Resource response:", json.dumps(data, indent=2))
        
        # Test executing a tool
        tool_data = {
            "name": "Test User",
            "role": "Tester",
            "department": "QA"
        }
        response = requests.post(
            f"{base_url}/tools/create_user_profile",
            json=tool_data
        )
        assert response.status_code == 200
        data = response.json()
        print("Tool response:", json.dumps(data, indent=2))
        
        print("All tests passed!")
        
    finally:
        # Clean up: terminate the server process
        server_process.terminate()
        server_process.wait()

if __name__ == "__main__":
    test_server()

Запустите ваш сервер:

python src/server.py

В отдельном терминале вы можете увидеть примерно такой вывод, когда сервер запущен:

2025-04-28 10:15:23 - mcp_server - INFO - Starting MCP server...
2025-04-28 10:15:23 - mcp_server - INFO - Server listening on 0.0.0.0:8000
2025-04-28 10:15:30 - mcp_server.resources - INFO - Successfully fetched 3 user profiles
2025-04-28 10:15:35 - mcp_server.tools - INFO - Creating new user: Test User - Tester in QA

Затем настройте ваш MCP-клиент (например, Claude Desktop) для подключения к вашему локальному MCP-серверу, указав URL сервера или команду для запуска сервера.

Соображения безопасности

При развертывании MCP-сервера учитывайте следующие лучшие практики безопасности:

1. Аутентификация: Реализуйте API-ключи или OAuth для аутентификации клиентов.

def authenticate_request(request):
    api_key = request.headers.get("X-API-Key")
    if not api_key or api_key != os.environ.get("MCP_API_KEY"):
        raise ValueError("Invalid API key")

2. Валидация входных данных: Всегда проверяйте все входные данные с помощью моделей Pydantic.
3. Ограничение скорости: Реализуйте ограничение скорости для предотвращения злоупотреблений.
4. HTTPS: Всегда используйте HTTPS в производственной среде.
5. Ограниченные действия: Определите четкие границы того, что могут делать инструменты.

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

1. Кэширование: Кэшируйте дорогостоящие запросы ресурсов:

from functools import lru_cache

@lru_cache(maxsize=128, ttl=300)  # Cache for 5 minutes
def fetch_user_profiles():
    # Expensive database query
    pass

2. Асинхронная обработка: Используйте async для операций, ограниченных вводом-выводом:

async def fetch_user_profiles():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com/users") as response:
            data = await response.json()
            return data

3. Пулы соединений: Используйте пулы соединений для доступа к базе данных.

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

Локальная разработка

Для локальной разработки запустите:

python src/server.py

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

Создайте Dockerfile:

FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["python", "src/server.py"]

Соберите и запустите:

docker build -t mcp-server .
docker run -p 8000:8000 mcp-server

Облачное развертывание (AWS)

1. Создайте экземпляр EC2 или используйте AWS App Runner
2. Разверните ваш Docker-контейнер
3. Настройте Application Load Balancer
4. Настройте группы безопасности для ограничения доступа

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

Создайте тестовый файл tests/test_resources.py:

import pytest
from src.resources.user_profiles import fetch_user_profiles

def test_fetch_user_profiles():
    """Test that user profiles can be fetched successfully."""
    profiles = fetch_user_profiles()
    
    # Check structure
    assert isinstance(profiles, list)
    assert len(profiles) > 0
    
    # Check content
    first_profile = profiles[0]
    assert "name" in first_profile
    assert "role" in first_profile
    assert isinstance(first_profile["name"], str)

Запустите тесты с помощью:

pytest

Распространенные ошибки и устранение неполадок

Проблема Решение Пример Невозможно подключиться к MCP-серверу Проверьте, что ваш сервер запущен и порт правильный netstat -tulpn | grep 8000 LLM не может найти ресурсы Убедитесь, что поля name и description правильно установлены Проверьте инициализацию вашего ресурса Ошибки при выполнении инструмента Проверьте, что входные параметры соответствуют ожидаемым типам Используйте Pydantic для валидации Клиент не может разобрать вывод Убедитесь, что ваши функции возвращают данные, сериализуемые в JSON Используйте .model_dump() вместо пользовательских объектов Сервер падает при запуске Проверьте ваши импорты и переменные окружения Установите DEBUG=True для подробного логирования Тайм-аут инструмента Добавьте обработку тайм-аутов для внешних API-вызовов Используйте asyncio.wait_for() с тайм-аутом Ошибки аутентификации Проверьте API-ключи и разрешения Проверьте заголовки запросов Ошибки парсинга XML/JSON Используйте правильные заголовки типа содержимого Установите Content-Type: application/json

Следующие шаги

После создания базового MCP-сервера рассмотрите эти расширенные возможности:

1. Интеграция с базами данных: Подключение к PostgreSQL, MongoDB или другим базам данных.
2. Файловые операции: Добавление инструментов для чтения, записи и преобразования файлов.
3. Внешние API: Интеграция с популярными сервисами, такими как GitHub, Slack или Google Drive.
4. Вебхуки: Позволяют LLM запускать события в других системах.
5. Потоковые ресурсы: Поддержка потоковой передачи больших наборов данных.
6. Контекстно-зависимые действия: Добавление инструментов, которые понимают текущий контекст LLM.

Пример: Добавление подключения к базе данных

import psycopg2
from contextlib import contextmanager

@contextmanager
def get_db_connection():
    """Create a database connection context manager."""
    conn = None
    try:
        conn = psycopg2.connect(
            host=os.environ.get("DB_HOST"),
            database=os.environ.get("DB_NAME"),
            user=os.environ.get("DB_USER"),
            password=os.environ.get("DB_PASSWORD")
        )
        yield conn
    finally:
        if conn is not None:
            conn.close()

def fetch_user_profiles_from_db():
    """Fetch user profiles from a PostgreSQL database."""
    with get_db_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT name, role, department FROM users")
            columns = [desc[0] for desc in cur.description]
            return [dict(zip(columns, row)) for row in cur.fetchall()]

Заключение

Создание простого MCP-сервера на Python открывает дверь к тому, чтобы сделать LLM гораздо мощнее. Предоставляя данные и действия через чистый, стандартизированный протокол, вы облегчаете ИИ-системам безопасное и осмысленное взаимодействие с внешними сервисами.

Начните с малого, сосредоточьтесь на одном ресурсе и одном инструменте, и со временем вы сможете расширить возможности до более продвинутых случаев использования, таких как базы данных, облачное хранилище или внутренние API.

Экосистема MCP быстро растет, и внедрение этих стандартов сейчас позволит вашим приложениям извлечь выгоду из улучшений как в протоколе, так и в LLM, которые его используют.

Часто задаваемые вопросы