ai-agent-admin/backend-fastapi/ai_platform/nodes/builtin/http_node.py

"""
HTTP 请求节点
"""
import logging
import time
from typing import Any, Dict

from ..base import BaseNode, NodeContext, NodeResult
from ..registry import NodeRegistry

logger = logging.getLogger(__name__)


@NodeRegistry.register
class HttpNode(BaseNode):
    """
    HTTP 请求节点
    
    发送 HTTP 请求并获取响应
    """
    
    node_type = 'http'
    node_name = 'HTTP 请求'
    node_category = 'data'
    node_icon = 'globe'
    node_description = '发送 HTTP 请求并获取响应'
    
    inputs = [
        {
            'name': 'url',
            'type': 'string',
            'description': '请求 URL',
        },
        {
            'name': 'body',
            'type': 'object',
            'description': '请求体',
        },
    ]
    
    outputs = [
        {
            'name': 'response',
            'type': 'object',
            'description': '响应数据',
        },
        {
            'name': 'status_code',
            'type': 'number',
            'description': '状态码',
        },
    ]
    
    def execute(self, context: NodeContext) -> NodeResult:
        """执行 HTTP 请求"""
        start_time = time.time()
        
        try:
            import httpx
            
            method = self.config.get('method', 'GET').upper()
            url = self.config.get('url', '')
            headers = self.config.get('headers', {})
            params = self.config.get('params', {})
            body = self.config.get('body', {})
            timeout = self.config.get('timeout', 30)
            output_variable = self.config.get('output_variable', 'http_response')
            
            # 解析模板变量
            url = context.resolve_template(url)
            
            # 解析 headers 中的变量
            resolved_headers = {}
            for key, value in headers.items():
                resolved_headers[key] = context.resolve_template(str(value))
            
            # 解析 params 中的变量
            resolved_params = {}
            for key, value in params.items():
                resolved_params[key] = context.resolve_template(str(value))
            
            # 解析 body 中的变量
            resolved_body = self._resolve_body(body, context)
            
            # 发送请求
            with httpx.Client(timeout=timeout) as client:
                if method in ('GET', 'DELETE'):
                    response = client.request(
                        method=method,
                        url=url,
                        headers=resolved_headers,
                        params=resolved_params,
                    )
                else:
                    response = client.request(
                        method=method,
                        url=url,
                        headers=resolved_headers,
                        params=resolved_params,
                        json=resolved_body,
                    )
            
            # 解析响应
            try:
                response_data = response.json()
            except Exception:
                response_data = response.text
            
            elapsed_time = int((time.time() - start_time) * 1000)
            
            return NodeResult(
                success=response.is_success,
                output=response_data,
                output_variables={
                    output_variable: response_data,
                    f'{output_variable}_status': response.status_code,
                },
                elapsed_time=elapsed_time,
                metadata={
                    'status_code': response.status_code,
                    'headers': dict(response.headers),
                },
            )
            
        except Exception as e:
            logger.exception(f'HTTP 节点执行失败: {e}')
            elapsed_time = int((time.time() - start_time) * 1000)
            return NodeResult(
                success=False,
                error=str(e),
                elapsed_time=elapsed_time,
            )
    
    def _resolve_body(self, body: Any, context: NodeContext) -> Any:
        """递归解析 body 中的变量"""
        if isinstance(body, str):
            return context.resolve_template(body)
        elif isinstance(body, dict):
            return {k: self._resolve_body(v, context) for k, v in body.items()}
        elif isinstance(body, list):
            return [self._resolve_body(item, context) for item in body]
        return body
    
    @classmethod
    def get_config_schema(cls) -> Dict[str, Any]:
        """获取配置 Schema"""
        return {
            'type': 'object',
            'properties': {
                'method': {
                    'type': 'string',
                    'title': '请求方法',
                    'enum': ['GET', 'POST', 'PUT', 'DELETE', 'PATCH'],
                    'default': 'GET',
                },
                'url': {
                    'type': 'string',
                    'title': 'URL',
                    'description': '支持变量引用，如 {{api_url}}',
                },
                'headers': {
                    'type': 'object',
                    'title': '请求头',
                    'additionalProperties': {'type': 'string'},
                },
                'params': {
                    'type': 'object',
                    'title': 'URL 参数',
                    'additionalProperties': {'type': 'string'},
                },
                'body': {
                    'type': 'object',
                    'title': '请求体',
                    'description': 'POST/PUT/PATCH 请求的 JSON 数据',
                },
                'timeout': {
                    'type': 'integer',
                    'title': '超时时间（秒）',
                    'default': 30,
                },
                'output_variable': {
                    'type': 'string',
                    'title': '输出变量名',
                    'default': 'http_response',
                },
            },
            'required': ['url'],
        }