Langfuse, помимо трассирования запросов, удобно использовать для prompt management и LLM‑as‑a-judge. Но в корпоративной среде он упирается в простую вещь: LF работает со статическим API key, а ваш inference gateway — нет. Если gateway требует короткоживущий JWT, Langfuse не умеет его получать. И в этот момент интеграция ломается.

Мы столкнулись именно с такой ситуацией. Модели self‑hosted, OpenAI подобный API, но для доступа к нему на каждый запрос нужен JWT, который выдаётся централизованным провайдером. Langfuse в LLM Connection умеет передать API key и заголовки, но не сможет сам сходить в auth‑сервис, получить временный токен и подставить его в запрос.

Идею предложил коллега — системный аналитик, а я ее реализовал: поставить между Langfuse и LLM API тонкий прокси. Он принимает обычный запрос от LF, достаёт из него идентификатор клиента и секрет, получает временный токен, подменяет Authorization и уже с ним идёт в gateway LLM.

Сразу оговорюсь: в этой статье я не буду подробно разбирать генерацию JWT. Для понимания общей картины достаточно знать, что JWT в нашем случае это короткоживущий подписанный access token, который удостоверяет клиента и используется для обращения к защищённому API. Нам важна не его внутренняя структура, а сам факт, что токен живёт ограниченное время и должен получаться динамически.

Зачем здесь вообще proxy

Если бы upstream LLM API принимал обычный статический ключ, Langfuse можно было бы подключить напрямую (так и работает с обычными провайдерами, например openrouter). Но когда доступ завязан на временные токены, появляется разрыв между двумя мирами:

Langfuse мыслит статическим connection‑конфигом, а модельный gateway мыслит временной авторизацией.

Proxy закрывает этот разрыв. Он делает три вещи:

Принимает OpenAI‑подобный запрос от Langfuse.
Меняет способ авторизации: из входных данных получает временный access token.
Прозрачно пересылает запрос дальше и возвращает ответ обратно.

По сути, это адаптер между Langfuse и вашей внутренней auth‑моделью.

Шаг 1. Поднимаем минимальное приложение на FastAPI

Начнём с простого FastAPI сервиса. Proxy в этом сценарии — это не CPU‑bound сервис.
Он делает несколько сетевых вызовов на каждый запрос:

запрос в auth‑сервис за JWT
запрос в upstream LLM API

Если обрабатывать их синхронно, поток будет простаивать во время ожидания I/O. Асинхронный FastAPI + httpx.AsyncClient позволяют:

не блокировать обработку других запросов
эффективно утилизировать соединения
выдерживать большую нагрузку при том же количестве ресурсов

То есть по сути, proxy — это I/O‑bound сервис, и async здесь даёт прямой прирост пропускной способности.

Подготовим структуру с основными модулями:

app.py для создания приложения и общей оркестрации;
routers.py для эндпоинтов;
proxy_service.py для основной логики прокси;
jwt_provider.py для получения токена;
settings.py для конфигурации.

Такой расклад удобен тем, что логика HTTP‑маршрутов, логика авторизации и логика конфигурации не смешиваются в один файл.

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.logging import configure_logging, get_logger
from app.proxy_service import LLMProxyService
from app.routers import root_router


def add_routes(app: FastAPI) -> None:
    app.include_router(root_router)


@asynccontextmanager
async def lifespan(app: FastAPI):
    configure_logging()
    logger = get_logger()
    logger.info("Microservice is starting...")

    proxy_service = LLMProxyService()
    app.state.proxy_service = proxy_service

    try:
        yield
    finally:
        await proxy_service.aclose()
        logger.info("Microservice is shutting down...")


def create_app() -> FastAPI:
    app = FastAPI(lifespan=lifespan)
    add_routes(app)
    return app

Шаг 2. Добавляем маршрутизацию

После этого добавим ендпоинты, которые будет вызывать LF. Для минимальной рабочей версии достаточно двух маршрутов:

POST /v1/chat/completions
GET /v1/models

