От монолита к модулям: строим масштабируемую архитектуру AI-агентов с FastMCP и LangChain

Привет, Хабр!

Последний год мы были участниками бума разработки AI-агентов, в процессе которого мы сталкивались с определёнными трудностями:

• Проблемы с интеграциями

• Галлюцинации

• Переполнение контекста

• Зацикливание

Но самая большая проблема, с которой я столкнулся, — архитектура и масштабирование.

Писать монолитных агентов удобно, но с ростом количества проектов и инструментов возникают определенные трудности:

• Копирование функционала из одного агента в другого

• Сложность тестирования

• Необходимость менять большое количество кода для добавления нового инструмента

Поэтому нам жизненно необходимо создавать правильную архитектуру агентских систем, которую можно будет легко масштабировать и тестировать.

В этой статье я расскажу, как создать MCP-сервер с помощью библиотеки fastmcp и как подключиться к нему с использованием LangChain, а также рассмотрим примеры работы нового React-агента.

MCP

На помощь к нам приходит Anthropic со своим протоколом для взаимодействия агентов с инструментами.

Model Context Protocol (MCP) - открытый стандарт, представленный компанией Anthropic в ноябре 2024 года. До его появления индустрия находилась в состоянии фрагментации: каждый разработчик AI-агента или IDE писал свои собственные коннекторы к базам данных, GitHub, Slack или Google Drive. Это порождало проблему M×NM×N, где MM — количество AI-приложений (Claude Desktop, Cursor, LangChain-агенты), а NN — количество внешних сервисов.

Основная цель MCP — создать универсальный интерфейс для подключения AI-моделей к внешним данным и инструментам. Создатели сравнивают его с портом USB-C: вместо того чтобы искать уникальный кабель для каждого устройства, вы используете один стандартный разъем. Один раз написав MCP-сервер (например, для доступа к внутренней базе знаний компании), вы можете подключить его к любому MCP-клиенту — будь то локальное приложение Claude Desktop или агент на LangChain.

Архитектура взаимодействия:

Протокол работает по классической клиент-серверной схеме

1. MCP Host (Клиент): Приложение, в котором находится LLM (Наш агент или какая то другая программа). Оно инициирует соединение.

2. MCP Server: Легковесное приложение, которое предоставляет доступ к трем примитивам:

   • Resources (данные для чтения),

   • Tools (функции для выполнения),

   • Prompts (шаблоны запросов).

3. Транспорт: Общение происходит либо через локальные потоки ввода-вывода (stdio), либо через HTTP для удаленных подключений.

Cам протокол агностичен к языку программирования. Вы можете написать сервер на Python (с помощью fastmcp), Rust или Node.js, а клиент будет взаимодействовать с ним одинаково.

FastMCP

Про fastmcp:

FastMCP - высокоуровневая Python-библиотека для быстрой разработки MCP-серверов и клиентов. Она оборачивает спецификацию Model Context Protocol (MCP) и даёт удобный API для объявления инструментов, ресурсов, шаблонных промптов

uv pip install fastmcp

Основные компоненты

Библиотека разграничивает элементы для создания агентов на 3 группы (в дальнейшем я подробнее рассмотрю каждый из них):

• Tools — инструменты, которые предоставляет сервер и которые может самостоятельно вызывать языковая модель.

Создать инструмент в fastmcp так же просто, как и в LangChain. Достаточно добавить декоратор:

@mcp.tool def add(a: int, b: int) -> int: """Adds two integer numbers together.""" return a + b

• Resources & Templates — это механизм FastMCP для предоставления языковой модели или клиентскому приложению данных в режиме «только для чтения». Главное отличие от инструментов — они не выполняют никаких действий и не вызываются непосредственно моделью (не передаются в список инструментов агента).

Все элементы в fastmcp можно создать с помощью декоратора:

@mcp.resource("data://config") def get_config() -> dict: """Provides application configuration as JSON.""" return { "theme": "dark", "version": "1.2.0", "features": ["tools", "resources"], }

• Prompts - создает параметризованные шаблоны сообщений, которые помогают LLM генерировать структурированные ответы.

Наиболее частый случай использования — загрузка готовых промптов для разных моделей в IDE и добавление специфичного контекста. Это освобождает вас от задачи самостоятельного написания запроса. Но помимо этого их можно использовать для быстрого и удалённого изменения поведения агента. Пример создания:

@mcp.prompt def ask_about_topic(topic: str) -> str: """Generates a user message asking for an explanation of a topic.""" return f"Can you please explain the concept of '{topic}'?"

