МСР: Трое в лодке, не считая контекста

Часть 2. Ресурсы (Resources): Даем модели «глаза»

В прошлой статье мы познакомились с основными понятиями протокола Model Context Protocol и написали простейшее приложение, которое позволило LLM читать файлы. Для этого мы использовали tools с оговоркой, что сделали это для упрощения, чтобы не лететь с места в карьер.

Мы уже говорили, что если tool можно сравнить с методом POST, то resource сравнивали с GET. Ресурсы (Resources) — это пассивные источники данных, которые MCP-сервер отдает клиенту для чтения. Такими источниками могут быть содержимое файла, лог консоли, строка в базе данных.

Идентификация: Каждый ресурс имеет уникальный URI (Uniform Resource Identifier).

• file://logs/error.txt

• postgres://db/users/schema

Отличие от Tools (Инструментов)

Кроме того, что ресурс не изменяет данные, есть и другие отличия. Небольшая табличка чтоб проще запомнить.

В библиотеке 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 = None

3. Хост (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-магии.

Пусть наш хост сам разбирается, какие файлы ему нужны

1. Пользователь: "Найди и объясни ошибку в логах".

2. LLM вызывает Tool: search_files("\*.log"). Необходимый фильтр для поиска выбирает самостоятельно исходя из запроса.

3. Ответом LLM должен стать список файлов, которые удовлетворяют критерию поиска.

4. LLM вызывает Resource: read_file(...).

5. LLM дает итоговый ответ на вопрос пользователя после анализа текста логов.

Ссылка на репозиторий simple_mcp