# Руководство по реализации сервера MCP на Python

## Обзор

Этот документ содержит лучшие практики и примеры для Python по реализации серверов MCP с помощью MCP Python SDK. Он охватывает настройку сервера, паттерны регистрации инструментов, проверку входных данных с помощью Pydantic, обработку ошибок и полные работающие примеры.

---

## Краткий справочник

### Основные импорты
```python
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field, field_validator, ConfigDict
from typing import Optional, List, Dict, Any
from enum import Enum
import httpx
```

### Инициализация сервера
```python
mcp = FastMCP("service_mcp")
```

### Паттерн регистрации инструментов
```python
@mcp.tool(name="tool_name", annotations={...})
async def tool_function(params: InputModel) -> str:
    # Implementation
    pass
```

---

## MCP Python SDK и FastMCP

Официальный MCP Python SDK предоставляет FastMCP — высокоуровневый фреймворк для создания серверов MCP. Он обеспечивает:
- Автоматическую генерацию description и inputSchema из сигнатур функций и строк документации
- Интеграцию моделей Pydantic для проверки входных данных
- Регистрацию инструментов на основе декоратора `@mcp.tool`

**Чтобы получить полную документацию SDK, используй WebFetch для загрузки:**
`https://raw.githubusercontent.com/modelcontextprotocol/python-sdk/main/README.md`

## Соглашение об именовании серверов

Серверы MCP на Python должны следовать такому шаблону именования:
- **Формат**: `{service}_mcp` (нижний регистр с подчёркиваниями)
- **Примеры**: `github_mcp`, `jira_mcp`, `stripe_mcp`

Имя должно быть:
- Общим (не привязанным к конкретным функциям)
- Описывающим интегрируемый сервис/API
- Легко выводимым из описания задачи
- Без номеров версий или дат

## Реализация инструментов

### Именование инструментов

Используй snake_case для имён инструментов (например, "search_users", "create_project", "get_channel_info") с ясными именами, ориентированными на действия.

**Избегай конфликтов имён**: включай контекст сервиса, чтобы предотвратить совпадения:
- Используй "slack_send_message" вместо просто "send_message"
- Используй "github_create_issue" вместо просто "create_issue"
- Используй "asana_list_tasks" вместо просто "list_tasks"

### Структура инструмента с FastMCP

Инструменты определяются с помощью декоратора `@mcp.tool` и моделей Pydantic для проверки входных данных:

```python
from pydantic import BaseModel, Field, ConfigDict
from mcp.server.fastmcp import FastMCP

# Initialize the MCP server
mcp = FastMCP("example_mcp")

# Define Pydantic model for input validation
class ServiceToolInput(BaseModel):
    '''Input model for service tool operation.'''
    model_config = ConfigDict(
        str_strip_whitespace=True,  # Auto-strip whitespace from strings
        validate_assignment=True,    # Validate on assignment
        extra='forbid'              # Forbid extra fields
    )

    param1: str = Field(..., description="First parameter description (e.g., 'user123', 'project-abc')", min_length=1, max_length=100)
    param2: Optional[int] = Field(default=None, description="Optional integer parameter with constraints", ge=0, le=1000)
    tags: Optional[List[str]] = Field(default_factory=list, description="List of tags to apply", max_items=10)

@mcp.tool(
    name="service_tool_name",
    annotations={
        "title": "Human-Readable Tool Title",
        "readOnlyHint": True,     # Tool does not modify environment
        "destructiveHint": False,  # Tool does not perform destructive operations
        "idempotentHint": True,    # Repeated calls have no additional effect
        "openWorldHint": False     # Tool does not interact with external entities
    }
)
async def service_tool_name(params: ServiceToolInput) -> str:
    '''Tool description automatically becomes the 'description' field.

    This tool performs a specific operation on the service. It validates all inputs
    using the ServiceToolInput Pydantic model before processing.

    Args:
        params (ServiceToolInput): Validated input parameters containing:
            - param1 (str): First parameter description
            - param2 (Optional[int]): Optional parameter with default
            - tags (Optional[List[str]]): List of tags

    Returns:
        str: JSON-formatted response containing operation results
    '''
    # Implementation here
    pass
```

