Как подружить MCP-сервер, клиент и LLM в вашем приложении - практическое руководство.

Всем привет! Меня зовут Владимир, последние несколько лет я занимаюсь разработкой приложений с использованием моделей компьютерного зрения (CV), обработки естественного языка (NLP) и больших языковых моделей (LLM).

Буду делиться своими наработками в этих областях. Начнем с серии материалов по МСР

ВВЕДЕНИЕ

Если вы - разработчик, и используете в своих приложениях LLM (или большие языковые модели - а куда сейчас без них), вы наверняка слышали про Model Context Protocol . Представленный в конце 2024 года компанией Anthropic, этот стандарт описывает взаимодействие языковых моделей со внешним миром.

Вызывать внешние функции модели “умели” и раньше. Но Anthropic наверно были первыми, кто оформил это в виде стандарта (есть и другие, например Agent2Agent от Google).

Когда я собирал материалы на данную тему - столкнулся с очень противоречивой информацией. От восторженных эпитетов “USB для ИИ-агентов” (такое определение дали сами разработчики протокола) или “швейцарский нож для LLM”, до серьезной критики проблем с безопасностью, транспортом и прочих “черных дыр”. Лучше всего эту ситуацию характеризует фраза из одной статьи: “У нас было 10 способов для связи моделей с инструментами. Решили унифицировать этот процесс с помощью MCP. Теперь у нас есть 11 разных способов”.
Дальше больше. Когда я пытался найти примеры реализации, столкнулся с определенной проблемой. Примеров реализации серверов очень много. Но ни одна модель “из коробки” с ними не может работать, в том числе модели Claude от разработчика протокола Anthropic. Нужен клиент, который позволит модели общаться с сервером. Если вы используете Claude desktop или Cursor - в них есть встроенные клиенты. А как использовать сервера в собственных проектах? Давайте разбираться.

Для нашего приложения будем использовать определения, описанные в стандарте

Чтобы понять архитектуру, представим управление бытовой техникой:

MCP Server («Устройство», например, Кондиционер). Обладает функциями («охладить»), но сам решений не принимает.
MCP Host («Пульт» или « шкаф автоматики» - наше приложение). Внутри него находится MCP Client - модуль, который умеет отправлять сигналы устройствам.
LLM («Мозг или процессор»). Принимает решение что нужно запустить в зависимости от условий. Стало жарко - включили кондиционер..
Main / Оркестратор («Человек»). Включает пульт и ставит задачу.

Будем писать код на python. Поскольку основная задача статьи - понять механизм реализации протокола MCP, многие вещи будут показаны упрощенно, дабы лес за деревьями проглядывался.

Глава 1. Создаем «Руки»: Разработка MCP Server

В этой главе мы создадим сервер, который предоставит доступ к локальной файловой системе.

Настройка окружения с uv

Будем использовать современный менеджер пакетов uv. Кто не знает как с ним работать - гляньте спойлер. Либо можно использовать другой менеджер, например привычный всем pip.

Скрытый текст

Инициализация проекта

# Устанавливаем uv если еще не установлен
# Mac/Linux: curl -LsSf https://astral.sh/uv/install.sh | sh 
# Windows: powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex" 

# Создаем проект
uv init mcp_article
cd mcp_article

# Установка необходимых библиотек
uv add mcp openai python-dotenv

uv сам создаст минимальную структуру проекта и гит репозиторий. Однако не забываем актуализировать созданные файлы. Например добавить в .gitignore файл с секретами.

Далее переходим к коду.

Для создания сервера используем библиотеку FastMCP .
Структура кода напоминает FastAPI.
Инициализируем сервер, далее создаем сущности при помощи декораторов.
Тут даже можно провести некую аналогию с http методами.
Протокол MCP описывает такие сущности, как

инструменты (tools);
ресурсы (resources);
подсказки (prompts);

Инструменты сравнивают с POST методом - вызов функции создает некий объект. Ресурс используется только для чтения, так что его можно сравнить с GET.

Подсказки вероятно нужны для каких то сложных сценариев, пока они нам не нужны.

