http_server.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
HTTP服务器模块 / HTTP Server Module
处理JSON-RPC HTTP接口和请求处理 / Handle JSON-RPC HTTP interface and request processing
"""

import http.server
import socketserver
import json
import subprocess
import os
import sys
import uuid
import traceback
import urllib.parse
from typing import Dict, Any, List
import time

# 导入OpenTelemetry集成 / Import OpenTelemetry integration
from opentelemetry_integration import OpenTelemetryManager
from src.utils.logging_utils import (
    capture_debug_logs,
    get_logger,
    print_to_logger,
    redirect_stdio_to_logger,
)


class JSONToolHandler(http.server.BaseHTTPRequestHandler):
    """处理JSON工具请求的HTTP处理器 / HTTP handler for JSON tool requests"""
    
    def __init__(self, *args, **kwargs):
        self.agent = kwargs.pop('agent', None)
        super().__init__(*args, **kwargs)
    
    def _get_request_language(self) -> str:
        """从请求头获取语言设置 / Get language setting from request headers"""
        accept_language = self.headers.get('Accept-Language', '')
        if 'zh' in accept_language:
            return 'zh'
        elif 'en' in accept_language:
            return 'en'
        else:
            return self.server.agent.current_language
    
    def _validate_request_params(self, tool_name: str, params: Dict[str, Any]):
        """验证请求参数 / Validate request parameters"""
        # 获取注册的工具信息 / Get registered tool information
        registered_tools = self.server.agent.tool_registry.get_registered_tools()
        if tool_name not in registered_tools:
            self.send_error(404, self.server.agent.get_text('tool_not_found', tool_name))
            return
        
        # 特定工具的参数验证 / Parameter validation for specific tools
        if tool_name == 'west_flash':
            if 'build_dir' not in params:
                self.send_error(400, self.server.agent.get_text('parameter_required', 'west_flash', 'build_dir'))
                return
        elif tool_name == 'west_update':
            if 'project_dir' not in params:
                self.send_error(400, self.server.agent.get_text('parameter_required', 'west_update', 'project_dir'))
                return
        elif tool_name == 'test_git_connection':
            if 'url' not in params:
                self.send_error(400, self.server.agent.get_text('missing_required_param', 'url'))
                return
            
            # 验证URL格式 / Validate URL format
            import re
            url_pattern = re.compile(r'^(https?|git)://[^\s/$.?#].[^\s]*$')
            if not url_pattern.match(params['url']):
                self.send_error(400, self.server.agent.get_text('invalid_param_format', 'URL'))
                return
        
        self.server.agent.logger.info(self.server.agent.get_text('tool_params_valid', tool_name))
    
    def _handle_tool_request(self, trace_id: str, span=None):
        """处理工具执行请求 / Handle tool execution request"""
        content_length = int(self.headers.get('Content-Length', 0))
        if content_length == 0:
            self.send_error(400, self.server.agent.get_text('missing_request_body'))
            if span:
                span.set_attribute("http.status_code", 400)
                span.set_attribute("error", True)
                span.set_attribute("error.message", "Missing request body")
            return
        
        post_data = self.rfile.read(content_length)
        
        debug: List[str] = []

        def _mask_params(d: Dict[str, Any]) -> Dict[str, Any]:
            masked: Dict[str, Any] = {}
            for k, v in d.items():
                key = str(k).lower()
                if any(s in key for s in ("password", "passwd", "token", "secret", "apikey", "api_key", "pat", "private_key")):
                    masked[k] = "<redacted>"
                else:
                    masked[k] = v
            return masked

        try:
            # 解析JSON请求 / Parse JSON request
            request = json.loads(post_data.decode('utf-8'))
            
            # 获取工具名称和参数 / Get tool name and parameters
            tool_name = request.get('tool')
            params = request.get('params', {})
            
            if span:
                span.set_attribute("tool.name", tool_name)
            
            if not tool_name:
                self.send_error(400, self.server.agent.get_text('missing_tool_name'))
                if span:
                    span.set_attribute("http.status_code", 400)
                    span.set_attribute("error", True)
                    span.set_attribute("error.message", "Missing tool name")
                return
            
            # 获取注册的工具 / Get registered tools
            registered_tools = self.server.agent.tool_registry.get_registered_tools()
            if tool_name not in registered_tools:
                self.send_error(404, self.server.agent.get_text('tool_not_found', tool_name))
                if span:
                    span.set_attribute("http.status_code", 404)
                    span.set_attribute("error", True)
                    span.set_attribute("error.message", f"Tool not found: {tool_name}")
                return
            
            # 执行参数验证 / Execute parameter validation
            self._validate_request_params(tool_name, params)
            
            # 执行工具 / Execute tool
            tool_info = registered_tools[tool_name]
            tool_func = tool_info['function']
            
            # 记录工具调用，包含trace_id / Log tool call, including trace_id
            self.server.agent.logger.info(f"[{trace_id}] 执行工具: {tool_name}，参数: {params}")

            debug.append(f"INFO http_server: Invoking tool {tool_name}")
            debug.append(f"INFO http_server: Params: {_mask_params(params)}")
            started = time.time()
            
            # 导入必要的模块并注入到工具函数的全局命名空间 / Import necessary modules and inject into tool function's global namespace
            tool_func_globals = tool_func.__globals__
            tool_func_globals['subprocess'] = subprocess
            tool_func_globals['sys'] = sys
            tool_func_globals['os'] = __import__('os')
            tool_func_globals['importlib'] = __import__('importlib')
            tool_func_globals['traceback'] = traceback
            
            # 执行工具函数 / Execute tool function
            with capture_debug_logs(debug):
                import builtins

                tool_logger = get_logger(f"tools.{tool_name}")
                old_print = builtins.print
                strict_stdio = os.getenv("ZEPHYR_MCP_STRICT_STDIO", "1") != "0"
                debug.append(f"INFO http_server: Strict stdio={strict_stdio}")

                def tool_print(*args, **kwargs):
                    return print_to_logger(tool_logger, *args, **kwargs)

                builtins.print = tool_print
                try:
                    with redirect_stdio_to_logger(tool_logger, strict=strict_stdio):
                        result = tool_func(**params)
                finally:
                    builtins.print = old_print

            debug.append(f"INFO http_server: Finished in {round(time.time() - started, 3)}s")

            # Ensure debug is visible to caller (and merge if tool already returns debug)
            if isinstance(result, dict):
                existing = result.get("debug")
                if isinstance(existing, list):
                    merged: List[Any] = []
                    seen: set[str] = set()
                    for item in existing + debug:
                        key = str(item)
                        if key in seen:
                            continue
                        seen.add(key)
                        merged.append(item)
                    result["debug"] = merged
                else:
                    result["debug"] = debug
            else:
                result = {"status": "success", "result": result, "debug": debug}
            
            # 构造响应 / Construct response
            response = {
                "success": True,
                "result": result,
                "tool": tool_name,
                "trace_id": trace_id
            }
            
            # 发送响应 / Send response
            self.send_response(200)
            self.send_header('Content-Type', 'application/json')
            self.end_headers()
            self.wfile.write(json.dumps(response).encode('utf-8'))
            
        except json.JSONDecodeError:
            # 添加trace_id到错误响应 / Add trace_id to error response
            self.send_response(400)
            self.send_header('Content-Type', 'application/json')
            self.send_header('X-Trace-ID', trace_id)
            self.end_headers()
            error_response = {
                "error": self.server.agent.get_text('invalid_json'),
                "trace_id": trace_id
            }
            self.wfile.write(json.dumps(error_response).encode('utf-8'))
        except Exception as e:  # noqa: BLE001
            # 发送错误响应 / Send error response
            error_response = {
                "success": False,
                "error": str(e),
                "tool": tool_name if 'tool_name' in locals() else None,
                "trace_id": trace_id,
                "error_code": "TOOL_EXECUTION_ERROR",
                "debug": debug,
            }
            self.send_response(500)
            self.send_header('Content-Type', 'application/json')
            self.send_header('X-Trace-ID', trace_id)
            self.end_headers()
            self.wfile.write(json.dumps(error_response).encode('utf-8'))
            
            if span:
                span.set_attribute("http.status_code", 500)
                span.set_attribute("error", True)
                span.set_attribute("error.message", str(e))
    
    def _handle_api_tools_request(self, trace_id: str, span=None):
        """处理/api/tools端点请求 / Handle /api/tools endpoint request"""
        _ = trace_id
        registered_tools = self.server.agent.tool_registry.get_registered_tools()
        
        # 构建工具信息列表 / Build tool information list
        tools_info = []
        for tool_name, tool_info in registered_tools.items():
            tools_info.append({
                "name": tool_name,
                "description": str(tool_info.get('description', '')),
                "parameters": tool_info.get('parameters', []),
                "returns": tool_info.get('returns', []),
                "module": str(tool_info.get('module', ''))
            })
        
        response = {
            "tools": tools_info,
            "total": len(tools_info),
            "llm_integration": self.server.agent.config.get("llm", {}).get("enabled", False)
        }
        
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        self.wfile.write(json.dumps(response).encode('utf-8'))
        
        if span:
            span.set_attribute("http.status_code", 200)
            span.set_attribute("returned_tools_count", len(tools_info))
    
    def _handle_api_docs_request(self, trace_id: str, span=None):
        """处理/api/docs端点请求 / Handle /api/docs endpoint request"""
        _ = trace_id
        host = self.server.server_address[0]
        port = self.server.server_address[1]
        
        # 获取当前请求的语言 / Get current request language
        current_language = self._get_request_language()
        
        # 基础API端点 / Basic API endpoints
        endpoints = [
            {
                "url": "/api/tools",
                "method": "GET",
                "description": self.server.agent.get_text('api_docs_get_tools'),
                "response_format": {
                    "tools": self.server.agent.get_text('api_docs_tools_list'),
                    "total": self.server.agent.get_text('api_docs_total_tools'),
                    "llm_integration": self.server.agent.get_text('api_docs_llm_integration')
                },
                "example": f"curl -X GET http://{host}:{port}/api/tools"
            },
            {
                "url": "/api/tool/info",
                "method": "GET",
                "description": self.server.agent.get_text('api_docs_get_tool_info'),
                "parameters": [
                    {"name": "name", "type": "query", "description": self.server.agent.get_text('api_docs_tool_name'), "required": True}
                ],
                "response_format": {
                    "name": self.server.agent.get_text('api_docs_tool_name'),
                    "description": self.server.agent.get_text('api_docs_tool_description'),
                    "parameters": self.server.agent.get_text('api_docs_parameters_list'),
                    "returns": self.server.agent.get_text('api_docs_returns_list'),
                    "module": self.server.agent.get_text('api_docs_tool_module')
                },
                "example": f"curl -X GET http://{host}:{port}/api/tool/info?name=test_git_connection"
            },
            {
                "url": "/api/tool",
                "method": "POST",
                "description": self.server.agent.get_text('api_docs_execute_tool'),
                "request_format": {
                    "tool": self.server.agent.get_text('api_docs_tool_name'),
                    "params": self.server.agent.get_text('api_docs_tool_params')
                },
                "response_format": {
                    "success": self.server.agent.get_text('api_docs_success'),
                    "result": self.server.agent.get_text('api_docs_execution_result'),
                    "error": self.server.agent.get_text('api_docs_error_info'),
                    "tool": self.server.agent.get_text('api_docs_called_tool')
                },
                "example": f"curl -X POST http://{host}:{port}/api/tool -H 'Content-Type: application/json' -d '{{\"tool\":\"test_git_connection\",\"params\":{{\"url\":\"https://github.com/zephyrproject-rtos/zephyr\"}}}}'"
            }
        ]
        
        # 如果启用了LLM集成，添加AI助手端点 / If LLM integration is enabled, add AI assistant endpoint
        if self.server.agent.config.get("llm", {}).get("enabled", False):
            endpoints.append({
                "url": "/api/ai_assistant",
                "method": "POST",
                "description": self.server.agent.get_text('api_docs_ai_assistant'),
                "request_format": {
                    "messages": self.server.agent.get_text('api_docs_messages_list'),
                    "model": self.server.agent.get_text('api_docs_model_name'),
                    "temperature": self.server.agent.get_text('api_docs_temperature'),
                    "max_tokens": self.server.agent.get_text('api_docs_max_tokens')
                },
                "response_format": {
                    "success": self.server.agent.get_text('api_docs_success'),
                    "response": self.server.agent.get_text('api_docs_ai_response'),
                    "model": self.server.agent.get_text('api_docs_used_model'),
                    "usage": self.server.agent.get_text('api_docs_token_usage')
                },
                "example": f"curl -X POST http://{host}:{port}/api/ai_assistant -H 'Content-Type: application/json' -d '{{\"messages\":[{{\"role\":\"user\",\"content\":\"你好\"}}]}}'"
            })
        
        response = {
            "endpoints": endpoints,
            "version": self.server.agent.config.get("version", "1.0.0"),
            "agent_name": self.server.agent.config.get("agent_name", "Zephyr MCP Agent"),
            "language": current_language,
            "supported_languages": ["zh", "en"]
        }
        
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        self.wfile.write(json.dumps(response).encode('utf-8'))
        
        if span:
            span.set_attribute("http.status_code", 200)
            span.set_attribute("response_language", current_language)
    
    def do_POST(self):
        """处理POST请求 / Handle POST request"""
        # 生成或获取trace_id / Generate or get trace_id
        trace_id = self.headers.get('X-Trace-ID', str(uuid.uuid4()))
        
        # 检查是否启用OpenTelemetry / Check if OpenTelemetry is enabled
        otel_manager = OpenTelemetryManager(self.server.agent.config, self.server.agent.logger)
        
        if otel_manager.is_enabled():
            # 使用追踪的版本 / Version with tracing
            span = otel_manager.create_span("HTTP_POST", {
                "http.method": "POST",
                "http.url": self.path,
                "trace_id": trace_id
            })
            
            try:
                if self.path == "/api/tool":
                    self._handle_tool_request(trace_id, span)
                elif self.path == "/api/ai_assistant":
                    # AI助手端点处理（简化版本） / AI assistant endpoint handling (simplified version)
                    self.send_response(501)
                    self.send_header('Content-Type', 'application/json')
                    self.end_headers()
                    self.wfile.write(json.dumps({
                        "error": "AI Assistant endpoint not implemented in this module",
                        "trace_id": trace_id
                    }).encode('utf-8'))
                    span.set_attribute("http.status_code", 501)
                else:
                    # 未找到路径，返回404 / Path not found, return 404
                    self.send_response(404)
                    self.send_header('Content-Type', 'application/json')
                    self.send_header('X-Trace-ID', trace_id)
                    self.end_headers()
                    error_response = {
                        "error": "Not Found",
                        "path": self.path,
                        "trace_id": trace_id
                    }
                    self.wfile.write(json.dumps(error_response).encode('utf-8'))
                    span.set_attribute("http.status_code", 404)
                    span.set_attribute("error", True)
                    span.set_attribute("error.message", f"Path not found: {self.path}")
            except Exception as e:  # noqa: BLE001
                self.server.agent.logger.error(f"[{trace_id}] POST请求处理错误: {str(e)}")
                span.set_attribute("error", True)
                span.set_attribute("error.message", str(e))
            finally:
                otel_manager.end_span(span)
        else:
            # 不使用追踪的版本 / Version without tracing
            if self.path == "/api/tool":
                self._handle_tool_request(trace_id)
            elif self.path == "/api/ai_assistant":
                self.send_response(501)
                self.send_header('Content-Type', 'application/json')
                self.end_headers()
                self.wfile.write(json.dumps({
                    "error": "AI Assistant endpoint not implemented in this module",
                    "trace_id": trace_id
                }).encode('utf-8'))
            else:
                self.send_response(404)
                self.send_header('Content-Type', 'application/json')
                self.send_header('X-Trace-ID', trace_id)
                self.end_headers()
                error_response = {
                    "error": "Not Found",
                    "path": self.path,
                    "trace_id": trace_id
                }
                self.wfile.write(json.dumps(error_response).encode('utf-8'))
    
    def do_GET(self):
        """处理GET请求 / Handle GET request"""
        # 生成或获取trace_id / Generate or get trace_id
        trace_id = self.headers.get('X-Trace-ID', str(uuid.uuid4()))
        
        # 解析路径和查询参数 / Parse path and query parameters
        parsed_path = urllib.parse.urlparse(self.path)
        path = parsed_path.path
        query_components = urllib.parse.parse_qs(parsed_path.query)
        
        # 检查是否启用OpenTelemetry / Check if OpenTelemetry is enabled
        otel_manager = OpenTelemetryManager(self.server.agent.config, self.server.agent.logger)
        
        if otel_manager.is_enabled():
            span = otel_manager.create_span("HTTP_GET", {
                "http.method": "GET",
                "http.url": self.path,
                "trace_id": trace_id
            })
            
            try:
                if path == "/api/tools":
                    span.set_attribute("endpoint", "api_tools")
                    self._handle_api_tools_request(trace_id, span)
                elif path == "/api/docs":
                    span.set_attribute("endpoint", "api_docs")
                    self._handle_api_docs_request(trace_id, span)
                elif path.startswith("/api/tool/info"):
                    # 解析查询参数获取工具名称 / Parse query parameters to get tool name
                    tool_name = query_components.get('name', [None])[0]
                    span.set_attribute("endpoint", "api_tool_info")
                    
                    if not tool_name:
                        self.send_response(400)
                        self.send_header('Content-Type', 'application/json')
                        self.send_header('X-Trace-ID', trace_id)
                        self.end_headers()
                        error_response = {
                            "error": self.server.agent.get_text('missing_tool_name'),
                            "trace_id": trace_id
                        }
                        self.wfile.write(json.dumps(error_response).encode('utf-8'))
                        span.set_attribute("http.status_code", 400)
                        span.set_attribute("error", True)
                        span.set_attribute("error.message", "Missing tool name")
                    else:
                        span.set_attribute("tool.name", tool_name)
                        registered_tools = self.server.agent.tool_registry.get_registered_tools()
                        
                        if tool_name not in registered_tools:
                            self.send_response(404)
                            self.send_header('Content-Type', 'application/json')
                            self.send_header('X-Trace-ID', trace_id)
                            self.end_headers()
                            error_response = {
                                "error": self.server.agent.get_text('tool_not_found', tool_name),
                                "trace_id": trace_id
                            }
                            self.wfile.write(json.dumps(error_response).encode('utf-8'))
                            span.set_attribute("http.status_code", 404)
                            span.set_attribute("error", True)
                            span.set_attribute("error.message", f"Tool not found: {tool_name}")
                        else:
                            tool_info = registered_tools[tool_name]
                            response = {
                                "name": tool_name,
                                "description": tool_info.get('description', ''),
                                "parameters": tool_info.get('parameters', []),
                                "returns": tool_info.get('returns', []),
                                "module": tool_info.get('module', '')
                            }
                            self.send_response(200)
                            self.send_header('Content-Type', 'application/json')
                            self.end_headers()
                            self.wfile.write(json.dumps(response).encode('utf-8'))
                            span.set_attribute("http.status_code", 200)
                else:
                    self.send_response(404)
                    self.send_header('Content-Type', 'application/json')
                    self.send_header('X-Trace-ID', trace_id)
                    self.end_headers()
                    error_response = {
                        "error": "Not Found",
                        "path": path,
                        "trace_id": trace_id
                    }
                    self.wfile.write(json.dumps(error_response).encode('utf-8'))
                    span.set_attribute("http.status_code", 404)
                    span.set_attribute("error", True)
                    span.set_attribute("error.message", f"Path not found: {path}")
            except Exception as e:  # noqa: BLE001
                self.server.agent.logger.error(f"[{trace_id}] GET请求处理错误: {str(e)}")
                span.set_attribute("error", True)
                span.set_attribute("error.message", str(e))
            finally:
                otel_manager.end_span(span)
        else:
            # 不使用追踪的版本 / Version without tracing
            if path == "/api/tools":
                self._handle_api_tools_request(trace_id)
            elif path == "/api/docs":
                self._handle_api_docs_request(trace_id)
            elif path.startswith("/api/tool/info"):
                tool_name = query_components.get('name', [None])[0]
                
                if not tool_name:
                    self.send_response(400)
                    self.send_header('Content-Type', 'application/json')
                    self.send_header('X-Trace-ID', trace_id)
                    self.end_headers()
                    error_response = {
                        "error": self.server.agent.get_text('missing_tool_name'),
                        "trace_id": trace_id
                    }
                    self.wfile.write(json.dumps(error_response).encode('utf-8'))
                else:
                    registered_tools = self.server.agent.tool_registry.get_registered_tools()
                    
                    if tool_name not in registered_tools:
                        self.send_response(404)
                        self.send_header('Content-Type', 'application/json')
                        self.send_header('X-Trace-ID', trace_id)
                        self.end_headers()
                        error_response = {
                            "error": self.server.agent.get_text('tool_not_found', tool_name),
                            "trace_id": trace_id
                        }
                        self.wfile.write(json.dumps(error_response).encode('utf-8'))
                    else:
                        tool_info = registered_tools[tool_name]
                        response = {
                            "name": tool_name,
                            "description": tool_info.get('description', ''),
                            "parameters": tool_info.get('parameters', []),
                            "returns": tool_info.get('returns', []),
                            "module": tool_info.get('module', '')
                        }
                        self.send_response(200)
                        self.send_header('Content-Type', 'application/json')
                        self.end_headers()
                        self.wfile.write(json.dumps(response).encode('utf-8'))
            else:
                self.send_response(404)
                self.send_header('Content-Type', 'application/json')
                self.send_header('X-Trace-ID', trace_id)
                self.end_headers()
                error_response = {
                    "error": "Not Found",
                    "path": path,
                    "trace_id": trace_id
                }
                self.wfile.write(json.dumps(error_response).encode('utf-8'))

    def log_message(self, format_str, *args):
        """自定义日志消息格式 / Custom log message format"""
        self.server.agent.logger.info(
            f"[{self.headers.get('X-Trace-ID', 'unknown')}] {format_str % args}"
        )


def start_json_server(agent):
    """启动JSON HTTP接口服务器 / Start JSON HTTP interface server"""
    # 获取配置中的端口，默认为8001 / Get port from config, default to 8001
    port = agent.config.get("port", 8001)
    host = agent.config.get("host", "localhost")
    
    # 创建自定义的HTTP服务器类 / Create custom HTTP server class
    class JSONHTTPServer(socketserver.TCPServer):
        def __init__(self, server_address, handler_class):
            super().__init__(server_address, handler_class)
            self.agent = agent
    
    # 创建处理器类工厂 / Create handler class factory
    def handler_factory(*args, **kwargs):
        return JSONToolHandler(*args, agent=agent, **kwargs)
    
    # 启动服务器 / Start server
    with JSONHTTPServer((host, port), handler_factory) as httpd:
        agent.logger.info(agent.get_text('server_started_full', f"{host}:{port}"))
        agent.logger.info(agent.get_text('available_endpoints'))
        agent.logger.info(f"  GET  http://{host}:{port}/api/tools - {agent.get_text('endpoint_get_tools')}")
        agent.logger.info(f"  GET  http://{host}:{port}/api/tool/info?name=<tool_name> - {agent.get_text('endpoint_get_tool_info')}")
        agent.logger.info(f"  POST http://{host}:{port}/api/tool - {agent.get_text('endpoint_execute_tool')}")
        agent.logger.info(f"  GET  http://{host}:{port}/api/docs - {agent.get_text('endpoint_get_docs')}")
        
        if agent.config.get("llm", {}).get("enabled", False):
            agent.logger.info(f"  POST http://{host}:{port}/api/ai_assistant - {agent.get_text('endpoint_ai_assistant')}")
        
        try:
            httpd.serve_forever()
        except KeyboardInterrupt:
            agent.logger.info(agent.get_text('server_interrupted'))
        except Exception as e:  # noqa: BLE001
            agent.logger.error(agent.get_text('server_error', str(e)))
        finally:
            httpd.server_close()
            agent.logger.info(agent.get_text('server_closed'))