23.12.2025 07:00
Viacheslav-hub 0 4100 Источник

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

Последний год мы были участниками бума разработки 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

Комментарии (0)