Доступ к данным: файлы, базы данных, внешние API. Чтение и запись ресурсов.
Resources — это данные MCP. В этой теме научимся предоставлять LLM доступ к файлам, базам данных, внешним API через стандартизированный интерфейс.
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 → Конфигурация приложения
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()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 атак.
Шаблоны позволяют создавать динамические ресурсы с параметрами.
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}")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-тип | Расширение | Описание |
|---|---|---|
text/plain | .txt | Обычный текст |
text/html | .html | HTML документ |
text/markdown | .md | Markdown |
application/json | .json | JSON данные |
application/xml | .xml | XML данные |
application/pdf | PDF документ | |
image/png | .png | PNG изображение |
image/jpeg | .jpg | JPEG изображение |
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.
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()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()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()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)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()# ❌ ПЛОХО: Нет валидации
@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),))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()@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}")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 — создание инструментов для вызова функций, валидация параметров, обработка результатов.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.