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

Model Context Protocol (MCP) становится новым стандартом для подключения ИИ-моделей к реальным инструментам и сервисам. Создание MCP-сервера позволяет предоставить данные, действия и ресурсы для LLM, таких как Claude, через простой стандартизированный интерфейс.
В этом руководстве вы пошагово узнаете, как настроить базовый MCP-сервер на Python, определить ресурсы и инструменты, а также подключить его к MCP-клиенту.
Ключевые выводы
- MCP-серверы позволяют ИИ-моделям взаимодействовать с внешними системами через стандартизированные ресурсы и инструменты.
- Вы можете создать MCP-сервер на Python с использованием официального SDK.
- Минимальный рабочий сервер может предоставлять как данные только для чтения (ресурсы), так и исполняемые действия (инструменты).
- Безопасность и обработка ошибок критически важны для производственных развертываний.
Что такое 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 может выполнять.
Поток коммуникации работает следующим образом:
- LLM (через клиент) запрашивает данные или действия с вашего сервера
- Ваш сервер обрабатывает эти запросы и возвращает стандартизированные ответы
- 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-сервера учитывайте следующие лучшие практики безопасности:
- Аутентификация: Реализуйте 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")
- Валидация входных данных: Всегда проверяйте все входные данные с помощью моделей Pydantic.
- Ограничение скорости: Реализуйте ограничение скорости для предотвращения злоупотреблений.
- HTTPS: Всегда используйте HTTPS в производственной среде.
- Ограниченные действия: Определите четкие границы того, что могут делать инструменты.
Оптимизация производительности
- Кэширование: Кэшируйте дорогостоящие запросы ресурсов:
from functools import lru_cache
@lru_cache(maxsize=128, ttl=300) # Cache for 5 minutes
def fetch_user_profiles():
# Expensive database query
pass
- Асинхронная обработка: Используйте 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
- Пулы соединений: Используйте пулы соединений для доступа к базе данных.
Развертывание
Локальная разработка
Для локальной разработки запустите:
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)
- Создайте экземпляр EC2 или используйте AWS App Runner
- Разверните ваш Docker-контейнер
- Настройте Application Load Balancer
- Настройте группы безопасности для ограничения доступа
Тестирование вашего 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-сервера рассмотрите эти расширенные возможности:
- Интеграция с базами данных: Подключение к PostgreSQL, MongoDB или другим базам данных.
- Файловые операции: Добавление инструментов для чтения, записи и преобразования файлов.
- Внешние API: Интеграция с популярными сервисами, такими как GitHub, Slack или Google Drive.
- Вебхуки: Позволяют LLM запускать события в других системах.
- Потоковые ресурсы: Поддержка потоковой передачи больших наборов данных.
- Контекстно-зависимые действия: Добавление инструментов, которые понимают текущий контекст 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, которые его используют.
Часто задаваемые вопросы
Требуется некоторый опыт работы с Python. MCP-серверы — это программные процессы, которые требуют правильного определения ресурсов и инструментов.
Да. Anthropic и контрибьюторы выпускают SDK для нескольких языков, включая Python и TypeScript.
Да. Вы можете разместить свой MCP-сервер на облачных платформах, за брандмауэрами, и сделать его безопасно доступным для ваших LLM-клиентов.