Первый нужен для реальных вызовов модели, второй полезен для проверки доступных моделей и для совместимости с OpenAI‑подобными клиентами.
Сами роуты могут быть очень тонкими: приняли Request, достали из app.state proxy service и передали ему выполнение.

from fastapi import APIRouter, Request
from app.proxy_service import LLMProxyService


root_router = APIRouter()

def _get_proxy_service(request: Request) -> LLMProxyService:
    return request.app.state.proxy_service

@root_router.post("/v1/chat/completions")
async def proxy_chat_completions(request: Request):
    proxy_service = _get_proxy_service(request)
    return await proxy_service.proxy_chat_completions(request)

@root_router.get("/v1/models")
async def proxy_models(request: Request):
    proxy_service = _get_proxy_service(request)
    return await proxy_service.proxy_models(request)

@root_router.get("/healthz")
def healthz():
    return {"status": "ok"}

Отдельно можно оставить health‑check и прочие служебные пробы вроде /healthz, /live, /readyz. К логике прокси они отношения почти не имеют, но для Docker/Kubernetes это полезно.

Шаг 3. Выносим настройки в.env

Следующий шаг — не хардкодить адреса и таймауты. Нам понадобятся как минимум:

LLM_BASE_URL — адрес upstream LLM API;
SYSTEM_ID_HEADER_NAME — имя заголовка, в котором придёт идентификатор системы;
JWT_ISSUER — issuer вашего identity provider;
таймауты и SSL‑настройки для LLM и auth‑сервиса.

Удобнее всего описать это через pydantic‑settings и читать из.env.

from functools import lru_cache
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        extra="ignore",
    )

    llm_base_url: str = Field(default="", alias="LLM_BASE_URL")
    llm_timeout_seconds: float = Field(default=120.0, alias="LLM_TIMEOUT_SECONDS")
    system_id_header_name: str = Field(
        default="systemId", alias="SYSTEM_ID_HEADER_NAME")

    jwt_issuer: str = Field(default="", alias="JWT_ISSUER")
    jwt_scope: str = Field(default="", alias="JWT_SCOPE")
    jwt_timeout_seconds: float = Field(
        default=5.0, alias="JWT_TIMEOUT_SECONDS")

@lru_cache()
def get_settings() -> Settings:
    return Settings()

На практике это даёт две вещи: сервис проще деплоить и его проще переносить между локальной средой, стендом и продом.

Шаг 4. Делаем отдельный provider для получения токена

Теперь главное: прокси должен уметь по client_id и client_secret получить временный access token.

Здесь принципиально важно не зашивать эту логику прямо в роуты. Лучше вынести её в отдельный класс, который ничего не знает про FastAPI и занимается только одним: возвращает валидный access token.

В нашей реализации провайдер работает так:

Принимает client_id и client_secret.
Через OIDC discovery получает token_endpoint.
Отправляет запрос по client_credentials.
Достаёт access_token и срок его жизни.
Кэширует токен до истечения срока действия, чтобы не ходить к провайдеру на каждый запрос.

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

Шаг 5. Реализуем проксирование запроса

Теперь соединяем всё вместе в LLMProxyService.

Логика получается прямо очень прямой:

Проверяем, что задан LLM_BASE_URL.
Читаем обязательный заголовок из настроек(SYSTEM_ID_HEADER_NAME).
Читаем входной Authorization: Bearer....
Используем значение этого заголовка как client_id, а Bearer credential как client_secret.
Получаем временный токен через token provider.
Формируем upstream URL и пересылаем туда исходный запрос.

При пересылке важно сохранить оригинальные метод, query‑параметры и тело. Это делает proxy действительно прозрачным. Из заголовков обычно стоит исключить транспортные вещи вроде host, content-length, connection, а Authorization заменить на новый Bearer token, полученный от провайдера JWT.

Это ключевой момент всей схемы: для Langfuse снаружи всё выглядит как обычный OpenAI‑подобный endpoint, а для внутреннего gateway запрос уже приходит в нужном формате авторизации.

