ErrorTool, пользовательские исключения, graceful degradation и логирование.
Хороший MCP-сервер не просто работает в идеальных условиях — он gracefully обрабатывает ошибки и сообщает клиенту, что пошло не так. В этой теме разберём ErrorTool, исключения, логирование и стратегию graceful degradation.
Правило: инструмент всегда должен завершаться корректно — либо с результатом, либо с понятной ошибкой. Молчаливый сбой — худший вариант.
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:
Для непредвиденных сбоев используйте обычные 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 — сервер продолжает работать частично при сбое зависимостей:
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)@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("Сервис погоды недоступен и кэш отсутствует")Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.