Resources API

Resources — это данные MCP. В этой теме научимся предоставлять LLM доступ к файлам, базам данных, внешним API через стандартизированный интерфейс.

#1. Что такое Resources?

Resources (Ресурсы) — единицы данных, доступные для чтения через MCP. LLM может запросить содержимое ресурса по URI.

#Характеристики ресурсов

Характеристика	Описание
URI	Уникальный адрес ресурса (file://, db://, https://)
MIME-тип	Формат данных (text/plain, application/json)
Содержимое	Текст или бинарные данные
Метаданные	Название, описание, версия

#Примеры ресурсов

file:///etc/hosts           → Файл в файловой системе
postgres://users/123        → Запись в базе данных
https://api.example.com/v1  → Внешний API
config://app/settings       → Конфигурация приложения

#2. Простые ресурсы

#Статический ресурс

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Resources Demo")


@mcp.resource("config://app/settings")
def get_app_settings() -> str:
    """Конфигурация приложения."""
    return """
APP_NAME=MyApp
VERSION=1.0.0
DEBUG=true
LOG_LEVEL=info
"""


@mcp.resource("data://greetings")
def get_greetings() -> str:
    """Список приветствий."""
    return "Hello, Hi, Welcome, Good day"

#Ресурс с динамическим содержимым

from datetime import datetime


@mcp.resource("system://clock")
def get_system_time() -> str:
    """Текущее системное время."""
    return datetime.now().isoformat()


@mcp.resource("system://hostname")
def get_hostname() -> str:
    """Имя хоста системы."""
    import socket
    return socket.gethostname()

#3. Ресурсы из файловой системы

#Чтение файлов

from pathlib import Path


@mcp.resource("file:///tmp/app.log")
def read_log_file() -> str:
    """Чтение лог-файла."""
    log_path = Path("/tmp/app.log")
    
    if not log_path.exists():
        raise FileNotFoundError(f"Log file not found: {log_path}")
    
    return log_path.read_text()


@mcp.resource("file:///etc/hosts")
def read_hosts_file() -> str:
    """Чтение файла hosts."""
    return Path("/etc/hosts").read_text()

#Безопасность: ограничение путей

from pathlib import Path


# Определяем безопасную базовую директорию
BASE_DIR = Path("/var/mcp/data").resolve()


def is_safe_path(path: Path) -> bool:
    """Проверяет, что путь находится внутри BASE_DIR."""
    try:
        path.resolve().relative_to(BASE_DIR)
        return True
    except ValueError:
        return False


@mcp.resource("file:///{filename}")
def read_data_file(filename: str) -> str:
    """Чтение файла из безопасной директории."""
    file_path = BASE_DIR / filename
    
    if not is_safe_path(file_path):
        raise PermissionError(f"Access denied: {filename}")
    
    if not file_path.exists():
        raise FileNotFoundError(f"File not found: {filename}")
    
    return file_path.read_text()

Важно: Всегда валидируйте пути к файлам для предотвращения path traversal атак.

#4. Resource Templates (Шаблоны ресурсов)

Шаблоны позволяют создавать динамические ресурсы с параметрами.

#Простой шаблон

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Templates Demo")


@mcp.resource("user://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """
    Профиль пользователя.
    
    Args:
        user_id: ID пользователя
    """
    # Имитация базы данных
    users = {
        "1": {"name": "Alice", "email": "alice@example.com"},
        "2": {"name": "Bob", "email": "bob@example.com"}
    }
    
    if user_id not in users:
        raise ValueError(f"User {user_id} not found")
    
    user = users[user_id]
    return f"Name: {user['name']}\nEmail: {user['email']}"

#Шаблон с несколькими параметрами

@mcp.resource("db://{table_name}/{record_id}")
def get_database_record(table_name: str, record_id: str) -> str:
    """
    Запись из базы данных.
    
    Args:
        table_name: Имя таблицы
        record_id: ID записи
    """
    # Валидация имени таблицы (защита от SQL-инъекций)
    allowed_tables = ["users", "posts", "comments"]
    if table_name not in allowed_tables:
        raise ValueError(f"Table {table_name} not allowed")
    
    # Имитация запроса к БД
    return f"Record {record_id} from {table_name}"

#Шаблон с форматом вывода

@mcp.resource("api://users/{user_id}/format:{format}")
def get_user_formatted(user_id: str, format: str = "json") -> str:
    """
    Данные пользователя в разном формате.
    
    Args:
        user_id: ID пользователя
        format: Формат вывода (json, xml, text)
    """
    user = {"id": user_id, "name": "Alice", "email": "alice@example.com"}
    
    if format == "json":
        import json
        return json.dumps(user)
    elif format == "xml":
        return f"<user><id>{user_id}</id><name>{user['name']}</name></user>"
    elif format == "text":
        return f"ID: {user_id}\nName: {user['name']}\nEmail: {user['email']}"
    else:
        raise ValueError(f"Unsupported format: {format}")

#5. MIME-типы ресурсов

#Явное указание MIME-типа

from mcp.server.fastmcp import FastMCP, Resource


@mcp.resource("data://config.json", mime_type="application/json")
def get_config_json() -> str:
    """Конфигурация в JSON формате."""
    import json
    return json.dumps({
        "app_name": "MyApp",
        "version": "1.0.0",
        "features": ["auth", "logging", "metrics"]
    })


@mcp.resource("data://report.html", mime_type="text/html")
def get_html_report() -> str:
    """HTML отчёт."""
    return """
<html>
<head><title>Report</title></head>
<body>
    <h1>Sales Report</h1>
    <p>Total: $10,000</p>
</body>
</html>
"""


@mcp.resource("data://image.png", mime_type="image/png")
def get_image() -> bytes:
    """PNG изображение (бинарные данные)."""
    from pathlib import Path
    return Path("/path/to/image.png").read_bytes()

#Распространённые MIME-типы

MIME-тип	Расширение	Описание
`text/plain`	.txt	Обычный текст
`text/html`	.html	HTML документ
`text/markdown`	.md	Markdown
`application/json`	.json	JSON данные
`application/xml`	.xml	XML данные
`application/pdf`	.pdf	PDF документ
`image/png`	.png	PNG изображение
`image/jpeg`	.jpg	JPEG изображение

#6. Подписка на изменения ресурсов

#Реализация подписки

from mcp.server.fastmcp import FastMCP
import asyncio

mcp = FastMCP("Subscribe Demo")


# Хранилище подписок
subscriptions: dict[str, set] = {}


@mcp.resource("logs://app")
def get_app_logs() -> str:
    """Логи приложения."""
    return "[2026-03-17 10:00:00] INFO: App started\n[2026-03-17 10:01:00] INFO: Processing..."


@mcp.tool()
async def subscribe_to_logs() -> str:
    """Подписаться на изменения логов."""
    # В реальной реализации здесь была бы логика подписки
    return "Subscribed to logs://app"


async def notify_subscribers():
    """Уведомить подписчиков об изменении ресурса."""
    # Отправка уведомления через MCP
    await mcp.request_context.session.send_resource_updated("logs://app")


# Фоновая задача для проверки изменений
async def check_log_changes():
    last_log = get_app_logs()
    while True:
        await asyncio.sleep(5)
        current_log = get_app_logs()
        if current_log != last_log:
            await notify_subscribers()
            last_log = current_log

#Ограничения подписки

Важно: Подписка на ресурсы требует поддержки со стороны транспорта. stdio транспорт не поддерживает push-уведомления — для этого нужен SSE или WebSocket.

#7. Списки ресурсов

#Явное определение списка

from mcp.server.fastmcp import FastMCP, Resource

mcp = FastMCP("Resources List Demo")


# Явное определение ресурсов с метаданными
RESOURCES = [
    Resource(
        uri="docs://readme",
        name="README",
        description="Основная документация проекта",
        mime_type="text/markdown"
    ),
    Resource(
        uri="docs://changelog",
        name="CHANGELOG",
        description="История изменений",
        mime_type="text/markdown"
    ),
    Resource(
        uri="docs://api",
        name="API Reference",
        description="Справочник API",
        mime_type="text/html"
    )
]


@mcp.resource("docs://readme")
def get_readme() -> str:
    """Содержимое README."""
    return "# My Project\n\nDocumentation..."


@mcp.resource("docs://changelog")
def get_changelog() -> str:
    """Содержимое CHANGELOG."""
    return "## v1.0.0\n\n- Initial release"


@mcp.resource("docs://api")
def get_api_ref() -> str:
    """Содержимое API Reference."""
    return "<h1>API Reference</h1>..."

#Динамический список ресурсов

from mcp.server.fastmcp import FastMCP, Resource
from pathlib import Path

mcp = FastMCP("Dynamic Resources")

DATA_DIR = Path("/var/mcp/data")


@mcp.list_resources()
def list_data_files() -> list[Resource]:
    """Динамический список файлов в DATA_DIR."""
    resources = []
    
    if DATA_DIR.exists():
        for file_path in DATA_DIR.iterdir():
            if file_path.is_file():
                resources.append(Resource(
                    uri=f"file://{file_path}",
                    name=file_path.name,
                    description=f"File: {file_path.name}",
                    mime_type="text/plain"
                ))
    
    return resources


@mcp.resource("file://{path}")
def read_data_file(path: str) -> str:
    """Чтение файла из DATA_DIR."""
    file_path = DATA_DIR / path
    
    if not file_path.exists():
        raise FileNotFoundError(f"File not found: {path}")
    
    return file_path.read_text()

#8. Работа с базами данных

#PostgreSQL ресурс

from mcp.server.fastmcp import FastMCP
import asyncpg
import os

mcp = FastMCP("PostgreSQL Demo")


async def get_db_connection():
    """Получение подключения к БД."""
    return await asyncpg.connect(
        host=os.getenv("DB_HOST", "localhost"),
        port=int(os.getenv("DB_PORT", 5432)),
        user=os.getenv("DB_USER", "postgres"),
        password=os.getenv("DB_PASSWORD"),
        database=os.getenv("DB_NAME", "mydb")
    )


@mcp.resource("postgres://users")
async def list_users() -> str:
    """Список всех пользователей."""
    conn = await get_db_connection()
    try:
        rows = await conn.fetch("SELECT id, name, email FROM users ORDER BY id")
        return "\n".join([f"{r['id']}: {r['name']} ({r['email']})" for r in rows])
    finally:
        await conn.close()


@mcp.resource("postgres://users/{user_id}")
async def get_user(user_id: str) -> str:
    """Данные конкретного пользователя."""
    conn = await get_db_connection()
    try:
        # Параметризированный запрос для защиты от SQL-инъекций
        row = await conn.fetchrow(
            "SELECT id, name, email, created_at FROM users WHERE id = $1",
            int(user_id)
        )
        if row is None:
            raise ValueError(f"User {user_id} not found")
        return f"ID: {row['id']}\nName: {row['name']}\nEmail: {row['email']}\nCreated: {row['created_at']}"
    finally:
        await conn.close()

#SQLite ресурс (простой вариант)

from mcp.server.fastmcp import FastMCP
import sqlite3
from pathlib import Path

mcp = FastMCP("SQLite Demo")

DB_PATH = Path("/tmp/mydb.sqlite")


def get_connection():
    """Получение подключения к SQLite."""
    return sqlite3.connect(DB_PATH)


@mcp.resource("sqlite://tables")
def list_tables() -> str:
    """Список таблиц в базе данных."""
    conn = get_connection()
    try:
        cursor = conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'"
        )
        tables = [row[0] for row in cursor.fetchall()]
        return "\n".join(tables)
    finally:
        conn.close()


@mcp.resource("sqlite://{table_name}")
def query_table(table_name: str) -> str:
    """Просмотр содержимого таблицы."""
    # Валидация имени таблицы
    allowed_tables = ["users", "posts", "comments"]
    if table_name not in allowed_tables:
        raise ValueError(f"Table {table_name} not allowed")
    
    conn = get_connection()
    try:
        cursor = conn.execute(f"SELECT * FROM {table_name} LIMIT 100")
        columns = [desc[0] for desc in cursor.description]
        rows = cursor.fetchall()
        
        result = " | ".join(columns) + "\n"
        result += "-" * len(result) + "\n"
        for row in rows:
            result += " | ".join(str(v) for v in row) + "\n"
        return result
    finally:
        conn.close()

#9. Внешние API как ресурсы

#HTTP API ресурс

from mcp.server.fastmcp import FastMCP
import aiohttp
import asyncio

mcp = FastMCP("HTTP API Demo")


@mcp.resource("https://api.github.com/repos/{owner}/{repo}")
async def get_github_repo(owner: str, repo: str) -> str:
    """
    Информация о репозитории GitHub.
    
    Args:
        owner: Владелец репозитория
        repo: Название репозитория
    """
    url = f"https://api.github.com/repos/{owner}/{repo}"
    
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            if response.status != 200:
                raise ValueError(f"GitHub API error: {response.status}")
            
            data = await response.json()
            
            return f"""
Repository: {data['full_name']}
Description: {data['description']}
Stars: {data['stargazers_count']}
Forks: {data['forks_count']}
Language: {data['language']}
URL: {data['html_url']}
"""


@mcp.resource("https://api.coindesk.com/v1/bpi/currentprice.json")
async def get_bitcoin_price() -> str:
    """Текущая цена Bitcoin."""
    url = "https://api.coindesk.com/v1/bpi/currentprice.json"
    
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.json()
            
            usd_price = data['bpi']['USD']['rate']
            return f"Bitcoin Price: ${usd_price}"

#Ресурс с кэшированием

from functools import lru_cache
import time


@lru_cache(maxsize=100)
def cached_api_response(endpoint: str, ttl: int = 300) -> str:
    """
    Кэшированный ответ API.
    
    Args:
        endpoint: Endpoint API
        ttl: Время жизни кэша в секундах
    """
    # В реальной реализации здесь был бы HTTP запрос
    # и проверка времени жизни кэша
    return f"Cached response for {endpoint}"


@mcp.resource("api://cached/{endpoint}")
def get_cached_data(endpoint: str) -> str:
    """Данные с кэшированием."""
    return cached_api_response(endpoint)

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

#Работа с изображениями

from mcp.server.fastmcp import FastMCP
from pathlib import Path

mcp = FastMCP("Binary Resources")


@mcp.resource("image://logo.png")
def get_logo() -> bytes:
    """Логотип компании (PNG)."""
    return Path("/path/to/logo.png").read_bytes()


@mcp.resource("image://avatar/{user_id}.jpg")
def get_user_avatar(user_id: str) -> bytes:
    """Аватар пользователя."""
    avatar_path = Path(f"/var/avatars/{user_id}.jpg")
    
    if not avatar_path.exists():
        # Возвращаем дефолтный аватар
        return Path("/var/avatars/default.jpg").read_bytes()
    
    return avatar_path.read_bytes()


@mcp.resource("file:///{path}")
def read_binary_file(path: str) -> bytes:
    """Чтение бинарного файла."""
    file_path = Path(f"/var/files/{path}")
    
    if not file_path.exists():
        raise FileNotFoundError(f"File not found: {path}")
    
    return file_path.read_bytes()

#11. Best Practices для Resources

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

# ❌ ПЛОХО: Нет валидации
@mcp.resource("db://{table}/{id}")
def get_record(table: str, id: str) -> str:
    return db.query(f"SELECT * FROM {table} WHERE id = {id}")

# ✅ ХОРОШО: Валидация и параметризация
@mcp.resource("db://{table}/{id}")
def get_record(table: str, id: str) -> str:
    allowed_tables = ["users", "posts", "comments"]
    if table not in allowed_tables:
        raise ValueError(f"Invalid table: {table}")
    
    if not id.isdigit():
        raise ValueError(f"Invalid id: {id}")
    
    return db.query("SELECT * FROM {} WHERE id = %s".format(table), (int(id),))

#2. Ограничение размера данных

MAX_RESOURCE_SIZE = 1024 * 1024  # 1 MB


@mcp.resource("file://{path}")
def read_file(path: str) -> str:
    file_path = Path(f"/data/{path}")
    
    # Проверка размера перед чтением
    if file_path.stat().st_size > MAX_RESOURCE_SIZE:
        raise ValueError(f"File too large: {file_path.stat().st_size} bytes")
    
    return file_path.read_text()

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

@mcp.resource("api://external/{id}")
def get_external_data(id: str) -> str:
    try:
        response = requests.get(f"https://api.example.com/data/{id}", timeout=5)
        response.raise_for_status()
        return response.text
    except requests.Timeout:
        raise ValueError("External API timeout")
    except requests.ConnectionError:
        raise ValueError("External API unavailable")
    except requests.HTTPError as e:
        raise ValueError(f"API error: {e.response.status_code}")

#4. Логирование доступа

import logging

logger = logging.getLogger(__name__)


@mcp.resource("sensitive://{path}")
def read_sensitive_file(path: str) -> str:
    logger.info(f"Access to sensitive resource: {path}")
    
    file_path = Path(f"/secure/{path}")
    
    if not file_path.exists():
        logger.warning(f"Sensitive file not found: {path}")
        raise FileNotFoundError(f"File not found: {path}")
    
    return file_path.read_text()

#Ключевые выводы

Концепт	Описание
URI ресурса	Уникальный адрес в формате scheme://path
MIME-тип	Описывает формат данных ресурса
Resource Template	Шаблон для динамических ресурсов с параметрами
list_resources()	Декоратор для динамического списка ресурсов
Подписка	Уведомления об изменении ресурсов (требует SSE/WebSocket)
Безопасность	Валидация путей, ограничение размера, логирование

Следующая тема: Tools API — создание инструментов для вызова функций, валидация параметров, обработка результатов.

Характеристика

Описание

URI

Уникальный адрес ресурса (file://, db://, https://)

MIME-тип

Формат данных (text/plain, application/json)

Содержимое

Текст или бинарные данные

Метаданные

Название, описание, версия

file:///etc/hosts → Файл в файловой системе postgres://users/123 → Запись в базе данных https://api.example.com/v1 → Внешний API config://app/settings → Конфигурация приложения

MIME-тип

Расширение

Описание

text/plain

.txt

Обычный текст

text/html

.html

HTML документ

text/markdown

.md

Markdown

application/json

.json

JSON данные

application/xml

.xml

XML данные

application/pdf

.pdf

PDF документ

image/png

.png

PNG изображение

image/jpeg

.jpg

JPEG изображение

Концепт

Описание

URI ресурса

Уникальный адрес в формате scheme://path

MIME-тип

Описывает формат данных ресурса

Resource Template

Шаблон для динамических ресурсов с параметрами

list_resources()

Декоратор для динамического списка ресурсов

Подписка

Уведомления об изменении ресурсов (требует SSE/WebSocket)

Безопасность

Валидация путей, ограничение размера, логирование