В парадигме MCP, предоставление сервером списка файлов подходит под определение ресурса. Однако в данном примере мы намеренно делаем упрощение, потому что использование ресурсов потребовало бы дополнительных методов в клиентском модуле. Ну и есть там свои нюансы, с безопасностью и с “замусориванием контекста”. Так что list_files() мы тоже оформим как инструмент. В целях безопасности, даем доступ только к одной папке внутри нашего проекта.

В каждой функции обязательно указываем типы и делаем докстринг. На основе этого формируется схема, которую будет использовать LLM для вызова функции. Без этого LLM не поймет, в каком случае необходимо эту функцию применить и какие аргументы передать.

Код модуля (mcp_file_server.py)

from mcp.server.fastmcp import FastMCP
import os

mcp = FastMCP("demo-files-server")

# Рабочая директория (Sandbox)
WORK_DIR = os.path.join(os.getcwd(), "demo_files")
os.makedirs(WORK_DIR, exist_ok=True)

@mcp.tool()
def list_files() -> str:    
  """Возвращает список файлов в рабочей директории."""    
  try:        
    files = os.listdir(WORK_DIR)        
    if not files:            
      return "Папка пуста."        
    return "\n".join(files)    
    
  except Exception as e:        
    return f"Ошибка чтения папки: {e}"
  
@mcp.tool()
def read_file(filename: str) -> str:    
  """Читает содержимое файла. Принимает имя файла."""    
  safe_path = os.path.join(WORK_DIR, filename)        
  if not os.path.exists(safe_path):        
    return f"Файл '{filename}' не найден." 
  
  try:        
    with open(safe_path, 'r', encoding='utf-8') as f:    
      return f.read()    
      
  except Exception as e:        
      return f"Ошибка чтения файла: {e}"
    
if __name__ == "__main__":    
  mcp.run()

По умолчанию в качестве транспорта используется стандартный ввод/вывод (stdio). Его мы и будем использовать. HTTP транспорт в рамках данной стать�� рассматривать не будем. Там вопросов больше чем ответов.

Важно: поскольку stdio используется для связи клиента и сервера, использование его для других целей (например вывод в консоль print()) может поломать взаимодействие. Для вывода можно использовать logger. Пока не будем на этом останавливаться.

Глава 2. Прокладываем «Кабель»: Разработка MCP Client

В предыдущей главе мы написали код MCP сервера. И что же, уже можно взаимодействовать с ним с помощью LLM? Нет, пока еще рано.

Но как–то можно управлять им? На самом деле, существуют готовые инструменты для тестирования MCP сервера. О них коротко скажу в конце. А сейчас напишем клиента.

Задача этого модуля - запустить процесс сервера, установить соединение и поддерживать сессию.

Код модуля (mcp_client.py)

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 list_tools(self) -> list[Tool]:
        """Получить список инструментов от сервера."""
        if not self.session:
            raise RuntimeError("Сессия не активна.")
        result = await self.session.list_tools()
        return result.tools

    async def call_tool(self, name: str, arguments: dict) -> Any:
        """Вызвать инструмент по имени."""
        if not self.session:
            raise RuntimeError("Сессия не активна.")
        # Отправка запроса на исполнение
        result: CallToolResult = await self.session.call_tool(name, arguments)
        if result.isError:
            return f"Ошибка инструмента: {result.content}"
        return result.content

    async def cleanup(self) -> None:
        """Закрытие ресурсов и завершение процесса сервера."""
        # Корректный выход из всех контекстных менеджеров
        await self._exit_stack.aclose()
        self.session = None

Специфика MCP требует постоянного соединения. Мы используем AsyncExitStack, чтобы вручную управлять временем жизни сессии и добавлять необходимые ресурсы на каждом шаге пайплайна.

Глава 3. Создаем «Мозг»: Разработка Host

И вот мы уже вплотную подошли к тому, чтобы связать LLM и MCP сервер. Создадим класс AgentHost. Он и будет взаимодействовать с LLM. Для связи с сервером используем ранее созданный MCPClient.

Я буду использовать формат OpenAI для LLM, поскольку этот формат в том или ином виде поддерживают большинство провайдеров (например gemini, deepseek или yandex).

Формат описания tools у OpenAI несколько отличается от того что использует Anthropic. Так что придется сделать адаптер.

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

В реальном приложении лучше использовать другой тип связи.

Код модуля (agent_host.py)