Сервер

Для того чтобы создать сервер достаточно написать:

from fastmcp import FastMCP mcp = FastMCP()

Теперь у нас есть экземпляр, который готов работать. Но при реальной разработке я советую использовать дополнительные параметры:

mcp = FastMCP( name="MathMCPServer", instructions=''' Сервер для математических операций ''', version='1.0', website_url="@ViacheslavVoo", on_duplicate_tools="error", ''' on_duplicate_tools - если вы каким то образом добавите несколько одинаковых интрументов, то при запуске сервера ваш код упадет с ошибкой ''' tools=[test_func, test_func], ''' tools = [func1, func2] - позволяет не навешивать декоратор @mcp.tool на каждую функцию. Особенно полезно когда функции находятся в других модулях. Важно чтобы функция отвечала требованиям инструмента ''' include_tags = ["public"], exclude_tags = ["deprecated"] ''' include_tags - показывает компоненты у которых есть хотя бы один совпадающий тег exclude_tags - скрывает компоненты с любым совпадающим тегом Приоритет: теги исключения всегда имеют приоритет над тегами включения ''' mask_error_details=True #скрывает ошибки от языковой модели # пример использования в блоке тестирования агента )

Запуск сервера

Транспортные протоколы

Вы можете поднять mcp с использованием нескольких протоколов (для локального и удаленного развертывания):

• STDIO (стандартный ввод/вывод) — это транспортный протокол по умолчанию для серверов FastMCP. Если вы вызываете run() без аргументов, ваш сервер использует транспортный протокол STDIO. Этот протокол обеспечивает связь через стандартные потоки ввода и вывода, что делает его идеальным для инструментов командной строки и приложений, таких как Claude Desktop.

Сервер считывает сообщения MCP из стандартного потока ввода и записывает ответы в стандартный поток вывода, поэтому серверы STDIO не работают постоянно — они запускаются по требованию клиента.

STDIO подходит для:

• Локальная разработка и тестирование

• Интеграция с Desktop инструментами

• Командной строки

• Однопользовательские приложения

HTTP

• HTTP-транспорт превращает ваш MCP-сервер в веб-сервис, доступный по URL-адресу. Этот транспорт использует потоковый HTTP-протокол, который позволяет клиентам подключаться по сети. В отличие от STDIO, где каждому клиенту выделяется отдельный процесс, HTTP-сервер может одновременно обслуживать несколько клиентов.

HTTP-протокол обеспечивает двустороннюю связь между клиентом и сервером и поддерживает все операции MCP, включая потоковую передачу ответов. Поэтому он рекомендуется для сетевых развертываний.

HTTP-транспорт обеспечивает:

• Доступность сети

• Несколько одновременных подключений

• Интеграция с веб-инфраструктурой

• Возможность удаленного развертывания

Примеры запуска:

mcp.run()

mcp.run(transport="http", host="0.0.0.0", port=8000)

Помимо этих способов fastmcp поддерживает SSE, но этот вариант считается устаревшим.

Инструменты

Как я уже говорил, инструмент в fastmcp — это такой же инструмент, как в LangChain. Основные требования к инструментам, которые улучшат работу агента:

• Функция должна иметь doc string (обязательное требование)

• Название функции и наименования аргументов должны соответствовать назначению

• Аргументы должны иметь аннотации типов

Инструменты проще всего создать с использованием декоратора @tool

@mcp.tool def add(item) -> str: """Some desc""" return "entry has been added to the list"

В этом случае декоратор возьмёт всю информацию из описания функции. Если вы хотите использовать другое описание и/или добавить информацию, вы можете использовать дополнительные параметры:

• name="add_in_list" - имя функции, которое будет использоваться вместо указанного

• description="adds an entry to the list" - новое описание doc string

• tags={"list", "add", "public"} - метки, с помощью которых можно фильтровать инструменты

• meta={"version": "1.2", "author": "product-team"}

• exclude_args - позволяет скрыть аргументы инструмента от языковой модели (это могут id, ключи и другая информация, которая не должна попадать в LLM)

Здесь я хочу подробнее остановиться на создании и изменение инструментов

Создание инструмента из функции, к которой у нас нет доступа напрямую. Например, если вы хотите использовать встроенную функцию или какой-то код из legacy

import secrets from fastmcp import Tool # Функция из стандартной библиотеки Python для генерации безопасных токенов token_tool = Tool.from_function( secrets.token_hex, name="generate_secure_token", description=""" Generates a cryptographically strong random hex string. Use this when the user needs a secure API key, password, or unique identifier. Default length is 32 bytes if not specified. """ )

Добавление описания к аргументам функции.

Модель будет лучше ориентироваться в инструментах, если будет знать не только описание функции, но и описание аргументов. Когда у нас есть возможность, то мы можем сделать это с помощью Field или Annotated:

@mcp.tool def find_user_field( user_id: Annotated[str, ''' "The unique identifier for the user, " "usually in the format 'usr-xxxxxxxx'."'''] ): """Finds a user by their ID.""" ...

Но такая возможность есть не всегда, поэтому fastmpc позволяет оборачивать функции в инструменты и добавлять к ним описание:

@mcp.tool def find_user(user_id: str): """Finds a user by their ID.""" ... new_tool = Tool.from_tool( find_user, transform_args={ "user_id": ArgTransform( description=( "The unique identifier for the user, " "usually in the format 'usr-xxxxxxxx'." ) ) } )

Если посмотреть схему, то мы увидим практически одинаковое описание:

name='find_user' title=None description='Finds a user by their ID.' icons=None tags=set() meta=None enabled=True parameters={'type': 'object', 'properties': {'user_id': {'type': 'string', 'description': "The unique identifier for the user, usually in the format 'usr-xxxxxxxx'."}}, 'required': ['user_id']} name='find_user' title=None description='Finds a user by their ID.' icons=None tags=set() meta=None enabled=True parameters={'properties': {'user_id': {'description': ' "The unique identifier for the user, "\n "usually in the format \'usr-xxxxxxxx\'."', 'type': 'string'}}, 'required': ['user_id'], 'type': 'object'}

Помимо добавления описания можно изменять названия аргументов, устанавливать default value и скрывать часть аргументов:

@mcp.tool def search(q: str): """Searches for items in the database.""" return "database.search(q)" new_tool = Tool.from_tool( search, transform_args={ "q": ArgTransform(name="search_query", default=10) } )

Скрытие аргументов:

def send_email(to: str, subject: str, body: str, api_key: str, timestamp): """Sends an email.""" ... new_tool = Tool.from_tool( send_email, name="send_notification", transform_args={ "api_key": ArgTransform( hide=True, default=os.environ.get("EMAIL_API_KEY"), ), 'timestamp': ArgTransform( hide=True, default_factory=lambda: datetime.now(), ) } )

Клиент

Когда мы разобрались с созданием сервера и инструментов, пора подключить к ним клиента. Для разработки я использую LangChain, поэтому примеры будут с использованием этой библиотеки.

• Экземпляр класса MultiServerMCPClient

#make_mcp_client from langchain_mcp_adapters.client import MultiServerMCPClient #Используем возможности langchain async def make_client(server_config: dict) -> MultiServerMCPClient: multi_client = MultiServerMCPClient(server_config) return multi_client ''' Примечание: у fastmcp также есть класс Client, который принимает url сервера, но я учитываю, что сервер и клиент находятся в разном окружении, а установка fastmcp в окружение клиента ради одного класса излишне '''

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

MCP_URL = "http://127.0.0.1:8081/mcp" MCP_CONFIG = { "my_tools": { "transport": "streamable_http", "url": MCP_URL, } }

• Получение инструментов от сервера

#load_tools from agent_mcp.mcp_connection.make_mcp_client import make_client async def load_tools(server_config: dict) -> list: """Получает инструменты от MCP-сервиса. Если сервис недоступен — бросает ConnectionError.""" client = await make_client(server_config) try: tools = await client.get_tools() except Exception as e: raise ConnectionError(f"Не удалось получить инструменты от сервиса: {e}") for tool in tools: tool.handle_tool_error = True #позже покажу зачем нам эта строка _langchain_tools = tools return _langchain_tools

Таким образом в main файле агента достаточно написать:

#main MCP_TOOLS_URL = "http://127.0.0.1:8080/mcp" MCP_TOOL_CONFIG = { "graphics-tools": { "transport": "streamable_http", "url": MCP_TOOLS_URL, } } _langchain_tools: Optional[list] = None async def load_agent_tools() -> list: """Получает инструменты от MCP-сервиса. Если сервис недоступен — бросает ConnectionError.""" global _langchain_tools, MCP_TOOL_CONFIG if _langchain_tools is not None: return _langchain_tools _langchain_tools = await load_tools(MCP_TOOL_CONFIG) return _langchain_tools

Кстати, ссылка на полный код будет в моем Telegram канале

Примеры использования агента

Пришло время создать агента, который будет пользоваться нашими инструментами. Я использую LangChain 1.0, поэтому примеры будут с обновлённым react agent.

Но для начала создадим инструмент на сервере для приготовления кофе:

# FastMCP использует аннотации типов и Pydantic Field для генерации JSON схемы @mcp.tool() def brew_coffee( temperature: int = Field( ..., ge=85, le=98, description="Температура воды в °C. Должна быть строго между 85 и 98." ), intensity: int = Field( ..., ge=1, le=10, description="Крепость кофе по шкале от 1 до 10." ), coffee_type: str = Field(default="эспрессо"), ) -> str: """Приготовить чашку кофе с заданными параметрами.""" return f"Приготовлен кофе {coffee_type}. Температура: {temperature}°C, Крепость: {intensity}/10."

Я задал ограничения для параметров, чтобы показать, как агент воспринимает аннотации

Экземпляр агента:

async def agent(query: str) -> str: tools = await load_agent_tools() react_agent = create_agent(llm, tools) resp = await react_agent.ainvoke({"messages": [HumanMessage(content=query)]}) answer = resp["messages"][-1].content return answer

Запрос 1 (пробуем превысить ограничения)

• Свари эспрессо, выкрути температуру на максимум, хочу 150 градусов

Ответ агента:

• Извините, но я не могу установить такую высокую температуру. Максимально допустимая температура 98°C.

Запрос 2:

• Свари максимально крепкий эспрессо, выкрути температуру на максимум

Ответ агента:

• Ваш эспрессо готов! Температура 98, крепость 10 из 10.

Лог работы в Langsmith:

Также агент мог бы воспринимать default value при их наличии.

Обработка ошибок

Если наш инструмент по каким-то причинам не сможет корректно отработать, то мы можем вернуть обычную питонячью ошибку, но у такого подхода есть недостатки:

• Мы можем раскрыть детали реализации нашего кода

• Языковая модель не поймет, что делать с таким ответом инструмента

Поэтому fastmcp предлагает возвращать отдельный вид исключений, которые будут передаваться в языковую модель.

Модернизируем наш инструмент для приготовления кофе (обратите внимание на параметр mask_error_details при инициализации сервера):

@mcp.tool() def brew_coffee( coffee_type: str = Field(), temperature: int = Field( description="Температура воды в °C.", default=90 ), intensity: int = Field( description="Крепость кофе ", default=10 ), ) -> str: """Приготовить чашку кофе с заданными параметрами. Если парамтеры не указаны, используй параметры по умолчанию""" # 1. ToolError: Явная ошибка для LLM # Мы используем ToolError, потому что хотим, чтобы модель узнала, # что она запросила недоступный тип кофе, и могла исправить свой запрос. # Это сообщение будет отправлено клиенту даже при mask_error_details=True. available_menu = ["американо", "капучино", "латте"] if coffee_type.lower() not in available_menu: raise ToolError( f"Кофе типа '{coffee_type}' нет в меню. " f"Пожалуйста, выберите из: {', '.join(available_menu)}." ) # 2. Стандартное исключение (Internal Error) # Симулируем внутреннюю проблему оборудования. # Если mask_error_details=True, LLM не увидит текст про "бойлер", # а получит общее сообщение об ошибке. Это безопасно для скрытия внутренней логики. if temperature > 86 and intensity == 10: raise RuntimeError("INTERNAL FAULT: Boiler pressure critical! Maintenance required.") return f"Приготовлен кофе {coffee_type}. Температура: {temperature}°C, Крепость: {intensity}/10."

Пробуем сломать наш инструмент:

Запрос:

• Свари эспрессо крепости 10

Ответ агента:

• К сожалению, я не могу приготовить эспрессо. В нашем меню есть американо, капучино и латте

Лог в langsmith:

Запрос 2:

• Свари американо температуры 90 крепости 10

Ответ агента:

• Извините, у меня не получилось приготовить американо. Пожалуйста, попробуйте еще раз.

Лог в langsmith:

Передача скрытых аргументов

Бывают ситуации, когда инструменту необходимо знать какой-нибудь id или ключ, который не должна видеть языковая модель. Создадим такой простой инструмент на сервере:

@mcp.tool( exclude_args=["user_id"] ) def add_item_to_card(item_name: str, quantity: int, user_id: str = None) -> str: '''Добавить товар в корзину пользователя''' return f"Товар {item_name} добавлен в корзину. user_id: {user_id}"

и добавим к агенту функцию для перехвата вызова инструментов:

@wrap_tool_call async def safe_inject_params(request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage | Command], ) -> ToolMessage | Command: tool_name = request.tool_call['name'] original_args = request.tool_call['args'] new_args = original_args.copy() if tool_name == "add_item_to_card": # Добавляем скрытый параметр new_args['user_id'] = "123" # Подменяем аргументы в запросе request.tool_call['args'] = new_args try: result = await handler(request) print(f"Tool completed successfully") return result except Exception as e: print(f"Tool failed: {e}") raise

