lzwcai-mcp-server-package/lzwcai_mcpskills_template/lzwcai_mcpskills_template/utils/api_client.py

"""
API 调用客户端
通用的 HTTP API 客户端封装
"""

import httpx
import json
from typing import Dict, Any, Optional

try:
    from .env_config import get_base_url, get_api_key
    from .logger_config import get_logger
except ImportError:
    from env_config import get_base_url, get_api_key
    from logger_config import get_logger

logger = get_logger(__name__)

# 默认超时配置（秒）
DEFAULT_TIMEOUT = 30.0
LONG_TIMEOUT = 300.0


class APIClient:
    """通用 API 客户端"""
    
    def __init__(
        self, 
        base_url: Optional[str] = None,
        api_key: Optional[str] = None,
        default_timeout: float = DEFAULT_TIMEOUT
    ):
        """
        初始化 API 客户端
        
        Args:
            base_url: API 基础 URL（默认从环境变量读取）
            api_key: API 密钥（默认从环境变量读取）
            default_timeout: 默认超时时间（秒）
        """
        self.base_url = (base_url or get_base_url()).rstrip('/')
        self.api_key = api_key or get_api_key()
        self.default_timeout = default_timeout
        self._client: Optional[httpx.Client] = None
        
        logger.info(f"[客户端初始化] base_url={self.base_url}")
    
    @property
    def client(self) -> httpx.Client:
        """懒加载 HTTP 客户端"""
        if self._client is None:
            self._client = httpx.Client(timeout=self.default_timeout)
        return self._client
        
    def _get_headers(self) -> Dict[str, str]:
        """获取请求头"""
        headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        }
        if self.api_key:
            headers['X-API-Key'] = self.api_key
        return headers
    
    def request(
        self, 
        endpoint: str, 
        method: str = "GET",
        data: Optional[Dict[str, Any]] = None,
        params: Optional[Dict[str, Any]] = None,
        timeout: Optional[float] = None
    ) -> Dict[str, Any]:
        """
        发送 HTTP 请求
        
        Args:
            endpoint: API 端点路径
            method: HTTP 方法
            data: 请求体数据
            params: URL 查询参数
            timeout: 超时时间
            
        Returns:
            API 响应数据
        """
        url = f"{self.base_url}{endpoint}"
        timeout = timeout or self.default_timeout
        
        try:
            logger.info(f"[API请求] {method} {url}")
            
            if method.upper() == "GET":
                response = self.client.get(
                    url,
                    headers=self._get_headers(),
                    params=params,
                    timeout=timeout
                )
            elif method.upper() == "POST":
                response = self.client.post(
                    url,
                    headers=self._get_headers(),
                    json=data,
                    params=params,
                    timeout=timeout
                )
            elif method.upper() == "PUT":
                response = self.client.put(
                    url,
                    headers=self._get_headers(),
                    json=data,
                    params=params,
                    timeout=timeout
                )
            elif method.upper() == "DELETE":
                response = self.client.delete(
                    url,
                    headers=self._get_headers(),
                    params=params,
                    timeout=timeout
                )
            else:
                raise ValueError(f"不支持的 HTTP 方法: {method}")
            
            logger.info(f"[API响应] HTTP {response.status_code}")
            response.raise_for_status()
            
            return response.json()
            
        except httpx.TimeoutException:
            error_msg = f"API 请求超时: {url}"
            logger.error(f"[API错误] {error_msg}")
            raise Exception(error_msg)
            
        except httpx.HTTPStatusError as e:
            error_msg = f"API 请求失败 (HTTP {e.response.status_code}): {url}"
            logger.error(f"[API错误] {error_msg}")
            raise Exception(error_msg)
            
        except Exception as e:
            error_msg = f"API 请求异常: {str(e)}"
            logger.error(f"[API错误] {error_msg}", exc_info=True)
            raise Exception(error_msg)
    
    def close(self):
        """关闭 HTTP 客户端"""
        if self._client is not None:
            self._client.close()
            self._client = None
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()
        return False


# 懒加载的默认客户端
_default_client: Optional[APIClient] = None


def get_default_client() -> APIClient:
    """获取默认客户端（懒加载）"""
    global _default_client
    if _default_client is None:
        _default_client = APIClient()
    return _default_client


def call_api(
    endpoint: str,
    method: str = "GET",
    data: Optional[Dict[str, Any]] = None,
    params: Optional[Dict[str, Any]] = None,
    timeout: Optional[float] = None
) -> Dict[str, Any]:
    """
    便捷函数：调用 API
    
    Args:
        endpoint: API 端点路径
        method: HTTP 方法
        data: 请求体数据
        params: URL 查询参数
        timeout: 超时时间
        
    Returns:
        API 响应数据
    """
    return get_default_client().request(
        endpoint=endpoint,
        method=method,
        data=data,
        params=params,
        timeout=timeout
    )