Управление ресурсами

Ресурсы в MCP — это не просто статические строки. Они могут быть динамическими, бинарными, кешируемыми и параметризованными. В этой теме разберём продвинутые техники работы с ресурсами в FastMCP.

Ресурс — это функция, которая возвращает данные. Всё остальное — как вы организуете эту функцию — зависит от вас.

#Текстовые ресурсы

Базовый случай — функция возвращает строку:

import json

@mcp.resource("config://app")
def get_app_config() -> str:
    """Конфигурация приложения в JSON."""
    config = {
        "version": "2.1.0",
        "debug": False,
        "features": ["search", "analytics"]
    }
    return json.dumps(config, indent=2)

FastMCP оборачивает строку в MCP-формат с MIME-типом text/plain. Клиент получает текст напрямую.

#Бинарные ресурсы

Для бинарных данных (изображения, PDF, архивы) функция возвращает bytes:

from pathlib import Path

@mcp.resource("files://{name}/download")
def download_file(name: str) -> bytes:
    """Возвращает содержимое файла по имени."""
    file_path = Path("/data/files") / name
    if not file_path.exists():
        raise FileNotFoundError(f"Файл {name} не найден")
    return file_path.read_bytes()

FastMCP кодирует байты в base64 и устанавливает MIME-тип application/octet-stream. Клиент декодирует base64 обратно в бинарные данные.

#Динамические ресурсы с логикой

Функция ресурса — это обычный Python-код. Можно обращаться к БД, внешним API, выполнять вычисления:

import sqlite3

@mcp.resource("stats://daily/{date}")
def get_daily_stats(date: str) -> str:
    """Возвращает статистику за указанную дату."""
    conn = sqlite3.connect("analytics.db")
    cursor = conn.execute(
        "SELECT COUNT(*), AVG(duration) FROM events WHERE date = ?",
        (date,)
    )
    row = cursor.fetchone()
    conn.close()
    
    if row[0] == 0:
        return json.dumps({"date": date, "events": 0, "avg_duration": 0})
    
    return json.dumps({
        "date": date,
        "events": row[0],
        "avg_duration": round(row[1], 2)
    }, indent=2)

Ресурс читается из базы данных при каждом запросе. Дата извлекается из URI и передаётся в функцию как параметр.

#Кэширование ресурсов

Если данные меняются редко, а читаются часто — кэширование экономит ресурсы сервера:

import time
from functools import lru_cache

# Простой кеш с TTL
_cache = {}
_CACHE_TTL = 300  # 5 минут

def get_cached_resource(key: str, fetch_fn):
    """Возвращает данные из кеша или обновляет кеш."""
    if key in _cache:
        data, timestamp = _cache[key]
        if time.time() - timestamp < _CACHE_TTL:
            return data
    
    # Кеш устарел или отсутствует — обновляем
    data = fetch_fn()
    _cache[key] = (data, time.time())
    return data

@mcp.resource("reports://weekly/{week_id}")
def get_weekly_report(week_id: str) -> str:
    """Возвращает еженедельный отчёт с кэшированием."""
    def fetch():
        # Дорогой запрос к БД или внешнему API
        return json.dumps({"week": week_id, "data": "..."})
    
    return get_cached_resource(f"weekly_{week_id}", fetch)

Кеш хранит данные 5 минут. Повторные запросы одного week_id в пределах TTL возвращают закэшированный результат без обращения к БД.

Когда кэшировать:

Данные меняются реже, чем частота запросов
Получение данных дорогое (сложный SQL-запрос, внешний API)
Клиенты читают один и тот же ресурс многократно

Когда не кэшировать:

Данные меняются в реальном времени (мониторинг, статус)
Каждый запрос уникален (персональные данные пользователя)
Данные всегда актуальны в памяти (in-memory структуры)

#Ресурсы с проверкой существования

Шаблон ресурса покрывает бесконечное множество URI. Не каждый URI соответствует реальным данным — функция должна обрабатывать отсутствие данных:

@mcp.resource("project://{project_id}/info")
def get_project_info(project_id: str) -> str:
    """Возвращает информацию о проекте или сообщение об отсутствии."""
    project = projects_db.get(project_id)
    if project is None:
        return json.dumps({"error": f"Проект {project_id} не найден"})
    return json.dumps(project, indent=2, ensure_ascii=False)

Вместо выброса исключения ресурс возвращает JSON с полем error. Это позволяет клиенту отличить «ресурс не найден» от «сервер сломан». Альтернатива — выбросить ValueError, и FastMCP вернёт JSON-RPC error.

#Ресурсы-агрегаторы

Ресурс может агрегировать данные из нескольких источников:

@mcp.resource("dashboard://overview")
def get_dashboard() -> str:
    """Агрегирует данные для панели мониторинга."""
    users_count = db.execute("SELECT COUNT(*) FROM users").fetchone()[0]
    orders_today = db.execute(
        "SELECT COUNT(*) FROM orders WHERE date = CURRENT_DATE"
    ).fetchone()[0]
    server_status = check_health()
    
    return json.dumps({
        "users_count": users_count,
        "orders_today": orders_today,
        "server_status": server_status,
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
    }, indent=2)

Один ресурс dashboard://overview объединяет несколько запросов. Клиент читает один URI и получает полную картину.

#Ресурсы и пагинация

Для больших наборов данных ресурс может возвращать метаданные пагинации:

@mcp.resource("logs://app/{date}")
def get_app_logs(date: str) -> str:
    """Возвращает логи приложения за дату с метаданными."""
    logs = fetch_logs_from_storage(date)
    total = len(logs)
    
    return json.dumps({
        "date": date,
        "total_entries": total,
        "entries": logs[:100],  # Ограничиваем вывод
        "truncated": total > 100
    }, indent=2, ensure_ascii=False)

Ресурс возвращает первые 100 записей и флаг truncated. Если клиенту нужны все записи, он может использовать инструмент с пагинацией — ресурс для быстрого просмотра, инструмент для полного доступа.