Привет, Хабр! Меня зовут Владимир, и это внеплановое продолжение цикла про локального MCP-агента. В первых частях мы собрали агента, научили его работать через LangGraph, добавили Gradio-интерфейс, чекпоинты и трассировку. Но как только агентом пришлось поделиться с коллегами, всплыл следующий бытовой вопрос: как понять, кто именно им пользуется, и как разнести историю, чекпоинты и трейсы по пользователям. Вычислять по IP не вышло (может не разобрался, но контейнер с приложением в Docker Desktop из-за связки Windows-WSL-Docker всегда показывал мой IP-адрес), и я решил прикрутить к Gradio аутентификацию.

В этой статье я покажу как добавить аутентификацию в Gradio-интерфейс и как использовать её для упорядочивания трейсов в Langfuse и чекпоинтов в LangGraph.

Окно входа в GUI

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

Но у нас тут не серьёзный фронтенд, а миленький Gradio-интерфейс для пары коллег. У Gradio нет полноценного встроенного окна регистрации: пользователей надо заводить самому, либо подключать внешнюю аутентификацию и отдельную логику управления аккаунтами. Переделывать GUI ради регистрации мне не хотелось, поэтому для начала я выбрал самый простой вариант: хранить выданные логины и пароли в текстовом файле.

# Формат: логин:пароль (по одному на строку)
admin:admin

Теперь файл надо распарсить и передать результат аутентификации в Gradio. Вообще Gradio предполагает использование не только функции, но и реализует возможность просто передать кортеж (логин, пароль). Не знаю почему, но прямая передача паролей немного противоречит моему чувству прекрасного, поэтому сделаем функцию. Если в будущем возникнет желание заменить текстовый файл на что-то с хешем паролей, то менять GUI не придётся (хотя формально там одна строчка кода):

import secrets
from pathlib import Path
from core import settings

def get_gradio_auth():
    if not settings.service.auth_enabled:
        return None
    path = Path(settings.service.auth_users_file)
    if not path.is_file():
        raise RuntimeError(f'Файл учётных записей не найден: {path}')

    credentials = {}
    for line in path.read_text(encoding='utf-8').splitlines():
        if not line or line.startswith('#') or ':' not in line:
            continue
        user, psw = line.split(':', 1)
        if (user := user.strip()) and (psw := psw.strip()):
            credentials[user] = psw

    if not credentials:
        raise RuntimeError(f'Файл учётных записей пуст или некорректен: {path}')

    def checker(username: str, password: str) -> bool:
        stored = credentials.get(username)
        return stored is not None and secrets.compare_digest(stored, password)

    return checker

Читаем и парсим файл паролей. По соглашению с самим собой символ # в файле - это комментарий, поэтому строки, начинающиеся с # пропускаем. Также, на всякий случай, пропускаем строки в которых нет разделителя :. Для каждой валидной строки извлекаем и сохраняем пару логин-пароль. Абонентов планировалось до трёх человек, поэтому отдельное хранилище не заводил: файл читается при старте приложения, а дальше функция аутентификации работает с уже разобранным словарём.

Далее замыкание на функцию проверки учётных данных. Почему в ней не использую просто if password == credentials.get(username) - ну это же проверка паролей, должна же быть безопасность ?.

Доработаем интерфейс:

class MCPCodingAgentApp:
    # Доработанный кусочек кода старого класса
    async def initialize_and_launch(self):
        await self.init_agent()
        self.build_interface()
        self.demo.launch(
            server_name=settings.service.host,
            server_port=settings.service.port,
            show_error=True,
            auth=get_gradio_auth(),  # Новая строка
            auth_message='<p>Введите выданный Вам логин и пароль</p>'  # Новая строка
        )

Вся интеграция аутентификации сводится к тому, что в метод demo.launch необходимо передать два дополнительных параметра - функцию проверки учётных данных (или, как говорил ранее, кортеж логин-паролей) и приветственное сообщение окошка ввода логина - пароля.

Для сохранения истории и фильтрации трассировок по пользователям сразу доработаем вызовы графа. Для этого необходимо передать пользователя во все методы графа. Покажу на примере метода продолжения диалога:

class MCPCodingAgentApp:
    # Доработанный кусочек кода старого класса
    async def _continue(self, history: list, request_id: str, request: gr.Request):
        phase = await self.agent.get_phase(request_id, username=request.username)

        if phase is None or phase == 'done':
            history.append({
                'role': 'assistant',
                'content': 'Нет активных задач для продолжения. Задайте новый вопрос.'
            })
            return history, request_id

        result = await self.agent.resume(
            user_messages='Продолжить', request_id=request_id, username=request.username)

Трассировка по пользователям

В предыдущей версии наш thread_id был завязан просто на request_id. Сейчас появилась новая сущность - пользователь, и его надо как-то связать с идентификацией:

class MCPAgent:
    # Новый кусочек кода старого класса
    @staticmethod
    def resolve_thread_id(username: str | None, request_id: str) -> str:
        return f'{username or "anonymous"}:{request_id}'

Если аутентификация выключена в настройках, request.username будет None и мы не получим от фронта username, поэтому используем username or "anonymous".

В коде класса MCPAgent во всех методах, которые используют request_id, необходимо добавить входной параметр username: str | None = None и вместо {'thread_id': request_id} использовать {'thread_id': self.resolve_thread_id(username, request_id)}.

Теперь осталось передать новые атрибуты для обеспечения фильтрации трейсов. Для этого используем метод для проброса атрибутов трассировки langfuse.propagate_attributes:

class MCPAgent:
    # Доработанный кусочек кода старого класса
    async def _ainvoke_with_tracing(self, data, request_id, trace_id, span_name, username, trace_input) -> list:
        user = username or 'anonymous'
        session_id = f'{user}_{request_id[:8]}'

        with settings.langfuse.client.start_as_current_observation(
                as_type='span',
                name=span_name,
                trace_context={'trace_id': trace_id},
        ) as span:
            with propagate_attributes(
                user_id=user,
                session_id=session_id,
                trace_name='MCP Agent',
                tags=['langgraph', 'mcp-agent'],
                metadata={'request_id': request_id, 'username': user},
            ):
                span.update(input={'message': trace_input})
                result = await self.graph.ainvoke(
                    data,
                    config=self._create_config(request_id, username, trace_id),
                )
                span.update(output=str(result['messages'][-1].content))
                return result['messages']

Начало похоже на старую реализацию - начинаем наблюдение типа span с конкретным trace_id. А дальше прокидываем в трейс параметры с помощью контекстного менеджера langfuse.propagate_attributes. Параметрами он принимает идентификатор пользователя, идентификатор сессии, словарик с метаданными и теги для трейса. В данном варианте в качестве идентификатора пользователя передаём его логин.

Ну и раз появилась возможность отличать диалоги разных пользователей, то перейдём от InMemorySaver к PostgreSQL-чекпоинтеру (AsyncPostgresSaver).

Checkpointer в Postgres

Для начала установим библиотеку:

uv add langgraph-checkpoint-postgres==2.0.0

Напишем метод создания чекпоинтера:

from loguru import logger
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from psycopg import AsyncConnection
from psycopg.rows import dict_row

async def create_checkpointer(db_url: str) -> AsyncPostgresSaver:
    try:
        conn = await AsyncConnection.connect(
            db_url,
            autocommit=True,
            row_factory=dict_row,
        )
        checkpointer = AsyncPostgresSaver(conn=conn)
        await checkpointer.setup()
        logger.success('Postgres checkpointer инициализирован')
        return checkpointer
    except Exception as e:
        raise RuntimeError(f'Не удалось подключиться к Postgres для checkpoint ({db_url}): {e}')

Создаём psycopg-соединение с базой данных в асинхронном режиме. В параметрах передаём адрес для подключения к БД, устанавливаем флаг автокоммита и формат возврата значений в виде dict. Следующим действием создаём и инициализируем Postgres-чекпоинтер. При первом запуске метод AsyncPostgresSaver.setup() создаст необходимые для чекпоинтера таблицы и выполнит миграции.

Заменим InMemorySaver на Postgres-чекпоинтер в агенте. Так как инициализация чекпоинтера теперь асинхронная, то в классе MCPAgent её необходимо перенести из конструктора в метод init_graph:

class MCPAgent:
    def __init__(self):
        self.llm_with_tools: ChatOpenAI | None = None
        self.tools: list[BaseTool] | None = None
        self.graph = None
        self.checkpointer: AsyncPostgresSaver | None = None  # AsyncPostgresSaver вместо InMemorySaver

    async def init_graph(self):
        try:
            self.tools = await init_tools()
        except Exception as e:
            logger.error(f'Ошибка инициализации инструментов: {e}')
            raise
        self.checkpointer = await create_checkpointer(settings.service.checkpoint_db_url)
        self.llm_with_tools = settings.llm.chat.llm.bind_tools(self.tools, parallel_tool_calls=False)
        self.graph = self._compile_graph()

Скриншотики

Это окно входа

Это я логин перепутал

А это я два раза один запрос из разных юзеров дал

А это так теперь фильтроваться можно

Итоги

В итоге у Gradio-интерфейса появилась простая аутентификация: пользователи вводят выданные логин и пароль, а приложение проверяет их через функцию, переданную в параметр auth. Главное не забыть .env файле указать AGENT__AUTH_ENABLED=true.

Этого оказалось достаточно, чтобы связать запросы с конкретными пользователями: разнести thread_id, аккуратнее хранить чекпоинты в Postgres и добавить user_id, session_id в Langfuse-трейсы. Получилось не полноценное управление пользователями, но для внутреннего инструмента - вполне себе.

Код с аутентификацией (ветка auth) тут