react_agent = create_agent(llm, tools, middleware=[safe_inject_params])

Такую возможность нам предоставляет новая версия LangChain. Возможно, я как-нибудь напишу статью с разбором последнего обновления).

Запрос 1:

• Добавь в корзину 2 упаковки зернового кофе

Ответ агента:

• Добавлено 2 упаковки зернового кофе в корзину.

Лог в langsmith:

Ресурсы и шаблоны

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

Создание ресурса:

@mcp.resource("config://vibe") def get_vibe(): return "Сегодня ты должен отвечать как суровый системный администратор из 90-х." @mcp.resource("functions://with-hidden-id") def get_func_names_with_id() -> list: """ Возвращает список функций, которые содержат скрытый параметр 'id'.""" return [...]

При их создании также можно указать дополнительные параметры:

@mcp.resource( uri="data://app-status", name="ApplicationStatus", description="Provides the current status of the application.", mime_type="application/json", tags={"status"}, meta={"version": "2.1", "owner": "Viacheslav"} )

В langchain ресурсы можно загрузить с помощью get_resources:

client = MultiServerMCPClient({...}) # Load all resources from a server res = await client.get_resources("server_name") res = await client.get_resources("server_name", uris=["file:///path/to/file.txt"])

Пример использования:

async def complex_analysis_pipeline(query: str): """Анализа с предзагрузкой данных""" # 1. Параллельная загрузка нескольких ресурсов resource_uris = [ "metrics://system/health", "logs://errors/recent", "config://current-settings" ] # Асинхронная загрузка всех ресурсов resources = await asyncio.gather(*[ client.get_resources(uri) for uri in resource_uris ]) # 2. Объединение контекста combined_context = "\n\n".join([ f"## {uri}\n{data}" for uri, data in zip(resource_uris, resources) ]) # 3. Создание промпта с богатым контекстом prompt = ChatPromptTemplate.from_messages([ ("system", """Ты эксперт по анализу систем. У тебя есть следующие данные: {context} Анализируй проблему шаг за шагом."""), ("human", "{question}") ]) chain = prompt | ChatOpenAI(temperature=0) return await chain.ainvoke({ "context": combined_context, "question": query })