import json
from typing import Any
from openai import AsyncOpenAI
from mcp_client import MCPClient

class AgentHost:
    def __init__(self, mcp_client: MCPClient, openai_api_key: str, model: str = "gpt-4.1"):
        self.client = mcp_client
        self.openai = AsyncOpenAI(api_key=openai_api_key)
        self.model = model
        # История диалога (Память агента)
        self.messages: list[dict[str, Any]] = []

    async def process_query(self, user_query: str) -> str:
        """Главный цикл ReAct (Reasoning + Acting)."""
        self.messages.append({"role": "user", "content": user_query})
        
        # Получение и адаптация инструментов
        mcp_tools = await self.client.list_tools()
        openai_tools = self._convert_tools_to_openai_format(mcp_tools)

        while True:
            # Шаг 1: Мысль (Запрос к LLM)
            response_message = await self._chat_completion(openai_tools)
            
            # Сохранение ответа ассистента в историю
            self.messages.append(response_message)
            
            tool_calls = response_message.get("tool_calls")
            if tool_calls:
                # Шаг 2: Действие (Acting)
                print(f" Агент хочет вызвать инструменты: {len(tool_calls)}")
                tool_results = await self._handle_tool_calls(tool_calls)
                
                # Шаг 3: Наблюдение (Observation)
                self.messages.extend(tool_results)
            else:
                # Финальный ответ
                return response_message.get("content") or ""

    async def _handle_tool_calls(self, tool_calls: list[dict[str, Any]]) -> list[dict[str, Any]]:
        """Исполнитель действий. Делегирует выполнение MCP-клиенту."""
        results_messages = []
        for tool_call in tool_calls:
            call_id = tool_call.get("id")
            function_data = tool_call.get("function", {})
            name = function_data.get("name")
            args_str = function_data.get("arguments", "{}")
            
            try:
                args = json.loads(args_str)
            except json.JSONDecodeError:
                args = {}
            
            print(f"   Вызов MCP: {name}({args})")
            
            # Вызов "черного ящика" через MCP Client
            try:
                mcp_result = await self.client.call_tool(name, args)
                content = str(mcp_result)
            except Exception as e:
                content = f"Ошибка выполнения: {str(e)}"

            results_messages.append({
                "role": "tool",
                "tool_call_id": call_id,
                "content": content
            })
        return results_messages

    async def _chat_completion(self, tools: list[dict]) -> dict[str, Any]:
        """Обертка для API OpenAI."""
        params = {"model": self.model, "messages": self.messages}
        if tools:
            params["tools"] = tools
        response = await self.openai.chat.completions.create(**params)
        return response.choices[0].message.model_dump()

    def _convert_tools_to_openai_format(self, mcp_tools: list) -> list[dict]:
        """Адаптер схемы MCP в формат OpenAI."""
        return [{
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.inputSchema
            }
        } for tool in mcp_tools]

Может показаться странным, что мы работаем с LLM через completions api, ведь responses api специально создавался для агентных систем. Тут важно понять, что LLM не имеет нативной поддержки протокола MCP. Взаимодействие происходит через tools. При использовании responses api, это взаимодействие происходит на стороне сервера. Это облегчает разработчику использование инструментов, однако обратиться к MCP серверу таким образом не получится. Рассмотрим более подробно процесс взаимодействия.

Анатомия памяти и Цикл ReAct

Модели не имеют памяти, поэтому список self.messages критически важен. Процесс наполнения памяти выглядит так:

Reasoning: Модель решает вызвать инструмент. Мы обязаны сохранить это намерение (tool_calls) в историю. Если этого не сделать, при получении результата модель "забудет", что она что-то запрашивала, и контекст сломается.
Acting: Метод handletool_calls() разбирает запрос и передает его в MCPClient. Хост при этом не знает, что именно делает инструмент — это "черный ящик".
Observation: Результат выполнения возвращается в историю с ролью tool.
Synthesis: Модель получает обновленную историю, видит результат и формирует финальный ответ.

Глава 4. Оркестратор: Сборка приложения

Финальный этап — скрипт, который объединяет все компоненты.

Код модуля (main.py)

import asyncio
import os
import sys
from dotenv import load_dotenv
from mcp_client import MCPClient
from agent_host import AgentHost