## Основные возможности Pydantic v2

- Используй `model_config` вместо вложенного класса `Config`
- Используй `field_validator` вместо устаревшего `validator`
- Используй `model_dump()` вместо устаревшего `dict()`
- Валидаторам требуется декоратор `@classmethod`
- Для методов-валидаторов обязательны аннотации типов

```python
from pydantic import BaseModel, Field, field_validator, ConfigDict

class CreateUserInput(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,
        validate_assignment=True
    )

    name: str = Field(..., description="User's full name", min_length=1, max_length=100)
    email: str = Field(..., description="User's email address", pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    age: int = Field(..., description="User's age", ge=0, le=150)

    @field_validator('email')
    @classmethod
    def validate_email(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Email cannot be empty")
        return v.lower()
```

## Варианты формата ответа

Для гибкости поддерживай несколько форматов вывода:

```python
from enum import Enum

class ResponseFormat(str, Enum):
    '''Output format for tool responses.'''
    MARKDOWN = "markdown"
    JSON = "json"

class UserSearchInput(BaseModel):
    query: str = Field(..., description="Search query")
    response_format: ResponseFormat = Field(
        default=ResponseFormat.MARKDOWN,
        description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
    )
```

**Формат Markdown**:
- Используй заголовки, списки и форматирование для ясности
- Преобразовывай временные метки в удобочитаемый формат (например, "2024-01-15 10:30:00 UTC" вместо времени эпохи)
- Показывай отображаемые имена с идентификаторами в скобках (например, "@john.doe (U123456)")
- Опускай многословные метаданные (например, показывай только один URL изображения профиля, а не все размеры)
- Логически группируй связанную информацию

**Формат JSON**:
- Возвращай полные структурированные данные, подходящие для программной обработки
- Включай все доступные поля и метаданные
- Используй единообразные имена и типы полей

## Реализация пагинации

Для инструментов, перечисляющих ресурсы:

```python
class ListInput(BaseModel):
    limit: Optional[int] = Field(default=20, description="Maximum results to return", ge=1, le=100)
    offset: Optional[int] = Field(default=0, description="Number of results to skip for pagination", ge=0)

async def list_items(params: ListInput) -> str:
    # Make API request with pagination
    data = await api_request(limit=params.limit, offset=params.offset)

    # Return pagination info
    response = {
        "total": data["total"],
        "count": len(data["items"]),
        "offset": params.offset,
        "items": data["items"],
        "has_more": data["total"] > params.offset + len(data["items"]),
        "next_offset": params.offset + len(data["items"]) if data["total"] > params.offset + len(data["items"]) else None
    }
    return json.dumps(response, indent=2)
```

## Обработка ошибок

Предоставляй ясные сообщения об ошибках, подсказывающие действия:

```python
def _handle_api_error(e: Exception) -> str:
    '''Consistent error formatting across all tools.'''
    if isinstance(e, httpx.HTTPStatusError):
        if e.response.status_code == 404:
            return "Error: Resource not found. Please check the ID is correct."
        elif e.response.status_code == 403:
            return "Error: Permission denied. You don't have access to this resource."
        elif e.response.status_code == 429:
            return "Error: Rate limit exceeded. Please wait before making more requests."
        return f"Error: API request failed with status {e.response.status_code}"
    elif isinstance(e, httpx.TimeoutException):
        return "Error: Request timed out. Please try again."
    return f"Error: Unexpected error occurred: {type(e).__name__}"
```

## Общие утилиты

Выноси общую функциональность в переиспользуемые функции:

```python
# Shared API request function
async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
    '''Reusable function for all API calls.'''
    async with httpx.AsyncClient() as client:
        response = await client.request(
            method,
            f"{API_BASE_URL}/{endpoint}",
            timeout=30.0,
            **kwargs
        )
        response.raise_for_status()
        return response.json()
```