Prompts

Я нашёл для себя вариант использования prompt в виде быстрой замены системных промптов у агента. Это можно было бы делать с помощью обычного обращения к базе, но я предпочитаю использовать минимально возможный стек технологий. Здесь я приведу примеры максимально упрощённого использования, чтобы вы сами нашли для себя способы применения.

@mcp.prompt def generate_code_request(language: str, task_description: str) -> dict: """Запрос на генерацию кода""" content = f"Write a {language} function that performs: {task_description}" return { "role": "user", "content": content }

#Сервер @mcp.prompt def analyze_data( numbers: list[int], metadata: dict[str, str], threshold: float ) -> str: """Анализ числовых данных""" return f"Analyze numbers: {numbers} with metadata: {metadata}. Threshold: {threshold}" #Клиент # Создаем инструмент из FastMCP промпта def data_analysis_tool(inputs: dict) -> str: prompt_text = analyze_data( inputs["numbers"], inputs["metadata"], inputs["threshold"] ) return llm([HumanMessage(content=prompt_text)]).content

Заключение

Относительно недавно мы столкнулись с ИИ. В прошедшем году мы пробовали, экспериментировали, сталкивались с определёнными трудностями, но главное — приобретённый опыт. Этот опыт показывает, что хорошо продуманная и легко масштабируемая архитектура — огромный вклад в создание действительно полезного агента, которым можно будет пользоваться.

Связка fastmcp и LangChain — отличное сочетание, которая решает проблемы:

• Масштабирование: Разделение сервера (MCP) и клиента (агент) позволяет обновлять инструменты независимо, запускать сервера на разных нодах.

• Тестирование: MCP-сервер можно тестировать изолированно, мокать, что упрощает CI/CD.

Если хотите обсудить архитектуру AI-агентов или поделиться своим опытом — добро пожаловать в мой Telegram