Часть 2. Ресурсы (Resources): Даем модели «глаза»
В прошлой статье мы познакомились с основными понятиями протокола Model Context Protocol и написали простейшее приложение, которое позволило LLM читать файлы. Для этого мы использовали tools с оговоркой, что сделали это для упрощения, чтобы не лететь с места в карьер.
Мы уже говорили, что если tool можно сравнить с методом POST, то resource сравнивали с GET. Ресурсы (Resources) — это пассивные источники данных, которые MCP-сервер отдает клиенту для чтения. Такими источниками могут быть содержимое файла, лог консоли, строка в базе данных.
Идентификация: Каждый ресурс имеет уникальный URI (Uniform Resource Identifier).
file://logs/error.txtpostgres://db/users/schema
Отличие от Tools (Инструментов)
Кроме того, что ресурс не изменяет данные, есть и другие отличия. Небольшая табличка чтоб проще запомнить.
| Tools (Инструменты) | Resources (Ресурсы) | |
|---|---|---|
| Роль | «Руки» (Выполнение действий) - аналог POST | «Глаза» (Чтение контекста) - аналог GET |
| Инициатор | LLM. Модель сама решает вызвать инструмент. | Хост/Пользователь. Приложение само «прикрепляет» ресурс к диалогу. |
| Внесение изменений | Есть (запись в БД, API запрос). Небезопасно | Нет (Read-only). Безопасно. |
В библиотеке FastMCP ресурсы определяются с помощью декораторов, очень похожие на роуты в веб-фреймворках (FastAPI/Flask). Имя ресурса мы можем конструировать по шаблону.
@mcp.resource("file:///{filename}") — сервер сам распарсит URI и передаст аргумент filename в функцию.
ВАЖНО FastMCP очень чувствителен к правильному написанию URI.
Почему то ни в одном источнике, изученном мной при написании статьи, акцент на этом моменте не делался. Для URI файлов нужно указать три слэша. Как я понял, FastMCP ждет структуру //{host}/{filename} Если передан один аргумент с двумя слэшами - он считает что передан хост. Файл найден не будет.
Безопасность: "Песочница" (Sandbox)
Вспомним про еще одно допущение, которое обсуждалось в первой части статьи.
В интернете много историй, про то как (без)умные агенты сносят все файлы на компьютере пользователей. Можно ли этого избежать?
Даже в самом простом варианте, мы давали доступ только к тестовой папке “песочнице”. Однако, используя относительные пути, всегда можно выйти из этой папки и попасть в основную директорию.
Решение: Всегда приводим пути к абсолютным (os.path.abspath) и проверяем, что итоговый путь начинается с разрешенной корневой директории.
Реализация в коде
1. Сервер (mcp_file_server.py)
Будем использовать код из первой части статьи. Только инструмент для чтения файла преобразуем в ресурс. Так мы можем быть твердо уверены, что содержимое файла при обращении к нему не изменится. К тому же мы улучшили защиту путей, преобразовывая относительный путь в абсолютный.
Еще я добавил пример логирования в MCP сервере. Можно делать это другими способами, через context, но пока так. Если что то напутаете с путями - хотя бы будет видно, где MCP сервер ищет файлы. Напомню, обычный способ через print или logger тут не сработает, поскольку клиент перехватывает поток stdio.
import sys
import os
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("demo-files-server")
# Определяем папку demo_files рядом со скриптом сервера
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
WORK_DIR = os.path.join(BASE_DIR, "demo_files")
os.makedirs(WORK_DIR, exist_ok=True)
def debug_log(msg):
"""Пишет отладочную информацию в поток ошибок (stderr)"""
sys.stderr.write(f"[SERVER DEBUG] {msg}\n")
sys.stderr.flush()
debug_log(f"Сервер запущен. Рабочая папка: {WORK_DIR}")
debug_log(f"Файлы в папке: {os.listdir(WORK_DIR)}")
# = ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ =
def get_safe_path(filename: str) -> str:
# Убираем префиксы схемы, если они вдруг попали в filename
clean_name = filename.replace("file://", "").lstrip("/")
target = os.path.join(WORK_DIR, clean_name)
abs_target = os.path.abspath(target)
debug_log(f"Запрос файла: '{filename}' -> Путь: '{abs_target}'")
if not abs_target.startswith(WORK_DIR):
debug_log(f"❌ Блокировка: {abs_target} вне {WORK_DIR}")
raise ValueError("Доступ запрещен (выход из песочницы)")
if not os.path.exists(abs_target):
debug_log(f"❌ Файл не найден: {abs_target}")
raise FileNotFoundError("Файл не найден")
return abs_target
# = РЕСУРСЫ (RESOURCES) =
# используем шаблон file:///{filename} - три слэша после file
@mcp.resource("file:///{filename}")
def read_file(filename: str) -> str:
"""Читает содержимое файла."""
debug_log(f"--- Вызов ресурса read_file с аргументом: {filename} ---")
try:
path = get_safe_path(filename)
with open(path, "r", encoding="utf-8") as f:
content = f.read()
debug_log(f" Файл прочитан успешно ({len(content)} байт)")
return content
except IOError as e:
debug_log(f" Ошибка внутри ресурса: {e}")
# Возвращаем текст ошибки, чтобы клиент не падал молча
return f"Error reading resource: {str(e)}"
if __name__ "__main__":
mcp.run()2 Клиент (mcp_client.py)
Добавляем метод read_resource(). Он принимает URI и возвращает контент. Инициализация и завершение сессии остаются из первой части. В данном примере считываем как текст, хотя можно читать разные типы
import asyncio
from contextlib import AsyncExitStack
from typing import Any
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import Tool, CallToolResult
class MCPClient:
def __init__(self, command: str, args: list[str], env: dict[str, str] | None = None):
# Параметры для запуска подпроцесса сервера
self.server_params = StdioServerParameters(
command=command,
args=args,
env=env
)
self.session: ClientSession | None = None
# Менеджер контекста для удержания соединения
self._exit_stack = AsyncExitStack()
async def connect(self) -> None:
"""Запуск сервера и инициализация сессии."""
try:
# Запуск процесса и создание транспорта
read, write = await self._exit_stack.enter_async_context(
stdio_client(self.server_params)
)
# Создание JSON-RPC сессии
self.session = await self._exit_stack.enter_async_context(
ClientSession(read, write)
)
# Рукопожатие (Handshake)
await self.session.initialize()
except Exception as e:
await self.cleanup()
raise e
# добавили чтение ресурса
async def read_resource(self, uri: str) -> str:
"""Запрашивает у сервера содержимое ресурса по URI"""
if not self.session:
raise RuntimeError("No session")
# список контента (может быть текст или бинарник)
try:
result = await self.session.read_resource(uri)
except IOError as e:
print(f"[Client]Ошибка чтения ресурса: {e}")
return result.contents[0].text
async def cleanup(self) -> None:
"""Закрытие ресурсов и завершение процесса сервера."""
# Корректный выход из всех контекстных менеджеров
await self._exit_stack.aclose()
self.session = None3. Хост (agent_host.py)
В хосте реализуем «Сборщик контекста». Он получает URI, скачивает данные и вставляет их в системный промпт.
import json
from typing import Any
from openai import AsyncOpenAI
from mcp_client import MCPClient
class AgentHost:
def __init__(self, mcp_client: MCPClient, openai_client: AsyncOpenAI, model: str = "gpt-4.1"):
self.client = mcp_client
self.openai = openai_client
self.model = model
# История диалога (Память агента)
self.messages: list[dict[str, Any]] = []
async def process_w_resources(self, user_query: str, resource_uri: str = None) -> str:
"""Чтение ресурса по ссылке"""
# ЗАГРУЗКА РЕСУРСА (Если передан)
if resource_uri:
print(f"[Host] Читаю ресурс: {resource_uri}...")
try:
content = await self.client.read_resource(resource_uri)
# Внедряем контент в System Prompt
system_msg = (
f"Используй содержимое этого файла для ответа:\n"
f"--- URI: {resource_uri} ---\n"
f"{content}\n"
f"--- END OF FILE ---"
)
self.messages.append({"role": "system", "content": system_msg})
except IOError as e:
return f"[Host]Ошибка загрузки ресурса: {e}"
# ЗАПРОС К LLM
self.messages.append({"role": "user", "content": user_query})
response = await self.openai.chat.completions.create(
model=self.model, messages=self.messages)
return response.choices[0].message.content 4. Main (main_res.py)
Для демонстрации чтения файла, попросим LLM посмотреть лог приложения и объяснить суть ошибки.
Можно взять фрагмент реального документа либо сгенерировать. Например вот такой:
# Создаем тестовый лог
os.makedirs("demo_files", exist_ok=True)
with open("demo_files/app.log", "w") as f:
f.write("ERROR 500: Database connection timeout")Далее реализуем простейший сценарий, с передачей имени файла хосту. Примерно так работает “скрепка” в приложениях с LLM, когда вы просите прочитать прилагаемый документ.
Если код скопирован без ошибок, пути прописаны верно - LLM должна выдать вердикт, что ошибка связана с Базой данных.
import asyncio
import os
import sys
from dotenv import load_dotenv
from openai import AsyncOpenAI
from agent_host import AgentHost
from mcp_client import MCPClient
load_dotenv()
async def main():
try:
openai_key = os.getenv("OPENAI_API_KEY")
openai_model = "gpt-4o"
server_script = "mcp_file_server.py"
mcp_client = MCPClient(
command=sys.executable, # Текущий интерпретатор
args=[server_script]
)
openai_client = AsyncOpenAI(api_key=openai_key)
await mcp_client.connect()
print("✅ Сервер подключен")
# Инициализация Агента
agent = AgentHost(
mcp_client=mcp_client,
openai_client=openai_client,
model= openai_model
)
# Явно указываем файл
question = "Почему упало приложение?"
target_file = "file:///app.log"
print(f"Вопрос: {question}")
print(f"Прикрепляем: {target_file}")
# Хост сам всё скачает и спросит LLM
answer = await agent.process_w_resources(question, resource_uri=target_file)
print(f"\nОтвет AI: {answer}")
except IOError as e:
print(f"Ошибка чтения ресурса {e}")
except Exception as e:
print(f"Ошибка {e}")
finally:
await mcp_client.cleanup()
if __name__ "__main__":
asyncio.run(main())Зачем всё это нужно? (Преимущества MCP)
У читателя может возникнуть резонный вопрос. Для чего все эти клиенты, сервера, хосты, если я могу прочитать любой файл обычным open() и передать в контекст?
В простых приложениях скорее всего проще именно так и сделать. Для многофункционального ИИ-комбайна всё-таки есть причины помучиться.
Подписки (Subscriptions)
Проблема: Допустим LLM нужно получать данные о тысячах пользователей, чтобы учесть их при ответе. Пусть нечасто, но эти данные могут поменяться. Обращаться каждый раз в базу - может быть ощутимо долго и затратно по ресурсам.
Решение MCP: Клиент может подписаться на ресурс. Сервер пришлет уведомление resource/updated при изменении файла.
MIME-типы
Нынешние модели часто мультимодальные. Они могут принимать текст, картинки, аудио.
MCP передает метаданные. Сервер сообщает: «Это картинка (image/png)» или «Это Python-код (text/x-python)».
В наших примерах мы не используем подписки и mime-типы, я сам толком не разобрался чтобы не перегружать статью.
Унификация (Абстракция)
Для Клиента (и Хоста) не важно, откуда идут данные.
file://report.txt(Файл)postgres://users/last(SQL запрос)- …
Всё это — просто URI.
Такой подход делает наш код более структурированным, читаемым. Кроме того, прелесть стандарта в том, что мы можем пользоваться чужими наработками. Делаем хост с возможностью чтения ресурсов - можем подключать MCP сервера для чтения интернет-ссылок, сканирования репозиториев и т.п.
Search Tool + Read Resource
Дабы закончить на мажорном аккорде, добавим немного MCP-магии.
Пусть наш хост сам разбирается, какие файлы ему нужны
Пользователь: "Найди и объясни ошибку в логах".
LLM вызывает Tool: search\_files("\*.log"). Необходимый фильтр для поиска выбирает самостоятельно исходя из запроса.
Ответом LLM должен стать список файлов, которые удовлетворяют критерию поиска.
LLM вызывает Resource: read_file(...).
LLM дает итоговый ответ на вопрос пользователя после анализа текста логов.
Комментарии