## Лучшие практики Async/Await

Всегда используй async/await для сетевых запросов и операций ввода-вывода:

```python
# Good: Async network request
async def fetch_data(resource_id: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(f"{API_URL}/resource/{resource_id}")
        response.raise_for_status()
        return response.json()

# Bad: Synchronous request
def fetch_data(resource_id: str) -> dict:
    response = requests.get(f"{API_URL}/resource/{resource_id}")  # Blocks
    return response.json()
```

## Аннотации типов

Повсеместно используй аннотации типов:

```python
from typing import Optional, List, Dict, Any

async def get_user(user_id: str) -> Dict[str, Any]:
    data = await fetch_user(user_id)
    return {"id": data["id"], "name": data["name"]}
```

## Строки документации инструментов

Каждый инструмент должен иметь исчерпывающие строки документации с явными сведениями о типах:

```python
async def search_users(params: UserSearchInput) -> str:
    '''
    Search for users in the Example system by name, email, or team.

    This tool searches across all user profiles in the Example platform,
    supporting partial matches and various search filters. It does NOT
    create or modify users, only searches existing ones.

    Args:
        params (UserSearchInput): Validated input parameters containing:
            - query (str): Search string to match against names/emails (e.g., "john", "@example.com", "team:marketing")
            - limit (Optional[int]): Maximum results to return, between 1-100 (default: 20)
            - offset (Optional[int]): Number of results to skip for pagination (default: 0)

    Returns:
        str: JSON-formatted string containing search results with the following schema:

        Success response:
        {
            "total": int,           # Total number of matches found
            "count": int,           # Number of results in this response
            "offset": int,          # Current pagination offset
            "users": [
                {
                    "id": str,      # User ID (e.g., "U123456789")
                    "name": str,    # Full name (e.g., "John Doe")
                    "email": str,   # Email address (e.g., "john@example.com")
                    "team": str     # Team name (e.g., "Marketing") - optional
                }
            ]
        }

        Error response:
        "Error: <error message>" or "No users found matching '<query>'"

    Examples:
        - Use when: "Find all marketing team members" -> params with query="team:marketing"
        - Use when: "Search for John's account" -> params with query="john"
        - Don't use when: You need to create a user (use example_create_user instead)
        - Don't use when: You have a user ID and need full details (use example_get_user instead)

    Error Handling:
        - Input validation errors are handled by Pydantic model
        - Returns "Error: Rate limit exceeded" if too many requests (429 status)
        - Returns "Error: Invalid API authentication" if API key is invalid (401 status)
        - Returns formatted list of results or "No users found matching 'query'"
    '''
```

## Полный пример

Ниже приведён полный пример MCP-сервера на Python:

