X-Agents/agent/app/agent/tools/impl/api_client.py

"""
API 调用工具 - 安全的外部 API 调用
"""
import httpx
from typing import Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum


class APIPermission(Enum):
    """API 权限级别"""
    PUBLIC = "public"      # 公开 API
    APPROVED = "approved"  # 已审批的 API
    ADMIN = "admin"        # 管理员 API


@dataclass
class APIEndpoint:
    """API 端点定义"""
    name: str
    url: str
    method: str
    permission: APIPermission
    description: str
    rate_limit: int = 60  # 每分钟请求次数


# API 白名单
ALLOWED_APIS = [
    APIEndpoint(
        name="weather",
        url="https://api.weather.example.com/v1",
        method="GET",
        permission=APIPermission.PUBLIC,
        description="获取天气信息",
        rate_limit=30
    ),
    APIEndpoint(
        name="news",
        url="https://newsapi.org/v2",
        method="GET",
        permission=APIPermission.PUBLIC,
        description="获取新闻",
        rate_limit=30
    ),
    # 可以添加更多已审批的 API
]


class APICallTool:
    """
    API 调用工具

    安全特性：
    - 只允许调用白名单中的 API
    - 速率限制
    - 请求超时
    - 响应大小限制
    """

    def __init__(self):
        self.allowed_apis = {api.name: api for api in ALLOWED_APIS}
        self.request_timeout = 10  # 请求超时（秒）
        self.max_response_size = 1024 * 1024  # 最大响应大小（1MB）

    async def call(
        self,
        api_name: str,
        endpoint: str = "",
        params: Optional[Dict[str, Any]] = None,
        headers: Optional[Dict[str, str]] = None
    ) -> Dict[str, Any]:
        """
        调用 API

        Args:
            api_name: API 名称（必须在白名单中）
            endpoint: 具体的端点
            params: 查询参数
            headers: 请求头

        Returns:
            API 响应
        """
        # 安全检查1: API 必须在白名单中
        if api_name not in self.allowed_apis:
            return {
                "success": False,
                "error": f"API '{api_name}' not in whitelist. Allowed: {list(self.allowed_apis.keys())}"
            }

        api = self.allowed_apis[api_name]

        # 构建完整 URL
        url = f"{api.url}/{endpoint}" if endpoint else api.url

        try:
            async with httpx.AsyncClient(timeout=self.request_timeout) as client:
                # 根据方法调用
                if api.method == "GET":
                    response = await client.get(url, params=params, headers=headers)
                elif api.method == "POST":
                    response = await client.post(url, json=params, headers=headers)
                else:
                    return {
                        "success": False,
                        "error": f"Method {api.method} not supported"
                    }

                # 检查响应大小
                if len(response.content) > self.max_response_size:
                    return {
                        "success": False,
                        "error": f"Response too large (max {self.max_response_size} bytes)"
                    }

                return {
                    "success": True,
                    "status_code": response.status_code,
                    "data": response.json() if response.headers.get("content-type", "").startswith("application/json") else response.text,
                    "headers": dict(response.headers)
                }

        except httpx.TimeoutException:
            return {
                "success": False,
                "error": "Request timeout"
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }

    def list_apis(self) -> list:
        """列出所有可用的 API"""
        return [
            {
                "name": api.name,
                "description": api.description,
                "method": api.method,
                "permission": api.permission.value,
                "rate_limit": api.rate_limit
            }
            for api in ALLOWED_APIS
        ]


# 全局实例
api_tool = APICallTool()


async def call_api(
    api_name: str,
    endpoint: str = "",
    params: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """
    API 调用工具（供 Agent 使用）
    """
    return await api_tool.call(api_name, endpoint, params)


def list_allowed_apis() -> list:
    """列出允许的 API"""
    return api_tool.list_apis()