import httpx
from fastapi import Request, Response, status
from fastapi.responses import JSONResponse

from app.jwt_provider import JWTTokenProvider
from app.settings import get_settings


class LLMProxyService:
    def __init__(self) -> None:
        self._settings = get_settings()
        self._token_provider = JWTTokenProvider(settings=self._settings)
        self._client = httpx.AsyncClient()

    async def proxy_chat_completions(self, request: Request) -> Response:
        return await self._forward(request, "/chat/completions")

    async def proxy_models(self, request: Request) -> Response:
        return await self._forward(request, "/models")

    async def _forward(self, request: Request, path: str) -> Response:
        base_url = self._settings.llm_base_url.strip()

        # Извлекаем специальный header - имя пользователя для jwt провайдера 
        client_id = request.headers.get(self._settings.system_id_header_name)
        # Извлекаем пароль для провайдера, LF передает его как `Bearer <client_secret>`
        authorization = request.headers.get("authorization")
        scheme, _, client_secret = authorization.partition(" ")
     
        token = await self._token_provider.get_token(
            client_id=client_id,
            client_secret=client_secret,
        )

        upstream_url = f"{base_url.rstrip('/')}{path}"
        return await self._send(
            request=request,
            url=upstream_url,
            token=token,
            client_id=client_id,
            system_id_header_name=system_id_header_name,
        )

Важно: прокси получает client_secret из запроса, поэтому:

не логируйте Authorization заголовки
ограничьте доступ к прокси Сервис в этом случае становится частью security‑контура.

Шаг 6. Возвращаем ответ обратно клиенту

На выходе нам не нужно изобретать новую схему ответа. Если upstream LLM API вернул обычный OpenAI‑подобный JSON, лучше просто вернуть его обратно с тем же статус‑кодом и нужными заголовками ответа.

Это важный момент. Чем меньше proxy вмешивается в payload, тем легче его сопровождать и тем меньше риск сломать совместимость с клиентом.

Если upstream вернул ошибку, её тоже лучше не маскировать, а пробрасывать обратно. Proxy в этом сценарии должен быть не ещё одной бизнес‑логикой, а прозрачным адаптером.

Как это подключается в Langfuse

На стороне Langfuse идея такая:

в Base URL указываем адрес прокси, например http://your-proxy/v1;
в API Key кладём секрет клиента;
в custom headers передаём systemId или другое имя заголовка, которое ждёт прокси;
модельные запросы Langfuse отправляет уже не напрямую в gateway, а в наш proxy.

Получается удобная схема: Langfuse не знает ничего про получение токена, а proxy берёт эту обязанность на себя.

Что происходит во время одного запроса

Если разложить один вызов по шагам, картина будет такой:

Langfuse отправляет POST /v1/chat/completions в proxy.
Proxy читает systemId и Authorization.
Proxy получает временный access token у JWT provider.
Proxy отправляет тот же запрос в upstream LLM API, но уже с новым Authorization: Bearer
Модельный gateway обрабатывает запрос и возвращает ответ.
Proxy возвращает ответ без изменений.

Такой сервис должен оставаться минимальным. Его задача не «умно работать с LLM», а честно и предсказуемо переводить один способ авторизации в другой.

Что можно улучшить дальше

Минимальная версия уже решает основную проблему, но в продовой эксплуатации обычно хочется добавить ещё, например, streaming‑ответы, метрики и трассировку. Но для первой версии всё это можно отложить. Сначала полезно получить простой, понятный и легко отлаживаемый путь от Langfuse до модели.

Итог

Главная идея проста: не нужно пытаться научить Langfuse работать с JWT — проще поставить тонкий адаптер между ним и вашим gateway.

Это даёт несколько плюсов:

вы не лезете в сложную систему Langfuse
контроль над авторизацией полностью на вашей стороне
переиспользуемый слой для любых LLM

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

Зачем вам нужен прокси для Langfuse