```python
#!/usr/bin/env python3
'''
MCP Server for Example Service.

This server provides tools to interact with Example API, including user search,
project management, and data export capabilities.
'''

from typing import Optional, List, Dict, Any
from enum import Enum
import httpx
from pydantic import BaseModel, Field, field_validator, ConfigDict
from mcp.server.fastmcp import FastMCP

# Initialize the MCP server
mcp = FastMCP("example_mcp")

# Constants
API_BASE_URL = "https://api.example.com/v1"

# Enums
class ResponseFormat(str, Enum):
    '''Output format for tool responses.'''
    MARKDOWN = "markdown"
    JSON = "json"

# Pydantic Models for Input Validation
class UserSearchInput(BaseModel):
    '''Input model for user search operations.'''
    model_config = ConfigDict(
        str_strip_whitespace=True,
        validate_assignment=True
    )

    query: str = Field(..., description="Search string to match against names/emails", min_length=2, max_length=200)
    limit: Optional[int] = Field(default=20, description="Maximum results to return", ge=1, le=100)
    offset: Optional[int] = Field(default=0, description="Number of results to skip for pagination", ge=0)
    response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="Output format")

    @field_validator('query')
    @classmethod
    def validate_query(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Query cannot be empty or whitespace only")
        return v.strip()

# Shared utility functions
async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
    '''Reusable function for all API calls.'''
    async with httpx.AsyncClient() as client:
        response = await client.request(
            method,
            f"{API_BASE_URL}/{endpoint}",
            timeout=30.0,
            **kwargs
        )
        response.raise_for_status()
        return response.json()

def _handle_api_error(e: Exception) -> str:
    '''Consistent error formatting across all tools.'''
    if isinstance(e, httpx.HTTPStatusError):
        if e.response.status_code == 404:
            return "Error: Resource not found. Please check the ID is correct."
        elif e.response.status_code == 403:
            return "Error: Permission denied. You don't have access to this resource."
        elif e.response.status_code == 429:
            return "Error: Rate limit exceeded. Please wait before making more requests."
        return f"Error: API request failed with status {e.response.status_code}"
    elif isinstance(e, httpx.TimeoutException):
        return "Error: Request timed out. Please try again."
    return f"Error: Unexpected error occurred: {type(e).__name__}"

# Tool definitions
@mcp.tool(
    name="example_search_users",
    annotations={
        "title": "Search Example Users",
        "readOnlyHint": True,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True
    }
)
async def example_search_users(params: UserSearchInput) -> str:
    '''Search for users in the Example system by name, email, or team.

    [Full docstring as shown above]
    '''
    try:
        # Make API request using validated parameters
        data = await _make_api_request(
            "users/search",
            params={
                "q": params.query,
                "limit": params.limit,
                "offset": params.offset
            }
        )

        users = data.get("users", [])
        total = data.get("total", 0)

        if not users:
            return f"No users found matching '{params.query}'"

        # Format response based on requested format
        if params.response_format == ResponseFormat.MARKDOWN:
            lines = [f"# User Search Results: '{params.query}'", ""]
            lines.append(f"Found {total} users (showing {len(users)})")
            lines.append("")

            for user in users:
                lines.append(f"## {user['name']} ({user['id']})")
                lines.append(f"- **Email**: {user['email']}")
                if user.get('team'):
                    lines.append(f"- **Team**: {user['team']}")
                lines.append("")

            return "\n".join(lines)

        else:
            # Machine-readable JSON format
            import json
            response = {
                "total": total,
                "count": len(users),
                "offset": params.offset,
                "users": users
            }
            return json.dumps(response, indent=2)

    except Exception as e:
        return _handle_api_error(e)

if __name__ == "__main__":
    mcp.run()
```

---

## Расширенные возможности FastMCP

### Внедрение параметра Context

FastMCP может автоматически внедрять в инструменты параметр `Context` для расширенных возможностей: журналирования, отчётов о ходе выполнения, чтения ресурсов и взаимодействия с пользователем:

```python
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("example_mcp")

@mcp.tool()
async def advanced_search(query: str, ctx: Context) -> str:
    '''Advanced tool with context access for logging and progress.'''

    # Report progress for long operations
    await ctx.report_progress(0.25, "Starting search...")

    # Log information for debugging
    await ctx.log_info("Processing query", {"query": query, "timestamp": datetime.now()})

    # Perform search
    results = await search_api(query)
    await ctx.report_progress(0.75, "Formatting results...")

    # Access server configuration
    server_name = ctx.fastmcp.name

    return format_results(results)

@mcp.tool()
async def interactive_tool(resource_id: str, ctx: Context) -> str:
    '''Tool that can request additional input from users.'''

    # Request sensitive information when needed
    api_key = await ctx.elicit(
        prompt="Please provide your API key:",
        input_type="password"
    )

    # Use the provided key
    return await api_call(resource_id, api_key)
```

**Возможности Context:**
- `ctx.report_progress(progress, message)` — сообщать о ходе выполнения длительных операций
- `ctx.log_info(message, data)` / `ctx.log_error()` / `ctx.log_debug()` — журналирование
- `ctx.elicit(prompt, input_type)` — запрашивать ввод у пользователей
- `ctx.fastmcp.name` — получать доступ к конфигурации сервера
- `ctx.read_resource(uri)` — читать ресурсы MCP

