Halo MCP Server

"""Halo API 客户端""" from typing import Any, Dict, List, Optional from loguru import logger from halo_mcp_server.client.base import BaseHTTPClient from halo_mcp_server.config import settings from halo_mcp_server.exceptions import AuthenticationError, ConfigurationError class HaloClient(BaseHTTPClient): """带认证的 Halo API 客户端""" def __init__(self): """初始化 Halo 客户端。""" super().__init__( base_url=settings.halo_base_url, timeout=settings.mcp_timeout, ) self._authenticated = False async def authenticate(self) -> None: """ 与 Halo 服务器进行认证。 异常: ConfigurationError：未配置认证方式 AuthenticationError：认证失败 """ # Try token authentication first if settings.has_token_auth: token = settings.halo_token if token: # Type guard to ensure token is not None self.set_auth_token(token) self._authenticated = True logger.info("使用令牌认证成功") return # Try password authentication if settings.has_password_auth: try: token = await self._login_with_password() self.set_auth_token(token) self._authenticated = True logger.info("使用用户名/密码认证成功") return except Exception as e: raise AuthenticationError(f"密码认证失败：{e}") raise ConfigurationError( "未配置任何认证方式。请设置 HALO_TOKEN 或 HALO_USERNAME/HALO_PASSWORD" ) async def _login_with_password(self) -> str: """ 使用用户名和密码登录。 返回: 访问令牌 异常: AuthenticationError：登录失败 """ try: response = await self.post( "/apis/api.console.halo.run/v1alpha1/auth/login", json={ "username": settings.halo_username, "password": settings.halo_password, }, ) token = response.get("access_token") if not token: raise AuthenticationError("响应中未找到访问令牌") return token except Exception as e: logger.error(f"登录失败：{e}") raise async def ensure_authenticated(self) -> None: """确保客户端已通过认证。""" if not self._authenticated: await self.authenticate()