В Pingera мы активно развиваем наш Model Context Protocol (MCP) сервер, который является основой для наших внутренних AI-агентов. FastMCP — это легковесный и высокопроизводительный фреймворк на Python для построения таких серверов, управляющих вызовами инструментов для языковых моделей. Изначально сервер работал в простом STDIO-режиме, идеально подходящем для локальных инсталляций, где аутентификация проходила через единственный API-ключ из переменных окружения.
Однако, чтобы использовать наш MCP сервер как сетевой сервис для работы множества пользователей (в режиме multi-tenancy), необходимо было обеспечить две ключевые вещи: доступ по сети и изоляцию аутентификации для каждого запроса.
В этой статье мы подробно рассмотрим архитектурные изменения, которые позволили нашему серверу на fastMCP перейти в полноценный многопользовательский режим через протоколы HTTP/SSE.
Архитектура: От одного клиента к изолированным контекстам
Основная проблема с которой мы столкнулись - нет четкой рекомендации для fastMCP как разделять пользователей и как их авторизовывать. Да, во фреймворке есть встроенная авторизация, но она не распространяется на инструменты, которые используют LLM. Мы решили, что будем использовать стандартные HTTP-заголовки, как мы это делаем в нашем API.
Работа в режиме STDIO:
HTTP/SSE режим:
В сетевом режиме каждый запрос инициирует создание отдельного клиента SDK.
🛠️ Шаг 1: Адаптация инструментов для контекста запроса
Все наши инструменты (tools) используют PingeraSDK. Ранее они напрямую обращались к глобально созданному PingeraClient. Чтобы обеспечить возможность подмены клиента, мы добавили в базовый класс инструментов метод get_client().
Этот метод позволяет переопределить дефолтный клиент (предоставленный при инициализации сервера) экземпляром, созданным специально для текущего запроса.
class BaseTools:
"""Базовый класс для MCP инструментов с общей функциональностью."""
def __init__(self, client: PingeraSDKClient):
self.client = client
self.logger = logging.getLogger(self.__class__.__name__)
def get_client(self, client_override: Optional[PingeraSDKClient] = None) -> PingeraSDKClient:
"""
Получить клиент для API вызовов.
Args:
client_override: Опциональный экземпляр клиента вместо дефолтного
Returns:
Клиент для использования (override если предоставлен, иначе дефолтный)
"""
return client_override if client_override is not None else self.clientТеперь каждый инструмент обращается к API через self.get_client(), что позволяет сохранить совместимость со STDIO и внедрить динамическую авторизацию в HTTP-режиме.
🔑 Шаг 2: Извлечение учетных данных из HTTP-заголовков
Мы разработали утилиту extract_auth_from_request(), которая извлекает аутентификационные данные из заголовка Authorization текущего HTTP-запроса.
def extract_auth_from_request() -> Optional[Tuple[Optional[str], Optional[str]]]:
"""
Извлечь авторизационные данные (API Key или JWT Token) из HTTP заголовков.
Returns:
Кортеж (api_key, jwt_token) или None. Только один элемент будет заполнен.
"""
try:
request = get_http_request()
auth_header = request.headers.get('Authorization')
if not auth_header:
logger.warning("Authorization заголовок не найден")
return None
# Логика парсинга заголовка
auth_type, credentials = get_auth_type(auth_header)
if auth_type == "bearer":
logger.debug("Извлечен Bearer токен из запроса")
return (None, credentials)
elif auth_type == "api_key":
logger.debug("Извлечен API ключ из запроса")
return (credentials, None)
except Exception as e:
logger.debug(f"Не в HTTP контексте: {e}")
return NoneПоддерживаются два формата: Authorization: Bearer JWT_TOKEN и Authorization: API_KEY.
🔄 Шаг 3: Логика выбора клиента
Функция get_request_client() решает, какой именно клиент SDK будет использоваться для обработки запроса.
def get_request_client(
config: Config,
default_client: PingeraSDKClient
) -> PingeraSDKClient:
"""
Получить подходящий клиент для текущего контекста запроса.
В HTTP/SSE режиме: Создает новый клиент из заголовков запроса.
В stdio режиме: Возвращает дефолтный глобальный клиент.
"""
if config.transport_mode == "sse":
auth = extract_auth_from_request()
if auth:
api_key, jwt_token = auth
logger.info(f"Создаем клиент из заголовка авторизации ({'JWT' if jwt_token else 'API Key'})")
# create_client_from_auth создает новый экземпляр PingeraSDKClient
return create_client_from_auth(api_key, jwt_token, config)
elif config.require_auth_header:
logger.warning("Заголовок Authorization обязателен, но не найден")
# Для stdio режима или если авторизация не обязательна - используем дефолтный
return default_clientВ режиме SSE при наличии заголовка авторизации создается новый, изолированный клиент.
🔌 Шаг 4: Обновление контекста перед выполнением
Чтобы гарантировать, что каждый вызов инструмента выполняется в контексте конкретного пользователя, мы реализовали функцию _update_tools_client(). Она вызывает get_request_client() и, если получен новый клиент, подменяет им клиент в каждом экземпляре инструмента.
def _update_tools_client():
"""
Обновляет PingeraClient для каждого экземпляра инструмента.
В SSE режиме это гарантирует, что клиент соответствует текущему контексту запроса.
"""
global status_tools, pages_tools, component_tools, checks_tools
# ... и другие инструменты
if config.transport_mode != "stdio":
request_client = get_request_client(config, pingera_client)
if request_client:
# Обновляем атрибут 'client' в каждом инструменте
status_tools.client = request_client
pages_tools.client = request_client
component_tools.client = request_client
checks_tools.client = request_client
# ... обновляем все инструментыЭту функцию мы вызываем в начале тела каждого инструмента:
@mcp.tool()
async def list_pages(
page: Optional[int] = None,
per_page: Optional[int] = None,
status: Optional[str] = None
) -> str:
"""Получить список всех status pages."""
_update_tools_client() # ← Обновляем клиент из контекста
return await pages_tools.list_pages(page, per_page, status)Это ключевой момент, обеспечивающий изоляцию данных пользователей.
⚙️ Шаг 5 и 6: Конфигурация и запуск транспорта
Конфигурация сервера была расширена для поддержки сетевого режима с необходимыми параметрами:
class Config:
def __init__(self) -> None:
# Конфигурация транспорта
self.transport_mode: str = os.getenv("TRANSPORT_MODE", "stdio").lower()
self.http_host: str = os.getenv("HTTP_HOST", "0.0.0.0")
self.http_port: int = int(os.getenv("HTTP_PORT", "5000"))
# Флаг для принудительной авторизации
self.require_auth_header: bool = os.getenv("REQUIRE_AUTH_HEADER", "false").lower() == "true"При запуске сервера мы проверяем режим и инициализируем соответствующий транспорт:
if __name__ == "__main__":
if config.transport_mode == "sse":
logger.info(f"Запуск MCP сервера в SSE режиме на {config.http_host}:{config.http_port}")
mcp.run(transport="streamable-http") # Запуск HTTP-сервера с SSE
else:
logger.info("Запуск MCP сервера в stdio режиме")
mcp.run()Пример использования клиента
В git репозитории вы найдете файл mcp_client_lg.py - пример MCP клиента с LangGraph, который использует наш сервер в SSE-режиме.
client = MultiServerMCPClient(
{
"pingera": {
"url": SERVER_URL,
"transport": "streamable_http",
"headers": {
"Authorization": PINGERA_API_KEY,
},
}
}
)Итоги решения
Мы успешно решили задачу по переводу нашего MCP сервера в многопользовательский режим с сетевым доступом:
✅ Реализована поддержка двух режимов: stdio и HTTP/SSE.
✅ Обеспечена динамическая авторизация и изоляция клиентов для каждого запроса.
✅ Сохранена обратная совместимость с локальным режимом.
✅ Достигнута чистота кода благодаря единому коду инструментов.
✅ Обеспечена динамическая авторизация и изоляция клиентов для каждого запроса.
✅ Сохранена обратная совместимость с локальным режимом.
✅ Достигнута чистота кода благодаря единому коду инструментов.
Весь код MCP сервера вы можете найти в нашем репозитории: pingera/pingera-mcp