### Регистрация ресурсов

Предоставляйте данные в виде ресурсов для эффективного доступа на основе шаблонов:

```python
@mcp.resource("file://documents/{name}")
async def get_document(name: str) -> str:
    '''Expose documents as MCP resources.

    Resources are useful for static or semi-static data that doesn't
    require complex parameters. They use URI templates for flexible access.
    '''
    document_path = f"./docs/{name}"
    with open(document_path, "r") as f:
        return f.read()

@mcp.resource("config://settings/{key}")
async def get_setting(key: str, ctx: Context) -> str:
    '''Expose configuration as resources with context.'''
    settings = await load_settings()
    return json.dumps(settings.get(key, {}))
```

**Когда использовать ресурсы, а когда инструменты:**
- **Ресурсы**: для доступа к данным с простыми параметрами (шаблонами URI)
- **Инструменты**: для сложных операций с проверкой данных и бизнес-логикой

### Типы структурированного вывода

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

```python
from typing import TypedDict
from dataclasses import dataclass
from pydantic import BaseModel

# TypedDict for structured returns
class UserData(TypedDict):
    id: str
    name: str
    email: str

@mcp.tool()
async def get_user_typed(user_id: str) -> UserData:
    '''Returns structured data - FastMCP handles serialization.'''
    return {"id": user_id, "name": "John Doe", "email": "john@example.com"}

# Pydantic models for complex validation
class DetailedUser(BaseModel):
    id: str
    name: str
    email: str
    created_at: datetime
    metadata: Dict[str, Any]

@mcp.tool()
async def get_user_detailed(user_id: str) -> DetailedUser:
    '''Returns Pydantic model - automatically generates schema.'''
    user = await fetch_user(user_id)
    return DetailedUser(**user)
```

### Управление жизненным циклом

Инициализируйте ресурсы, которые сохраняются между запросами:

```python
from contextlib import asynccontextmanager

@asynccontextmanager
async def app_lifespan():
    '''Manage resources that live for the server's lifetime.'''
    # Initialize connections, load config, etc.
    db = await connect_to_database()
    config = load_configuration()

    # Make available to all tools
    yield {"db": db, "config": config}

    # Cleanup on shutdown
    await db.close()

mcp = FastMCP("example_mcp", lifespan=app_lifespan)

@mcp.tool()
async def query_data(query: str, ctx: Context) -> str:
    '''Access lifespan resources through context.'''
    db = ctx.request_context.lifespan_state["db"]
    results = await db.query(query)
    return format_results(results)
```

### Варианты транспорта

FastMCP поддерживает два основных транспортных механизма:

```python
# stdio transport (for local tools) - default
if __name__ == "__main__":
    mcp.run()

# Streamable HTTP transport (for remote servers)
if __name__ == "__main__":
    mcp.run(transport="streamable_http", port=8000)
```

**Выбор транспорта:**
- **stdio**: инструменты командной строки, локальные интеграции, выполнение подпроцессов
- **Streamable HTTP**: веб-сервисы, удалённый доступ, несколько клиентов

---

## Рекомендации по написанию кода

### Компонуемость и повторное использование кода

Ваша реализация ДОЛЖНА отдавать приоритет компонуемости и повторному использованию кода:

1. **Выделяйте общую функциональность**:
   - Создавайте повторно используемые вспомогательные функции для операций, применяемых в нескольких инструментах
   - Создавайте общие API-клиенты для HTTP-запросов вместо дублирования кода
   - Централизуйте логику обработки ошибок во вспомогательных функциях
   - Выделяйте бизнес-логику в отдельные функции, которые можно комбинировать
   - Выделяйте общую функциональность выбора и форматирования полей Markdown или JSON

2. **Избегайте дублирования**:
   - НИКОГДА не копируйте и не вставляйте похожий код между инструментами
   - Если вы пишете похожую логику дважды, выделите её в функцию
   - Общие операции, такие как пагинация, фильтрация, выбор полей и форматирование, должны использоваться совместно
   - Логика аутентификации/авторизации должна быть централизована