# Загрузка API ключа OpenAI
load_dotenv()

async def main():
    # Настройки
    openai_key = os.getenv("OPENAI_API_KEY")
    server_script = "mcp_file_server.py" 
    
    # Фиксированный запрос для демонстрации
    user_query = "Какие файлы лежат в рабочей папке? Если есть текстовые, прочитай любой из них."

    # Инициализация Клиента
    print(f"--- Запускаем сервер: {server_script} ---")
    mcp_client = MCPClient(
        command=sys.executable, # Текущий интерпретатор 
        args=[server_script]
    )

    try:
        await mcp_client.connect()
        print("✅ Сервер подключен")

        # Инициализация Агента
        agent = AgentHost(
            mcp_client=mcp_client,
            openai_api_key=openai_key,
            model="gpt-4.1"
        )

        # Запуск
        print(f"\nВопрос: {user_query}")
        response = await agent.process_query(user_query)
        
        print("\n--- Ответ ИИ ---")
        print(response)

    finally:
        # Обязательная очистка ресурсов
        await mcp_client.cleanup()
        print("\n--- Соединение закрыто ---")

if __name__ == "__main__":
    asyncio.run(main())

Этот скрипт не содержит бизнес-логики. Он лишь соединяет компоненты. Стоит отметить

Универсальность запуска: Мы используем sys.executable, чтобы гарантировать, что сервер запустится в том же виртуальном окружении, что и Host, и с теми же версиями библиотек.
Конфигурация: В отличие от более сложных систем, здесь мы не передаем env серверу, так как работа с локальными файлами не требует API-ключей.
Статичный запрос: Переменная user_query задана жестко для наглядности тестирования цикла ReAct. В реальном приложении здесь был бы цикл input() или API веб-сервиса.

Для тестирования попробуем создать в директории проекта папку demo_files и поместить туда несколько файлов. Запускаем и наблюдаем MCP-магию:

Заключение

Стоила ли игра свеч?

Может показаться, что мы просто усложнили стандартный Function Calling. (Спойлер: так оно и есть). Но есть нюансы:

В обычном подходе мы хардкодим функции передаваемые в LLM. Если мы потом захотели бы внести изменения, например выдавать не все файлы в папке, а только с определенным расширением - нам пришлось бы переписать не только серверную функцию, но и исправлять схему tools, валидировать ее. FastMCP делает это “под капотом”. Клиент запрашивает схемы инструментов автоматически.
Несмотря на все “болячки”, MCP комьюнити развивается. Доступно большое количество готовых серверов.Код клиента и хоста из данной статьи с небольшими доработками вполне можно использовать для разного рода интеграций.

Что дальше?

В этом руководстве мы создали базовый прототип. Для полноценного продакшн-решения могут понадобиться следующие решения (большинство из них мы упоминали ранее):

Отладка через MCP Inspector. Проверяйте сервер с помощью официального инструмента npx @modelcontextprotocol/inspector. Это позволит вам вручную вызвать инструменты и убедиться, что сервер работает как надо.
Логирование через MCP. Протокол поддерживает отправку логов от Сервера к Хосту. Вместо того чтобы писать ошибки в stderr, сервер может отправлять сообщения уровней info, warning или error, которые Хост будет показывать пользователю или записывать в файл.
Внедрение Ресурсов. В нашем примере мы использовали только Инструменты (Tools). Для пассивных данных (например список файлов), которые модель должна "видеть" всегда, стоит реализовать поддержку Ресурсов (Resources). Это позволит подписываться на изменения данных без постоянного опроса сервера.
Конфигурация: Передачу env в MCPClient для серверов, требующих авторизации (например, Postgres или SMTP).
Динамический ввод: Замену фиксированного user_query на цикл input() или веб-интерфейс.
Абстракция провайдера: Создание класса LLMProvider для легкого переключения между OpenAI, Anthropic и локальными моделями.

Если интересна данная тема - пишите в комментариях предложения, что включить в следующие выпуски. Можно усовершенствовать клиент и сервер из нынешней статьи, попробовать подключить сервер с гитхаб, ну а может даже эксперимент из серии “не пытайтесь это повторить” - попробовать использовать транспорт HTTP+SSE, разобраться с безопасностью.

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