MuMuAINovel/backend/app/services/mcp_test_service.py

"""MCP插件测试服务 - 专门处理插件测试逻辑

重构后使用统一的MCPClientFacade门面来管理所有MCP操作。
"""

import time
import json
from typing import Dict, Any, Optional
from datetime import datetime

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from app.models.mcp_plugin import MCPPlugin
from app.models.settings import Settings as UserSettings
from app.mcp import mcp_client, MCPPluginConfig  # 使用新的统一门面
from app.services.ai_service import create_user_ai_service
from app.schemas.mcp_plugin import MCPTestResult
from app.services.prompt_service import prompt_service
from app.logger import get_logger
from app.user_manager import User

logger = get_logger(__name__)


class MCPTestService:
    """MCP插件测试服务（使用统一门面重构）"""

    def _check_plugin_registered(self, plugin: MCPPlugin, user_id: str) -> bool:
        """
        检查插件是否已注册（同步方法，不触发新的连接）

        Args:
            plugin: 插件配置
            user_id: 用户ID

        Returns:
            是否已注册
        """
        return mcp_client.is_registered(user_id, plugin.plugin_name)

    async def test_plugin_connection(
        self,
        plugin: MCPPlugin,
        user_id: str
    ) -> MCPTestResult:
        """
        简单连接测试

        注意：调用此方法前，需要确保插件已通过后台任务注册。

        Args:
            plugin: 插件配置
            user_id: 用户ID

        Returns:
            测试结果
        """
        start_time = time.time()

        try:
            # 检查插件是否已注册（不触发新连接）
            if not self._check_plugin_registered(plugin, user_id):
                return MCPTestResult(
                    success=False,
                    message="插件未注册",
                    error="MCP会话不存在，请先启用插件",
                    suggestions=["请先启用插件", "如果已启用，请稍等片刻后重试"]
                )

            # 使用统一门面测试连接
            test_result = await mcp_client.test_connection(user_id, plugin.plugin_name)
            
            end_time = time.time()
            response_time = round((end_time - start_time) * 1000, 2)
            
            if test_result["success"]:
                return MCPTestResult(
                    success=True,
                    message=f"✅ 连接测试成功",
                    response_time_ms=response_time,
                    tools_count=test_result.get("tools_count", 0),
                    suggestions=[
                        f"响应时间: {response_time}ms",
                        f"可用工具数: {test_result.get('tools_count', 0)}"
                    ]
                )
            else:
                return MCPTestResult(
                    success=False,
                    message="❌ 连接测试失败",
                    response_time_ms=response_time,
                    error=test_result.get("message", "未知错误"),
                    error_type=test_result.get("error_type"),
                    suggestions=[
                        "请检查服务器是否在线",
                        "请确认配置正确",
                        "请检查API Key是否有效"
                    ]
                )
                
        except Exception as e:
            end_time = time.time()
            response_time = round((end_time - start_time) * 1000, 2)
            
            logger.error(f"测试插件失败: {plugin.plugin_name}, 错误: {e}")
            
            return MCPTestResult(
                success=False,
                message="❌ 测试失败",
                response_time_ms=response_time,
                error=str(e),
                error_type=type(e).__name__,
                suggestions=[
                    "请检查服务器是否在线",
                    "请确认配置正确",
                    "请检查API Key是否有效"
                ]
            )
    
    async def test_plugin_with_ai(
        self,
        plugin: MCPPlugin,
        user: User,
        db_session: AsyncSession
    ) -> MCPTestResult:
        """
        使用AI进行智能工具调用测试
        
        Args:
            plugin: 插件配置
            user: 用户对象
            db_session: 数据库会话
            
        Returns:
            测试结果
        """
        start_time = time.time()
        
        try:
            # 1. 先进行连接测试
            connection_result = await self.test_plugin_connection(plugin, user.user_id)
            
            if not connection_result.success:
                return connection_result
            
            # 2. 使用统一门面获取工具列表
            tools = await mcp_client.get_tools(user.user_id, plugin.plugin_name)
            
            if not tools:
                return MCPTestResult(
                    success=False,
                    message="插件没有提供任何工具",
                    error="工具列表为空",
                    response_time_ms=connection_result.response_time_ms,
                    suggestions=["请检查插件配置", "请确认MCP服务器正常运行"]
                )
            
            # 3. 获取用户的AI设置
            settings_result = await db_session.execute(
                select(UserSettings).where(UserSettings.user_id == user.user_id)
            )
            user_settings = settings_result.scalar_one_or_none()
            
            if not user_settings or not user_settings.api_key:
                # 没有AI配置，返回简单测试结果
                logger.warning("用户未配置AI服务，跳过智能测试")
                return MCPTestResult(
                    success=True,
                    message=f"✅ 连接测试成功（未配置AI，跳过工具调用测试）",
                    response_time_ms=connection_result.response_time_ms,
                    tools_count=len(tools),
                    suggestions=[
                        f"连接测试: 成功",
                        f"可用工具数: {len(tools)}",
                        "提示: 配置AI服务后可进行智能工具调用测试"
                    ]
                )
            
            # 4. 使用AI选择工具并生成测试参数
            logger.info(f"使用AI分析工具并生成测试计划...")
            
            ai_service = create_user_ai_service(
                api_provider=user_settings.api_provider,
                api_key=user_settings.api_key,
                api_base_url=user_settings.api_base_url,
                model_name=user_settings.llm_model,
                temperature=0.3,
                max_tokens=1000
            )
            
            # 使用统一门面转换为OpenAI Function Calling格式
            openai_tools = mcp_client.format_tools_for_openai(tools, plugin.plugin_name)
            
            logger.info(f"📋 转换后的OpenAI工具数量: {len(openai_tools)}")
            logger.debug(f"📋 OpenAI工具列表: {[t['function']['name'] for t in openai_tools]}")
            
            # 调用AI选择工具（使用自定义模板系统）
            prompts = await prompt_service.get_mcp_tool_test_prompts(
                plugin_name=plugin.plugin_name,
                user_id=user.user_id,
                db=db_session
            )
            
            # 使用 generate_text 进行 Function Calling（非流式）
            ai_response = await ai_service.generate_text(
                prompt=prompts["user"],
                system_prompt=prompts["system"],
                tools=openai_tools,
                tool_choice="auto"
            )
            
            accumulated_text = ai_response.get("content", "")
            tool_calls = ai_response.get("tool_calls")
            
            # 5. 检查AI是否返回工具调用
            if not tool_calls:
                logger.error(f"❌ AI未返回工具调用")
                return MCPTestResult(
                    success=False,
                    message="❌ AI Function Calling失败",
                    error=f"AI未返回工具调用请求。响应: {accumulated_text[:200] if accumulated_text else 'N/A'}",
                    tools_count=len(tools),
                    suggestions=[
                        "请确认使用的AI模型支持Function Calling",
                        f"当前Provider: {user_settings.api_provider}",
                        f"当前模型: {user_settings.llm_model}"
                    ]
                )
            
            # 6. 解析工具调用
            tool_call = tool_calls[0]
            function = tool_call["function"]
            tool_name_with_prefix = function["name"]
            test_arguments = function["arguments"]
            
            if isinstance(test_arguments, str):
                try:
                    # 使用统一的JSON清洗方法
                    cleaned_args = ai_service._clean_json_response(test_arguments)
                    test_arguments = json.loads(cleaned_args)
                except json.JSONDecodeError as e:
                    logger.error(f"❌ 解析AI参数失败: {e}")
                    return MCPTestResult(
                        success=False,
                        message="❌ AI返回的参数格式错误",
                        error=f"无法解析参数JSON: {str(e)}",
                        tools_count=len(tools)
                    )
            
            # 解析插件名和工具名
            try:
                _, tool_name = mcp_client.parse_function_name(tool_name_with_prefix)
            except ValueError:
                tool_name = tool_name_with_prefix
            
            logger.info(f"🤖 AI选择的工具: {tool_name}")
            logger.info(f"📝 AI生成的参数: {test_arguments}")
            
            # 7. 使用统一门面调用MCP工具
            call_start = time.time()
            try:
                tool_result = await mcp_client.call_tool(
                    user_id=user.user_id,
                    plugin_name=plugin.plugin_name,
                    tool_name=tool_name,
                    arguments=test_arguments
                )
                
                call_end = time.time()
                call_time = round((call_end - call_start) * 1000, 2)
                total_time = round((call_end - start_time) * 1000, 2)
                
                # 格式化结果
                result_str = str(tool_result)
                if len(result_str) > 800:
                    result_preview = result_str[:800] + "\n...(结果已截断)"
                else:
                    result_preview = result_str
                
                return MCPTestResult(
                    success=True,
                    message=f"✅ Function Calling测试成功！工具 '{tool_name}' 调用正常",
                    response_time_ms=total_time,
                    tools_count=len(tools),
                    suggestions=[
                        f"🤖 AI选择: {tool_name}",
                        f"📝 参数: {json.dumps(test_arguments, ensure_ascii=False)}",
                        f"⏱️ 耗时: {call_time}ms",
                        f"📊 结果:\n{result_preview}"
                    ]
                )
                
            except Exception as call_error:
                call_end = time.time()
                total_time = round((call_end - start_time) * 1000, 2)
                
                logger.warning(f"工具调用失败: {tool_name}, 错误: {call_error}")
                
                return MCPTestResult(
                    success=True,  # 连接成功就算测试通过
                    message=f"⚠️ 连接成功，但工具调用失败",
                    response_time_ms=total_time,
                    tools_count=len(tools),
                    error=f"工具 '{tool_name}' 调用失败: {str(call_error)}",
                    suggestions=[
                        f"✅ 连接测试: 成功",
                        f"❌ 工具调用测试: 失败",
                        f"🤖 AI选择: {tool_name}",
                        f"❌ 错误: {str(call_error)}",
                        "💡 可能原因: API Key无效、参数错误或服务限制"
                    ]
                )
                
        except Exception as e:
            end_time = time.time()
            total_time = round((end_time - start_time) * 1000, 2)
            
            logger.error(f"测试插件失败: {plugin.plugin_name}, 错误: {e}")
            
            return MCPTestResult(
                success=False,
                message="❌ 测试失败",
                response_time_ms=total_time,
                error=str(e),
                error_type=type(e).__name__,
                suggestions=[
                    "请检查服务器是否在线",
                    "请确认配置正确",
                    "请检查API Key是否有效"
                ]
            )


# 全局单例
mcp_test_service = MCPTestService()