FastMCP и Multi-tenancy: Переход от локального STDIO к сетевому HTTP/SSE

Архитектура: От одного клиента к изолированным контекстам

🛠️ Шаг 1: Адаптация инструментов для контекста запроса

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

🔑 Шаг 2: Извлечение учетных данных из 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

🔄 Шаг 3: Логика выбора клиента

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

🔌 Шаг 4: Обновление контекста перед выполнением

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()

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

client = MultiServerMCPClient(
        {
            "pingera": {
                "url": SERVER_URL,
                "transport": "streamable_http",
                "headers": {
                    "Authorization": PINGERA_API_KEY,
                },
            }
        }
    )

Итоги решения