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

Хороший MCP-сервер не просто работает в идеальных условиях — он gracefully обрабатывает ошибки и сообщает клиенту, что пошло не так. В этой теме разберём ErrorTool, исключения, логирование и стратегию graceful degradation.

Правило: инструмент всегда должен завершаться корректно — либо с результатом, либо с понятной ошибкой. Молчаливый сбой — худший вариант.

#ErrorTool

ErrorTool — это способ инструмента сообщить клиенту об ожидаемой ошибке без выброса исключения:

from fastmcp import FastMCP, ErrorTool

mcp = FastMCP("UserServer")

@mcp.tool()
def get_user(user_id: int) -> str:
    """Возвращает данные пользователя по ID."""
    user = users_db.get(user_id)
    if user is None:
        return ErrorTool(f"Пользователь с ID {user_id} не найден")
    return str(user)

Когда user_id не существует, инструмент возвращает ErrorTool с сообщением. AI-клиент видит это как валидный результат (не ошибку протокола) и может объяснить пользователю: «Пользователь 999 не найден в базе».

Когда использовать ErrorTool:

Запись не найдена (user not found, file missing)
Неверный формат данных (email без @, дата в неправильном формате)
Отказано в доступе (у пользователя нет прав)
Бизнес-ограничение (лимит исчерпан, операция недоступна)

#Исключения (raise)

Для непредвиденных сбоев используйте обычные Python-исключения:

import logging

logger = logging.getLogger(__name__)

@mcp.tool()
def fetch_external_data(url: str) -> str:
    """Загружает данные из внешнего источника."""
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return response.text
    except requests.Timeout:
        logger.error("Таймаут при запросе к %s", url)
        return ErrorTool("Внешний сервис не отвечает. Попробуйте позже.")
    except requests.ConnectionError:
        logger.error("Ошибка подключения к %s", url)
        return ErrorTool("Не удалось подключиться к внешнему сервису.")
    except Exception as e:
        logger.exception("Непредвиденная ошибка при запросе %s", url)
        raise

Здесь мы различаем:

Ожидаемые ошибки (таймаут, нет соединения) → ErrorTool с понятным сообщением
Непредвиденные ошибки (баг в коде, падение библиотеки) → raise, FastMCP вернёт JSON-RPC error

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

#Логирование

Настройте логирование для диагностики проблем:

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[
        logging.FileHandler("mcp_server.log"),
        logging.StreamHandler()  # дублирует в stdout
    ]
)

logger = logging.getLogger(__name__)

@mcp.tool()
def process_order(order_id: int) -> str:
    """Обрабатывает заказ."""
    logger.info("Обработка заказа %d", order_id)
    order = orders_db.get(order_id)
    if order is None:
        logger.warning("Заказ %d не найден", order_id)
        return ErrorTool(f"Заказ {order_id} не найден")
    
    try:
        result = process(order)
        logger.info("Заказ %d успешно обработан", order_id)
        return result
    except Exception as e:
        logger.exception("Ошибка обработки заказа %d", order_id)
        raise

Лог-файл содержит полную картину: какие запросы приходили, какие были ошибки, сколько времени заняла обработка.

#Graceful degradation

Graceful degradation — сервер продолжает работать частично при сбое зависимостей:

import redis

# Попытка подключиться к Redis для кэширования
try:
    cache = redis.Redis(host="localhost", port=6379, decode_responses=True)
    cache.ping()
    cache_available = True
except (redis.ConnectionError, redis.TimeoutError):
    cache = None
    cache_available = False
    logger.warning("Redis недоступен — кэширование отключено")

@mcp.resource("stats://hourly/{hour}")
def get_hourly_stats(hour: str) -> str:
    """Возвращает статистику за час с fallback при сбое кэша."""
    cache_key = f"stats:{hour}"
    
    if cache_available:
        try:
            cached = cache.get(cache_key)
            if cached:
                return cached
        except (redis.ConnectionError, redis.TimeoutError):
            logger.warning("Кэш недоступен для ключа %s", cache_key)
    
    # Вычисляем данные из БД (медленнее, но работает)
    stats = compute_stats_from_db(hour)
    result = json.dumps(stats, indent=2)
    
    if cache_available:
        try:
            cache.setex(cache_key, 3600, result)  # кеш на 1 час
        except (redis.ConnectionError, redis.TimeoutError):
            pass  # Не критично — кэш восстановится позже
    
    return result

При недоступности Redis ресурс продолжает работать — просто без кэширования. Сервер деградирует gracefully: медленнее, но функционален.

#Типичные паттерны обработки ошибок

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

import re

@mcp.tool()
def send_email(to: str, subject: str, body: str) -> str:
    """Отправляет email."""
    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    if not re.match(email_pattern, to):
        return ErrorTool(f"Некорректный email-адрес: {to}")
    if len(subject) > 200:
        return ErrorTool("Тема письма слишком длинная (максимум 200 символов)")
    # Отправка...
    return f"Письмо отправлено на {to}"

#Ограничение ресурсов

MAX_RESULTS = 100

@mcp.tool()
def search_logs(query: str, days_back: int = 7) -> str:
    """Ищет логи по запросу."""
    if days_back > 30:
        return ErrorTool("Поиск возможен максимум за 30 дней")
    
    results = query_logs(query, days_back)
    if len(results) > MAX_RESULTS:
        return ErrorTool(
            f"Найдено {len(results)} записей. Уточните запрос "
            f"(максимум {MAX_RESULTS} результатов)."
        )
    return json.dumps(results, indent=2, ensure_ascii=False)

#Fallback для внешних API

@mcp.tool()
def get_weather(city: str) -> str:
    """Возвращает текущую погоду."""
    try:
        data = fetch_weather_api(city)
        return format_weather(data)
    except APIError:
        # Fallback: возвращаем данные из локального кэша
        cached = get_cached_weather(city)
        if cached:
            return f"{cached} (данные из кэша, API недоступен)"
        return ErrorTool("Сервис погоды недоступен и кэш отсутствует")