Мой личный джуниор. Бонус: аутентификация в Gradio и пользователи в Langfuse

Привет, Хабр! Меня зовут Владимир, и это внеплановое продолжение цикла про локального 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) — ну это же проверка паролей, должна же быть безопасность 😁.

Кому интересно про `secrets.compare_digest`

Метод secrets.compare_digest обеспечивает защиту от атак по времени выполнения (timing attack) — грубо говоря, вычисление секретов по времени ответа бекенда. Сравнение if password == credentials.get(username) идёт посимвольно и возвращает False при первом несовпадении, что может быть использовано для подбора пароля. secrets.compare_digest же обеспечивает сопоставимое время сравнения для данных одинаковой длины.

Для системы, где пароли хранятся в текстовом файле — это перебор, но совсем расслабляться не стоит

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

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) тут

Источник: habr.com