### Рекомендации, специфичные для Python

1. **Используйте аннотации типов**: всегда указывайте аннотации типов параметров функций и возвращаемых значений
2. **Модели Pydantic**: определяйте понятные модели Pydantic для проверки всех входных данных
3. **Избегайте ручной проверки**: поручайте Pydantic проверку входных данных с помощью ограничений
4. **Правильные импорты**: группируйте импорты (стандартная библиотека, сторонние пакеты, локальные модули)
5. **Обработка ошибок**: используйте конкретные типы исключений (httpx.HTTPStatusError, а не общее Exception)
6. **Асинхронные контекстные менеджеры**: используйте `async with` для ресурсов, требующих освобождения
7. **Константы**: определяйте константы уровня модуля в UPPER_CASE

## Контрольный список качества

Прежде чем завершить реализацию MCP-сервера на Python, убедитесь в следующем:

### Стратегический дизайн
- [ ] Инструменты позволяют выполнять полные рабочие процессы, а не просто оборачивают конечные точки API
- [ ] Имена инструментов отражают естественное разделение задач
- [ ] Форматы ответов оптимизированы для эффективного использования контекста агента
- [ ] Где это уместно, используются понятные человеку идентификаторы
- [ ] Сообщения об ошибках направляют агентов к правильному использованию

### Качество реализации
- [ ] СФОКУСИРОВАННАЯ РЕАЛИЗАЦИЯ: реализованы самые важные и ценные инструменты
- [ ] У всех инструментов описательные имена и документация
- [ ] Типы возвращаемых значений согласованы для сходных операций
- [ ] Обработка ошибок реализована для всех внешних вызовов
- [ ] Имя сервера соответствует формату: `{service}_mcp`
- [ ] Все сетевые операции используют async/await
- [ ] Общая функциональность выделена в повторно используемые функции
- [ ] Сообщения об ошибках понятны, подсказывают действия и помогают разобраться
- [ ] Выходные данные корректно проверяются и форматируются

### Конфигурация инструментов
- [ ] Все инструменты реализуют 'name' и 'annotations' в декораторе
- [ ] Аннотации заданы правильно (readOnlyHint, destructiveHint, idempotentHint, openWorldHint)
- [ ] Все инструменты используют Pydantic BaseModel для проверки входных данных с определениями Field()
- [ ] Все поля Pydantic имеют явные типы и описания с ограничениями
- [ ] У всех инструментов подробные docstring с явными типами входных и выходных данных
- [ ] Docstring включают полную структуру схемы для возвращаемых значений dict/JSON
- [ ] Модели Pydantic выполняют проверку входных данных (ручная проверка не требуется)

### Расширенные возможности (где применимо)
- [ ] Внедрение Context используется для журналирования, отображения прогресса или запроса ввода
- [ ] Для соответствующих конечных точек данных зарегистрированы ресурсы
- [ ] Для постоянных соединений реализовано управление жизненным циклом
- [ ] Используются структурированные типы вывода (TypedDict, модели Pydantic)
- [ ] Настроен подходящий транспорт (stdio или streamable HTTP)

### Качество кода
- [ ] Файл содержит правильные импорты, включая импорты Pydantic
- [ ] Где применимо, корректно реализована пагинация
- [ ] Для потенциально больших наборов результатов предусмотрены параметры фильтрации
- [ ] Все асинхронные функции правильно определены с помощью `async def`
- [ ] Использование HTTP-клиента следует асинхронным подходам с подходящими контекстными менеджерами
- [ ] Аннотации типов используются во всём коде
- [ ] Константы определены на уровне модуля в UPPER_CASE

### Тестирование
- [ ] Сервер успешно запускается: `python your_server.py --help`
- [ ] Все импорты разрешаются правильно
- [ ] Примерные вызовы инструментов работают ожидаемым образом
- [ ] Ошибочные сценарии корректно обрабатываются