Динамические ресурсы, текстовые и бинарные данные, аннотации, кэширование.
Ресурсы в 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 возвращают закэшированный результат без обращения к БД.
Когда кэшировать:
Когда не кэшировать:
Шаблон ресурса покрывает бесконечное множество 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. Если клиенту нужны все записи, он может использовать инструмент с пагинацией — ресурс для быстрого просмотра, инструмент для полного доступа.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.