✨ feat(service): 增加服务唯一性验证功能

- 增加服务名称配置项，用于进程唯一性验证
- 实现进程管理模块，提供服务唯一性检查和进程管理功能
- 在主函数中集成服务唯一性检查逻辑
- 优化日志输出，增加启动时的信息日志
- 更新 README，添加服务唯一性验证相关说明

.env.example

@@ -28,6 +28,9 @@ AUTH_TOKENS_FILE=tokens.txt
 # 服务监听端口
 LISTEN_PORT=8080
 
+# 服务名称（用于进程唯一性验证）
+SERVICE_NAME=z-ai2api-server
+
 # 调试日志
 DEBUG_LOGGING=true
 
@@ -45,6 +48,6 @@ SCAN_LIMIT=200000
 # ========== 错误码400处理 ==========
 
 # 重试次数
-MAX_RETRIES=5
+MAX_RETRIES=6
 # 初始重试延迟
 RETRY_DELAY=1

.gitignore

@@ -14,6 +14,7 @@ main.onefile-build/
 *.yaml
 logs/
 backup/
+uv.lock
 
 # AI Toolset
 .augment/

README.md

@@ -23,6 +23,7 @@
 - 📊 **多模型映射** - 智能上游模型路由
 - 🔄 **Token 池管理** - 自动轮询、容错恢复、动态更新
 - 🛡️ **错误处理** - 完善的异常捕获和重试机制
+- 🔒 **服务唯一性** - 基于进程名称(pname)的服务唯一性验证，防止重复启动
 
 ## 🚀 快速开始
 
@@ -245,6 +246,13 @@ A: 这通常是因为 Token 获取失败导致的。请检查：
 - Token 文件是否存在且格式正确（`tokens.txt`）
 - 网络连接是否正常，能否访问 Z.AI API
 
+**Q: 启动时提示"服务已在运行"怎么办？**
+A: 这是服务唯一性验证功能，防止重复启动。解决方法：
+- 检查是否已有服务实例在运行：`ps aux | grep z-ai2api-server`
+- 停止现有实例后再启动新的
+- 如果确认没有实例运行，删除 PID 文件：`rm z-ai2api-server.pid`
+- 可通过环境变量 `SERVICE_NAME` 自定义服务名称避免冲突
+
 **Q: 如何通过 Claude Code 使用本服务？**
 
 A: 创建 [zai.js](https://gist.githubusercontent.com/musistudio/b35402d6f9c95c64269c7666b8405348/raw/f108d66fa050f308387938f149a2b14a295d29e9/gistfile1.txt) 这个 ccr 插件放在`./.claude-code-router/plugins`目录下，配置 `./.claude-code-router/config.json` 指向本服务地址，使用 `AUTH_TOKEN` 进行认证。

app/core/config.py

@@ -109,6 +109,7 @@ class Settings(BaseSettings):
     # Server Configuration
     LISTEN_PORT: int = int(os.getenv("LISTEN_PORT", "8080"))
     DEBUG_LOGGING: bool = os.getenv("DEBUG_LOGGING", "true").lower() == "true"
+    SERVICE_NAME: str = os.getenv("SERVICE_NAME", "z-ai2api-server")
 
     ANONYMOUS_MODE: bool = os.getenv("ANONYMOUS_MODE", "true").lower() == "true"
     TOOL_SUPPORT: bool = os.getenv("TOOL_SUPPORT", "true").lower() == "true"

app/core/openai.py

@@ -156,12 +156,21 @@ async def chat_completions(request: OpenAIRequest, authorization: str = Header(.
 
                             # 初始化工具处理器（如果需要）
                             has_tools = transformed["body"].get("tools") is not None
+                            has_mcp_servers = bool(transformed["body"].get("mcp_servers"))
                             tool_handler = None
-                            if has_tools:
+
+                            # 如果有工具定义或MCP服务器，都需要工具处理器
+                            if has_tools or has_mcp_servers:
                                 chat_id = transformed["body"]["chat_id"]
                                 model = request.model
                                 tool_handler = SSEToolHandler(chat_id, model)
-                                logger.info(f"🔧 初始化工具处理器: {len(transformed['body'].get('tools', []))} 个工具")
+
+                                if has_tools and has_mcp_servers:
+                                    logger.info(f"🔧 初始化工具处理器: {len(transformed['body'].get('tools', []))} 个OpenAI工具 + {len(transformed['body'].get('mcp_servers', []))} 个MCP服务器")
+                                elif has_tools:
+                                    logger.info(f"🔧 初始化工具处理器: {len(transformed['body'].get('tools', []))} 个OpenAI工具")
+                                elif has_mcp_servers:
+                                    logger.info(f"🔧 初始化工具处理器: {len(transformed['body'].get('mcp_servers', []))} 个MCP服务器")
 
                             # 处理状态
                             has_thinking = False
@@ -368,7 +377,7 @@ async def chat_completions(request: OpenAIRequest, authorization: str = Header(.
                                                             "system_fingerprint": "fp_zai_001",
                                                         }
                                                         output_data = f"data: {json.dumps(content_chunk)}\n\n"
-                                                        logger.debug(f"➡️ 输出内容块到客户端: {output_data[:1000]}...")
+                                                        logger.debug(f"➡️ 输出内容块到客户端: {output_data}")
                                                         yield output_data
 
                                                     # 处理完成

app/core/zai_transformer.py

@@ -209,10 +209,19 @@ class ZAITransformer:
         is_search = requested_model == settings.SEARCH_MODEL
         is_air = requested_model == settings.AIR_MODEL
 
+        # 检查是否需要搜索功能（更灵活的检测）
+        needs_search = (
+            is_search or  # 明确的搜索模型
+            "search" in requested_model.lower() or  # 模型名包含search
+            request.get("enable_search", False) or  # 显式启用搜索
+            (request.get("tools") and any("search" in str(tool).lower() for tool in request.get("tools", [])))  # 工具中包含搜索
+        )
+
         # 获取上游模型ID（使用模型映射）
         upstream_model_id = self.model_mapping.get(requested_model, "0727-360B-API")
         logger.debug(f"  模型映射: {requested_model} -> {upstream_model_id}")
         logger.debug(f"  模型特性检测: is_search={is_search}, is_thinking={is_thinking}, is_air={is_air}")
+        logger.debug(f"  需要搜索功能: {needs_search}")
         logger.debug(f"  SEARCH_MODEL配置: {settings.SEARCH_MODEL}")
 
         # 处理消息列表
@@ -263,17 +272,28 @@ class ZAITransformer:
 
         # 构建MCP服务器列表
         mcp_servers = []
-        if is_search:
+        if needs_search:
             mcp_servers.append("deep-web-search")
-            logger.info(f"🔍 检测到搜索模型，添加 deep-web-search MCP 服务器")
+            logger.info(f"🔍 检测到需要搜索功能，添加 deep-web-search MCP 服务器")
+            logger.debug(f"  搜索检测原因: is_search={is_search}, model_name_contains_search={'search' in requested_model.lower()}")
         else:
-            logger.debug(f"  非搜索模型，不添加 MCP 服务器")
+            logger.debug(f"  不需要搜索功能，不添加 MCP 服务器")
 
         logger.debug(f"  MCP服务器列表: {mcp_servers}")
             
         # 构建上游请求体
         chat_id = generate_uuid()
-        
+
+        # 根据参考文档构建更完整的features配置
+        features_list = []
+        if needs_search:
+            features_list.extend([
+                {"type": "mcp", "server": "deep-web-search", "status": "selected"},
+                {"type": "mcp", "server": "vibe-coding", "status": "hidden"},
+                {"type": "mcp", "server": "ppt-maker", "status": "hidden"},
+                {"type": "mcp", "server": "image-search", "status": "hidden"}
+            ])
+
         body = {
             "stream": True,  # 总是使用流式
             "model": upstream_model_id,  # 使用映射后的模型ID
@@ -281,11 +301,11 @@ class ZAITransformer:
             "params": {},
             "features": {
                 "image_generation": False,
-                "web_search": is_search,
-                "auto_web_search": is_search,
-                "preview_mode": False,
+                "web_search": needs_search,
+                "auto_web_search": needs_search,
+                "preview_mode": True if needs_search else False,  # 搜索时启用预览模式
                 "flags": [],
-                "features": [],
+                "features": features_list,
                 "enable_thinking": is_thinking,
             },
             "background_tasks": {
@@ -300,10 +320,14 @@ class ZAITransformer:
                 "{{CURRENT_DATE}}": datetime.now().strftime("%Y-%m-%d"),
                 "{{CURRENT_TIME}}": datetime.now().strftime("%H:%M:%S"),
                 "{{CURRENT_WEEKDAY}}": datetime.now().strftime("%A"),
-                "{{CURRENT_TIMEZONE}}": "UTC",
+                "{{CURRENT_TIMEZONE}}": "Asia/Shanghai",  # 使用更合适的时区
                 "{{USER_LANGUAGE}}": "zh-CN",
             },
-            "model_item": {},
+            "model_item": {
+                "id": upstream_model_id,
+                "name": requested_model,
+                "owned_by": "z.ai"
+            },
             "chat_id": chat_id,
             "id": generate_uuid(),
         }
@@ -336,11 +360,24 @@ class ZAITransformer:
 
         # 记录关键的请求信息用于调试
         logger.debug(f"  📋 发送到Z.AI的关键信息:")
+        logger.debug(f"    - 原始模型: {requested_model}")
         logger.debug(f"    - 上游模型: {body['model']}")
         logger.debug(f"    - MCP服务器: {body['mcp_servers']}")
         logger.debug(f"    - web_search: {body['features']['web_search']}")
         logger.debug(f"    - auto_web_search: {body['features']['auto_web_search']}")
         logger.debug(f"    - 消息数量: {len(body['messages'])}")
+        tools_count = len(body.get('tools') or [])
+        logger.debug(f"    - 工具数量: {tools_count}")
+
+        # 特别记录MCP相关信息
+        if body['mcp_servers']:
+            logger.info(f"🎯 MCP服务器配置成功: {body['mcp_servers']}")
+            logger.debug(f"  📋 完整的features配置: {json.dumps(body['features'], ensure_ascii=False, indent=2)}")
+        else:
+            logger.warning(f"⚠️ 未配置MCP服务器 - 检查模型: {requested_model}, 搜索需��: {needs_search}")
+
+        # 记录完整的请求体（用于调试）
+        logger.debug(f"  📋 完整请求体: {json.dumps(body, ensure_ascii=False, indent=2)}")
 
         return {"body": body, "config": config, "token": token}
 

app/utils/process_manager.py

@@ -0,0 +1,303 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+进程管理模块
+提供服务唯一性验证和进程管理功能
+"""
+
+import os
+import sys
+import time
+import psutil
+from typing import Optional, List
+from pathlib import Path
+
+from app.utils.logger import get_logger
+
+logger = get_logger()
+
+
+class ProcessManager:
+    """进程管理器 - 负责服务唯一性验证和进程管理"""
+    
+    def __init__(self, service_name: str = "z-ai2api-server", port: int = 8080):
+        """
+        初始化进程管理器
+        
+        Args:
+            service_name: 服务名称，用于进程名称标识
+            port: 服务端口，用于唯一性检查
+        """
+        self.service_name = service_name
+        self.port = port
+        self.current_pid = os.getpid()
+        self.pid_file = Path(f"{service_name}.pid")
+        
+    def check_service_uniqueness(self) -> bool:
+        """
+        检查服务唯一性
+
+        通过以下方式验证：
+        1. 检查 PID 文件
+        2. 检查端口是否被占用
+        3. 检查进程名称 (pname) 是否已存在（可选）
+
+        Returns:
+            bool: True 表示可以启动服务，False 表示已有实例运行
+        """
+        logger.info(f"🔍 检查服务唯一性: {self.service_name} (端口: {self.port})")
+
+        # 1. 优先检查 PID 文件（最可靠）
+        if self._check_pid_file():
+            return False
+
+        # 2. 检查端口占用
+        if self._check_port_usage():
+            return False
+
+        # 3. 检查进程名称（作为额外保障）
+        if self._check_process_by_name():
+            return False
+
+        logger.info("✅ 服务唯一性检查通过，可以启动服务")
+        return True
+    
+    def _check_process_by_name(self) -> bool:
+        """
+        通过进程名称检查是否已有实例运行
+
+        这是一个保守的检查，只检查明确的服务进程标识
+
+        Returns:
+            bool: True 表示发现同名进程，False 表示未发现
+        """
+        try:
+            running_processes = []
+
+            for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
+                try:
+                    proc_info = proc.info
+
+                    # 跳过当前进程
+                    if proc_info['pid'] == self.current_pid:
+                        continue
+
+                    # 只检查进程名称直接匹配服务名称的情况
+                    # 这通常发生在使用 Granian 的 process_name 参数时
+                    if proc_info['name'] and proc_info['name'] == self.service_name:
+                        running_processes.append(proc_info)
+                        continue
+
+                    # 检查命令行参数中是否包含明确的服务标识
+                    cmdline = proc_info.get('cmdline', [])
+                    if cmdline and len(cmdline) >= 2:
+                        cmdline_str = ' '.join(cmdline)
+
+                        # 只检查通过 Granian 启动且明确指定了进程名称的服务
+                        if (f'--process-name={self.service_name}' in cmdline_str or
+                            f'process_name={self.service_name}' in cmdline_str):
+                            running_processes.append(proc_info)
+
+                except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
+                    # 进程可能已经结束或无权限访问
+                    continue
+
+            if running_processes:
+                logger.warning(f"⚠️ 发现 {len(running_processes)} 个同名进程正在运行:")
+                for proc_info in running_processes:
+                    cmdline = proc_info.get('cmdline', [])
+                    cmdline_preview = ' '.join(cmdline[:3]) + '...' if len(cmdline) > 3 else ' '.join(cmdline)
+                    logger.warning(f"   PID: {proc_info['pid']}, 名称: {proc_info['name']}, 命令: {cmdline_preview}")
+                logger.warning(f"❌ 服务 {self.service_name} 已在运行，请先停止现有实例")
+                return True
+
+            return False
+
+        except Exception as e:
+            logger.error(f"❌ 检查进程名称时发生错误: {e}")
+            return False
+    
+    def _check_port_usage(self) -> bool:
+        """
+        检查端口是否被占用
+        
+        Returns:
+            bool: True 表示端口被占用，False 表示端口可用
+        """
+        try:
+            # 获取所有网络连接
+            connections = psutil.net_connections(kind='inet')
+            
+            for conn in connections:
+                if (conn.laddr.port == self.port and 
+                    conn.status in [psutil.CONN_LISTEN, psutil.CONN_ESTABLISHED]):
+                    
+                    # 尝试获取占用端口的进程信息
+                    try:
+                        proc = psutil.Process(conn.pid) if conn.pid else None
+                        proc_name = proc.name() if proc else "未知进程"
+                        logger.warning(f"⚠️ 端口 {self.port} 已被占用")
+                        logger.warning(f"   占用进程: PID {conn.pid}, 名称: {proc_name}")
+                        logger.warning(f"❌ 无法启动服务，端口 {self.port} 不可用")
+                        return True
+                    except (psutil.NoSuchProcess, psutil.AccessDenied):
+                        logger.warning(f"⚠️ 端口 {self.port} 已被占用（无法获取进程信息）")
+                        return True
+            
+            return False
+            
+        except Exception as e:
+            logger.error(f"❌ 检查端口占用时发生错误: {e}")
+            return False
+    
+    def _check_pid_file(self) -> bool:
+        """
+        检查 PID 文件
+        
+        Returns:
+            bool: True 表示发现有效的 PID 文件，False 表示无冲突
+        """
+        try:
+            if not self.pid_file.exists():
+                return False
+            
+            # 读取 PID 文件
+            pid_content = self.pid_file.read_text().strip()
+            if not pid_content.isdigit():
+                logger.warning(f"⚠️ PID 文件格式无效: {self.pid_file}")
+                self._cleanup_pid_file()
+                return False
+            
+            old_pid = int(pid_content)
+            
+            # 检查进程是否仍在运行
+            try:
+                proc = psutil.Process(old_pid)
+                if proc.is_running():
+                    logger.warning(f"⚠️ 发现有效的 PID 文件: {self.pid_file}")
+                    logger.warning(f"   进程 PID {old_pid} 仍在运行: {proc.name()}")
+                    logger.warning(f"❌ 服务可能已在运行，请检查进程或删除 PID 文件")
+                    return True
+                else:
+                    logger.info(f"🧹 清理无效的 PID 文件: {self.pid_file}")
+                    self._cleanup_pid_file()
+                    return False
+            except psutil.NoSuchProcess:
+                logger.info(f"🧹 清理过期的 PID 文件: {self.pid_file}")
+                self._cleanup_pid_file()
+                return False
+                
+        except Exception as e:
+            logger.error(f"❌ 检查 PID 文件时发生错误: {e}")
+            return False
+    
+    def _cleanup_pid_file(self):
+        """清理 PID 文件"""
+        try:
+            if self.pid_file.exists():
+                self.pid_file.unlink()
+                logger.debug(f"🧹 已删除 PID 文件: {self.pid_file}")
+        except Exception as e:
+            logger.error(f"❌ 删除 PID 文件失败: {e}")
+    
+    def create_pid_file(self):
+        """创建 PID 文件"""
+        try:
+            self.pid_file.write_text(str(self.current_pid))
+            logger.info(f"📝 创建 PID 文件: {self.pid_file} (PID: {self.current_pid})")
+        except Exception as e:
+            logger.error(f"❌ 创建 PID 文件失败: {e}")
+    
+    def cleanup_on_exit(self):
+        """退出时清理资源"""
+        logger.info(f"🧹 清理进程资源 (PID: {self.current_pid})")
+        self._cleanup_pid_file()
+    
+    def get_running_instances(self) -> List[dict]:
+        """
+        获取所有运行中的服务实例
+
+        Returns:
+            List[dict]: 运行中的实例信息列表
+        """
+        instances = []
+
+        try:
+            for proc in psutil.process_iter(['pid', 'name', 'cmdline', 'create_time']):
+                try:
+                    proc_info = proc.info
+
+                    # 跳过当前进程
+                    if proc_info['pid'] == self.current_pid:
+                        continue
+
+                    # 使用与 _check_process_by_name 相同的保守逻辑
+                    is_service = False
+
+                    # 只检查进程名称直接匹配服务名称的情况
+                    if proc_info['name'] and proc_info['name'] == self.service_name:
+                        is_service = True
+
+                    # 检查命令行参数中是否包含明确的服务标识
+                    cmdline = proc_info.get('cmdline', [])
+                    if cmdline and len(cmdline) >= 2:
+                        cmdline_str = ' '.join(cmdline)
+
+                        # 只检查通过 Granian 启动且明确指定了进程名称的服务
+                        if (f'--process-name={self.service_name}' in cmdline_str or
+                            f'process_name={self.service_name}' in cmdline_str):
+                            is_service = True
+
+                    if is_service:
+                        instances.append({
+                            'pid': proc_info['pid'],
+                            'name': proc_info['name'],
+                            'cmdline': cmdline,
+                            'create_time': proc_info['create_time'],
+                            'start_time': time.strftime('%Y-%m-%d %H:%M:%S',
+                                                      time.localtime(proc_info['create_time']))
+                        })
+
+                except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
+                    continue
+
+        except Exception as e:
+            logger.error(f"❌ 获取运行实例时发生错误: {e}")
+
+        return instances
+
+
+def ensure_service_uniqueness(service_name: str = "z-ai2api-server", port: int = 8080) -> bool:
+    """
+    确保服务唯一性的便捷函数
+    
+    Args:
+        service_name: 服务名称
+        port: 服务端口
+        
+    Returns:
+        bool: True 表示可以启动，False 表示应该退出
+    """
+    manager = ProcessManager(service_name, port)
+    
+    if not manager.check_service_uniqueness():
+        logger.error("❌ 服务唯一性检查失败，程序退出")
+        
+        # 显示运行中的实例
+        instances = manager.get_running_instances()
+        if instances:
+            logger.info("📋 当前运行的实例:")
+            for instance in instances:
+                logger.info(f"   PID: {instance['pid']}, 启动时间: {instance['start_time']}")
+        
+        return False
+    
+    # 创建 PID 文件
+    manager.create_pid_file()
+    
+    # 注册退出清理
+    import atexit
+    atexit.register(manager.cleanup_on_exit)
+    
+    return True

main.py

@@ -1,6 +1,9 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
+import os
+import sys
+import psutil
 from contextlib import asynccontextmanager
 from fastapi import FastAPI, Response
 from fastapi.middleware.cors import CORSMiddleware
@@ -10,6 +13,7 @@ from app.core import openai
 from app.utils.reload_config import RELOAD_CONFIG
 from app.utils.logger import setup_logger
 from app.utils.token_pool import initialize_token_pool
+from app.utils.process_manager import ensure_service_uniqueness
 
 from granian import Granian
 
@@ -62,14 +66,32 @@ async def root():
 
 
 def run_server():
-    Granian(
-        "main:app",
-        interface="asgi",
-        address="0.0.0.0",
-        port=settings.LISTEN_PORT,
-        reload=False,  # 生产环境请关闭热重载
-        **RELOAD_CONFIG,
-    ).serve()
+    # 服务唯一性检查
+    service_name = settings.SERVICE_NAME
+    if not ensure_service_uniqueness(service_name=service_name, port=settings.LISTEN_PORT):
+        logger.error("❌ 服务已在运行，程序退出")
+        sys.exit(1)
+
+    logger.info(f"🚀 启动 {service_name} 服务...")
+    logger.info(f"📡 监听地址: 0.0.0.0:{settings.LISTEN_PORT}")
+    logger.info(f"🔧 调试模式: {'开启' if settings.DEBUG_LOGGING else '关闭'}")
+    logger.info(f"🔐 匿名模式: {'开启' if settings.ANONYMOUS_MODE else '关闭'}")
+
+    try:
+        Granian(
+            "main:app",
+            interface="asgi",
+            address="0.0.0.0",
+            port=settings.LISTEN_PORT,
+            reload=True,  # 生产环境请关闭热重载
+            process_name=service_name,  # 设置进程名称
+            **RELOAD_CONFIG,
+        ).serve()
+    except KeyboardInterrupt:
+        logger.info("🛑 收到中断信号，正在关闭服务...")
+    except Exception as e:
+        logger.error(f"❌ 服务启动失败: {e}")
+        sys.exit(1)
 
 
 if __name__ == "__main__":

pyproject.toml

@@ -25,15 +25,16 @@ classifiers = [
 ]
 dependencies = [
     "fastapi==0.104.1",
-    "granian[reload]==2.5.2",
+    "granian[reload,pname]==2.5.2",
     "requests==2.32.5",
+    "httpx==0.27.0",
     "pydantic==2.11.7",
     "pydantic-settings==2.10.1",
     "pydantic-core==2.33.2",
     "typing-inspection==0.4.1",
     "fake-useragent==2.2.0",
     "loguru==0.7.3",
-    "httpx==0.27.0"
+    "psutil>=7.0.0",
 ]
 
 [project.scripts]

requirements.txt

@@ -1,5 +1,5 @@
 fastapi==0.116.1
-granian[reload]==2.5.2
+granian[reload,pname]==2.5.2
 requests==2.32.5
 httpx==0.27.0
 pydantic==2.11.7
@@ -7,4 +7,5 @@ pydantic-settings==2.10.1
 pydantic-core==2.33.2
 typing-inspection==0.4.1
 fake-useragent==2.2.0
-loguru==0.7.3
+loguru==0.7.3
+psutil>=7.0.0

tests/test_comprehensive_tool_calls.py

@@ -0,0 +1,254 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+全面的工具调用测试套件
+覆盖各种工具类型、参数格式、传输模式和边界情况
+"""
+
+import json
+import time
+from typing import Dict, Any, List
+from app.utils.sse_tool_handler import SSEToolHandler
+from app.utils.logger import get_logger
+
+logger = get_logger()
+
+class TestResult:
+    """测试结果统计"""
+    def __init__(self, test_name: str):
+        self.test_name = test_name
+        self.passed = 0
+        self.failed = 0
+        self.errors = []
+    
+    def add_pass(self):
+        self.passed += 1
+    
+    def add_fail(self, error_msg: str):
+        self.failed += 1
+        self.errors.append(error_msg)
+    
+    def print_summary(self):
+        total = self.passed + self.failed
+        success_rate = (self.passed / total * 100) if total > 0 else 0
+        
+        print(f"\n📊 {self.test_name} 测试汇总:")
+        print(f"  总测试数: {total}")
+        print(f"  ✅ 通过: {self.passed}")
+        print(f"  ❌ 失败: {self.failed}")
+        print(f"  📈 成功率: {success_rate:.1f}%")
+        
+        if self.errors:
+            print(f"\n❌ 失败详情:")
+            for i, error in enumerate(self.errors, 1):
+                print(f"  {i}. {error}")
+
+def test_various_tool_types():
+    """测试各种类型的工具调用"""
+    
+    result = TestResult("工具类型测试")
+    
+    # 定义各种工具类型的测试用例
+    tool_scenarios = [
+        {
+            "name": "浏览器导航工具",
+            "tool_name": "browser_navigate",
+            "arguments": '{"url": "https://www.google.com"}',
+            "expected_args": {"url": "https://www.google.com"},
+            "description": "测试浏览器导航工具的URL参数"
+        },
+        {
+            "name": "天气查询工具",
+            "tool_name": "get_weather",
+            "arguments": '{"city": "北京", "unit": "celsius"}',
+            "expected_args": {"city": "北京", "unit": "celsius"},
+            "description": "测试天气查询工具的城市和单位参数"
+        },
+        {
+            "name": "文件操作工具",
+            "tool_name": "file_write",
+            "arguments": '{"path": "/tmp/test.txt", "content": "Hello World", "encoding": "utf-8"}',
+            "expected_args": {"path": "/tmp/test.txt", "content": "Hello World", "encoding": "utf-8"},
+            "description": "测试文件写入工具的多参数"
+        },
+        {
+            "name": "搜索工具",
+            "tool_name": "web_search",
+            "arguments": '{"query": "Python编程", "limit": 10, "safe_search": true}',
+            "expected_args": {"query": "Python编程", "limit": 10, "safe_search": True},
+            "description": "测试搜索工具的混合类型参数"
+        },
+        {
+            "name": "数据库查询工具",
+            "tool_name": "db_query",
+            "arguments": '{"sql": "SELECT * FROM users WHERE age > ?", "params": [18], "timeout": 30.5}',
+            "expected_args": {"sql": "SELECT * FROM users WHERE age > ?", "params": [18], "timeout": 30.5},
+            "description": "测试数据库工具的复杂参数结构"
+        },
+        {
+            "name": "API调用工具",
+            "tool_name": "api_call",
+            "arguments": '{"method": "POST", "url": "https://api.example.com/data", "headers": {"Content-Type": "application/json"}, "body": {"key": "value"}}',
+            "expected_args": {"method": "POST", "url": "https://api.example.com/data", "headers": {"Content-Type": "application/json"}, "body": {"key": "value"}},
+            "description": "测试API调用工具的嵌套对象参数"
+        },
+        {
+            "name": "图像处理工具",
+            "tool_name": "image_resize",
+            "arguments": '{"input_path": "image.jpg", "output_path": "resized.jpg", "width": 800, "height": 600, "maintain_aspect": false}',
+            "expected_args": {"input_path": "image.jpg", "output_path": "resized.jpg", "width": 800, "height": 600, "maintain_aspect": False},
+            "description": "测试图像处理工具的数值和布尔参数"
+        },
+        {
+            "name": "邮件发送工具",
+            "tool_name": "send_email",
+            "arguments": '{"to": ["user1@example.com", "user2@example.com"], "subject": "测试邮件", "body": "这是一封测试邮件\\n包含换行符", "attachments": []}',
+            "expected_args": {"to": ["user1@example.com", "user2@example.com"], "subject": "测试邮件", "body": "这是一封测试邮件\n包含换行符", "attachments": []},
+            "description": "测试邮件工具的数组参数和转义字符"
+        }
+    ]
+    
+    print("🔧 测试各种类型的工具调用")
+    print("=" * 80)
+    
+    for i, scenario in enumerate(tool_scenarios, 1):
+        print(f"\n测试 {i}: {scenario['name']}")
+        print(f"描述: {scenario['description']}")
+        
+        try:
+            handler = SSEToolHandler("test_chat_id", "GLM-4.5")
+            
+            # 构造完整的工具调用数据
+            tool_data = {
+                "edit_index": 0,
+                "edit_content": f'<glm_block >{{"type": "mcp", "data": {{"metadata": {{"id": "call_{i}", "name": "{scenario["tool_name"]}", "arguments": "{scenario["arguments"]}", "result": "", "status": "completed"}}}}, "thought": null}}</glm_block>',
+                "phase": "tool_call"
+            }
+            
+            # 处理工具调用
+            chunks = list(handler.process_tool_call_phase(tool_data, is_stream=False))
+            
+            # 验证结果
+            if handler.active_tools:
+                tool = list(handler.active_tools.values())[0]
+                actual_args = tool["arguments"]
+                expected_args = scenario["expected_args"]
+                
+                if actual_args == expected_args:
+                    print(f"  ✅ 参数解析正确: {actual_args}")
+                    result.add_pass()
+                else:
+                    error_msg = f"{scenario['name']}: 参数不匹配 - 期望: {expected_args}, 实际: {actual_args}"
+                    print(f"  ❌ {error_msg}")
+                    result.add_fail(error_msg)
+            else:
+                error_msg = f"{scenario['name']}: 未检测到工具调用"
+                print(f"  ❌ {error_msg}")
+                result.add_fail(error_msg)
+                
+        except Exception as e:
+            error_msg = f"{scenario['name']}: 处理异常 - {str(e)}"
+            print(f"  ❌ {error_msg}")
+            result.add_fail(error_msg)
+    
+    result.print_summary()
+    return result
+
+def test_parameter_formats():
+    """测试各种参数格式"""
+    
+    result = TestResult("参数格式测试")
+    
+    # 定义各种参数格式的测试用例
+    format_scenarios = [
+        {
+            "name": "空参数",
+            "arguments": "{}",
+            "expected": {},
+            "description": "测试空参数对象"
+        },
+        {
+            "name": "null参数",
+            "arguments": "null",
+            "expected": {},
+            "description": "测试null参数值"
+        },
+        {
+            "name": "转义JSON字符串",
+            "arguments": '{\\"key\\": \\"value\\"}',
+            "expected": {"key": "value"},
+            "description": "测试转义的JSON字符串"
+        },
+        {
+            "name": "包含特殊字符",
+            "arguments": '{"text": "Hello\\nWorld\\t!", "emoji": "😀🎉", "unicode": "中文测试"}',
+            "expected": {"text": "Hello\nWorld\t!", "emoji": "😀🎉", "unicode": "中文测试"},
+            "description": "测试包含换行符、制表符、emoji和中文的参数"
+        },
+        {
+            "name": "数值类型",
+            "arguments": '{"int": 42, "float": 3.14159, "negative": -100, "zero": 0}',
+            "expected": {"int": 42, "float": 3.14159, "negative": -100, "zero": 0},
+            "description": "测试各种数值类型参数"
+        },
+        {
+            "name": "布尔类型",
+            "arguments": '{"true_val": true, "false_val": false}',
+            "expected": {"true_val": True, "false_val": False},
+            "description": "测试布尔类型参数"
+        },
+        {
+            "name": "数组参数",
+            "arguments": '{"empty_array": [], "string_array": ["a", "b", "c"], "mixed_array": [1, "two", true, null]}',
+            "expected": {"empty_array": [], "string_array": ["a", "b", "c"], "mixed_array": [1, "two", True, None]},
+            "description": "测试各种数组类型参数"
+        },
+        {
+            "name": "嵌套对象",
+            "arguments": '{"nested": {"level1": {"level2": {"value": "deep"}}}, "array_of_objects": [{"id": 1}, {"id": 2}]}',
+            "expected": {"nested": {"level1": {"level2": {"value": "deep"}}}, "array_of_objects": [{"id": 1}, {"id": 2}]},
+            "description": "测试深度嵌套的对象和对象数组"
+        },
+        {
+            "name": "长字符串",
+            "arguments": '{"long_text": "' + "A" * 1000 + '"}',
+            "expected": {"long_text": "A" * 1000},
+            "description": "测试长字符串参数"
+        },
+        {
+            "name": "包含引号的字符串",
+            "arguments": '{"quoted": "He said \\"Hello\\" to me", "single_quote": "It\'s working"}',
+            "expected": {"quoted": 'He said "Hello" to me', "single_quote": "It's working"},
+            "description": "测试包含引号的字符串参数"
+        }
+    ]
+    
+    print("\n📝 测试各种参数格式")
+    print("=" * 80)
+    
+    for i, scenario in enumerate(format_scenarios, 1):
+        print(f"\n测试 {i}: {scenario['name']}")
+        print(f"描述: {scenario['description']}")
+        
+        try:
+            handler = SSEToolHandler("test_chat_id", "GLM-4.5")
+            
+            # 直接测试参数解析
+            result_args = handler._parse_partial_arguments(scenario["arguments"])
+            
+            if result_args == scenario["expected"]:
+                print(f"  ✅ 参数解析正确")
+                result.add_pass()
+            else:
+                error_msg = f"{scenario['name']}: 参数解析错误 - 期望: {scenario['expected']}, 实际: {result_args}"
+                print(f"  ❌ {error_msg}")
+                result.add_fail(error_msg)
+                
+        except Exception as e:
+            error_msg = f"{scenario['name']}: 解析异常 - {str(e)}"
+            print(f"  ❌ {error_msg}")
+            result.add_fail(error_msg)
+    
+    result.print_summary()
+    return result

tests/test_live_server.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+测试当前运行的服务器是否正确处理GLM-4.5-Search模型
+"""
+
+import asyncio
+import json
+import httpx
+from app.core.config import settings
+
+async def test_live_server():
+    """测试实际运行的服务器"""
+    
+    print("🧪 测试当前运行的服务器...")
+    print(f"服务器地址: http://localhost:{settings.LISTEN_PORT}")
+    print()
+    
+    try:
+        async with httpx.AsyncClient() as client:
+            # 测试搜索模型请求
+            search_request = {
+                "model": "GLM-4.5-Search",
+                "messages": [
+                    {"role": "user", "content": "请搜索今天北京的天气"}
+                ],
+                "stream": True  # 使用流式以便观察日志
+            }
+            
+            headers = {
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {settings.AUTH_TOKEN}"
+            }
+            
+            print(f"📤 发送GLM-4.5-Search请求...")
+            print(f"请求内容: {json.dumps(search_request, ensure_ascii=False, indent=2)}")
+            print()
+            
+            # 发送请求并接收流式响应
+            async with client.stream(
+                "POST",
+                f"http://localhost:{settings.LISTEN_PORT}/v1/chat/completions",
+                json=search_request,
+                headers=headers,
+                timeout=30.0
+            ) as response:
+                
+                print(f"📥 响应状态: {response.status_code}")
+                
+                if response.status_code == 200:
+                    print(f"✅ 请求成功，开始接收流式响应...")
+                    print(f"💡 请查看服务器日志以确认是否正确添加了 deep-web-search MCP 服务器")
+                    print()
+                    
+                    # 读取前几个响应块
+                    chunk_count = 0
+                    async for line in response.aiter_lines():
+                        if line.startswith("data: "):
+                            chunk_count += 1
+                            if chunk_count <= 3:  # 只显示前3个块
+                                data = line[6:]  # 去掉 "data: " 前缀
+                                if data.strip() and data.strip() != "[DONE]":
+                                    try:
+                                        chunk_data = json.loads(data)
+                                        content = chunk_data.get("choices", [{}])[0].get("delta", {}).get("content", "")
+                                        if content:
+                                            print(f"📦 响应块 {chunk_count}: {content}")
+                                    except:
+                                        pass
+                            elif chunk_count > 10:  # 读取足够的块后停止
+                                break
+                    
+                    print(f"\n✅ 流式响应正常，共接收 {chunk_count} 个数据块")
+                    print(f"🔍 请检查服务器日志中是否包含以下信息:")
+                    print(f"   - '模型特性检测: is_search=True'")
+                    print(f"   - '🔍 检测到搜索模型，添加 deep-web-search MCP 服务器'")
+                    print(f"   - 'MCP服务器列表: [\"deep-web-search\"]'")
+                    
+                else:
+                    error_text = await response.aread()
+                    print(f"❌ 请求失败: {response.status_code}")
+                    print(f"错误信息: {error_text.decode('utf-8', errors='ignore')}")
+                    
+    except httpx.ConnectError:
+        print(f"❌ 无法连接到服务器 localhost:{settings.LISTEN_PORT}")
+        print(f"   请确保服务器正在运行: python main.py")
+    except Exception as e:
+        print(f"❌ 请求异常: {e}")
+
+async def main():
+    """主函数"""
+    print("=" * 60)
+    print("GLM-4.5-Search 实时服务器测试")
+    print("=" * 60)
+    print()
+    
+    await test_live_server()
+    
+    print()
+    print("=" * 60)
+    print("测试完成")
+    print("=" * 60)
+    print()
+    print("📋 检查清单:")
+    print("1. 服务器是否正常响应 GLM-4.5-Search 请求？")
+    print("2. 日志中是否显示 'is_search=True'？")
+    print("3. 日志中是否显示添加 deep-web-search MCP 服务器？")
+    print("4. 如果以上信息缺失，请重启服务器以加载最新代码")
+
+if __name__ == "__main__":
+    asyncio.run(main())

tests/test_model_comparison.py

@@ -0,0 +1,118 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+对比不同模型的搜索行为
+"""
+
+import asyncio
+import json
+import httpx
+from app.core.config import settings
+
+async def test_model(model_name: str, question: str):
+    """测试特定模型的响应"""
+    
+    print(f"🧪 测试模型: {model_name}")
+    print(f"问题: {question}")
+    print()
+    
+    try:
+        async with httpx.AsyncClient() as client:
+            request_data = {
+                "model": model_name,
+                "messages": [
+                    {"role": "user", "content": question}
+                ],
+                "stream": False  # 使用非流式以便完整查看响应
+            }
+            
+            headers = {
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {settings.AUTH_TOKEN}"
+            }
+            
+            response = await client.post(
+                f"http://localhost:{settings.LISTEN_PORT}/v1/chat/completions",
+                json=request_data,
+                headers=headers,
+                timeout=60.0
+            )
+            
+            if response.status_code == 200:
+                result = response.json()
+                content = result["choices"][0]["message"]["content"]
+                print(f"✅ 响应成功:")
+                print(f"内容: {content[:200]}...")
+                print()
+                
+                # 检查是否包含搜索相关的内容
+                search_indicators = [
+                    "搜索", "查询", "实时", "最新", "网络", "互联网",
+                    "search", "query", "real-time", "latest", "web", "internet"
+                ]
+                
+                has_search_content = any(indicator in content.lower() for indicator in search_indicators)
+                if has_search_content:
+                    print(f"🔍 检测到搜索相关内容")
+                else:
+                    print(f"❌ 未检测到搜索相关内容")
+                
+                return content
+            else:
+                print(f"❌ 请求失败: {response.status_code}")
+                print(f"错误: {response.text}")
+                return None
+                
+    except Exception as e:
+        print(f"❌ 请求异常: {e}")
+        return None
+
+async def main():
+    """主测试函数"""
+    print("=" * 80)
+    print("GLM模型搜索能力对比测试")
+    print("=" * 80)
+    print()
+    
+    # 测试问题
+    search_question = "请搜索今天北京的天气情况"
+    general_question = "你好，请介绍一下自己"
+    
+    models_to_test = [
+        "GLM-4.5",
+        "GLM-4.5-Search", 
+        "GLM-4.5-Thinking",
+        "GLM-4.5-Air"
+    ]
+    
+    print("🔍 测试搜索相关问题:")
+    print(f"问题: {search_question}")
+    print("-" * 80)
+    
+    for model in models_to_test:
+        await test_model(model, search_question)
+        print("-" * 40)
+    
+    print()
+    print("💬 测试一般问题:")
+    print(f"问题: {general_question}")
+    print("-" * 80)
+    
+    for model in models_to_test:
+        await test_model(model, general_question)
+        print("-" * 40)
+    
+    print()
+    print("=" * 80)
+    print("测试完成")
+    print("=" * 80)
+    print()
+    print("📋 分析要点:")
+    print("1. GLM-4.5-Search 是否表现出不同的搜索行为？")
+    print("2. 其他模型是否都拒绝搜索请求？")
+    print("3. 模型响应中是否包含实际的搜索结果？")
+    print("4. 检查服务器日志中的MCP服务器配置是否正确")
+
+if __name__ == "__main__":
+    asyncio.run(main())

tests/test_search_model.py

@@ -0,0 +1,180 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+测试GLM-4.5-Search模型的deep-web-search MCP服务器功能
+"""
+
+import asyncio
+import json
+import httpx
+from app.core.config import settings
+from app.core.zai_transformer import ZAITransformer
+from app.utils.logger import setup_logger
+
+# 设置日志
+logger = setup_logger(log_dir="logs", debug_mode=True)
+
+async def test_search_model_mcp():
+    """测试搜索模型的MCP服务器配置"""
+    
+    # 创建转换器实例
+    transformer = ZAITransformer()
+    
+    # 模拟OpenAI请求 - 使用GLM-4.5-Search模型
+    openai_request = {
+        "model": "GLM-4.5-Search",
+        "messages": [
+            {"role": "user", "content": "请搜索一下今天的新闻"}
+        ],
+        "stream": True
+    }
+    
+    print(f"🧪 测试请求:")
+    print(f"  模型: {openai_request['model']}")
+    print(f"  SEARCH_MODEL配置: {settings.SEARCH_MODEL}")
+    print(f"  模型匹配: {openai_request['model'] == settings.SEARCH_MODEL}")
+    print()
+    
+    try:
+        # 转换请求
+        transformed = await transformer.transform_request_in(openai_request)
+        
+        print(f"✅ 转换成功!")
+        print(f"  上游模型: {transformed['body']['model']}")
+        print(f"  MCP服务器: {transformed['body']['mcp_servers']}")
+        print(f"  web_search特性: {transformed['body']['features']['web_search']}")
+        print(f"  auto_web_search特性: {transformed['body']['features']['auto_web_search']}")
+        print()
+        
+        # 检查是否正确添加了deep-web-search
+        mcp_servers = transformed['body']['mcp_servers']
+        if "deep-web-search" in mcp_servers:
+            print("✅ deep-web-search MCP服务器已正确添加!")
+        else:
+            print("❌ deep-web-search MCP服务器未添加!")
+            print(f"   实际MCP服务器列表: {mcp_servers}")
+        
+        return transformed
+        
+    except Exception as e:
+        print(f"❌ 转换失败: {e}")
+        return None
+
+async def test_non_search_model():
+    """测试非搜索模型不应该添加MCP服务器"""
+    
+    transformer = ZAITransformer()
+    
+    # 模拟OpenAI请求 - 使用普通GLM-4.5模型
+    openai_request = {
+        "model": "GLM-4.5",
+        "messages": [
+            {"role": "user", "content": "你好"}
+        ],
+        "stream": True
+    }
+    
+    print(f"🧪 测试普通模型:")
+    print(f"  模型: {openai_request['model']}")
+    print()
+    
+    try:
+        # 转换请求
+        transformed = await transformer.transform_request_in(openai_request)
+        
+        print(f"✅ 转换成功!")
+        print(f"  上游模型: {transformed['body']['model']}")
+        print(f"  MCP服务器: {transformed['body']['mcp_servers']}")
+        print(f"  web_search特性: {transformed['body']['features']['web_search']}")
+        print()
+        
+        # 检查MCP服务器列表应该为空
+        mcp_servers = transformed['body']['mcp_servers']
+        if not mcp_servers:
+            print("✅ 普通模型正确地没有添加MCP服务器!")
+        else:
+            print(f"❌ 普通模型意外添加了MCP服务器: {mcp_servers}")
+        
+        return transformed
+        
+    except Exception as e:
+        print(f"❌ 转换失败: {e}")
+        return None
+
+async def test_actual_request():
+    """测试实际的HTTP请求"""
+    
+    print(f"🌐 测试实际HTTP请求到本地服务器...")
+    
+    # 检查服务器是否运行
+    try:
+        async with httpx.AsyncClient() as client:
+            # 测试服务器是否可达
+            response = await client.get(f"http://localhost:{settings.LISTEN_PORT}/v1/models", timeout=5.0)
+            if response.status_code != 200:
+                print(f"❌ 服务器未运行或不可达: {response.status_code}")
+                return
+                
+            print(f"✅ 服务器运行正常")
+            
+            # 发送搜索模型请求
+            search_request = {
+                "model": "GLM-4.5-Search",
+                "messages": [
+                    {"role": "user", "content": "搜索今天的天气"}
+                ],
+                "stream": False
+            }
+            
+            headers = {
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {settings.AUTH_TOKEN}"
+            }
+            
+            print(f"📤 发送搜索请求...")
+            response = await client.post(
+                f"http://localhost:{settings.LISTEN_PORT}/v1/chat/completions",
+                json=search_request,
+                headers=headers,
+                timeout=30.0
+            )
+            
+            print(f"📥 响应状态: {response.status_code}")
+            if response.status_code == 200:
+                print(f"✅ 请求成功!")
+                # 不打印完整响应，只显示状态
+            else:
+                print(f"❌ 请求失败: {response.text}")
+                
+    except httpx.ConnectError:
+        print(f"❌ 无法连接到服务器 localhost:{settings.LISTEN_PORT}")
+        print(f"   请确保服务器正在运行: python main.py")
+    except Exception as e:
+        print(f"❌ 请求异常: {e}")
+
+async def main():
+    """主测试函数"""
+    print("=" * 60)
+    print("GLM-4.5-Search MCP服务器测试")
+    print("=" * 60)
+    print()
+    
+    # 测试1: 搜索模型应该添加MCP服务器
+    await test_search_model_mcp()
+    print()
+    
+    # 测试2: 普通模型不应该添加MCP服务器
+    await test_non_search_model()
+    print()
+    
+    # 测试3: 实际HTTP请求（如果服务器运行）
+    await test_actual_request()
+    print()
+    
+    print("=" * 60)
+    print("测试完成")
+    print("=" * 60)
+
+if __name__ == "__main__":
+    asyncio.run(main())

tests/test_service_uniqueness.py

@@ -0,0 +1,173 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+测试服务唯一性验证功能
+"""
+
+import time
+import subprocess
+import sys
+from pathlib import Path
+
+from app.core.config import settings
+from app.utils.process_manager import ProcessManager, ensure_service_uniqueness
+from app.utils.logger import setup_logger
+
+# 设置日志
+logger = setup_logger(log_dir="logs", debug_mode=True)
+
+
+def test_process_manager():
+    """测试进程管理器功能"""
+    print("=" * 60)
+    print("测试进程管理器功能")
+    print("=" * 60)
+    
+    service_name = "test-z-ai2api-server"
+    port = 8081
+    
+    # 创建进程管理器
+    manager = ProcessManager(service_name=service_name, port=port)
+    
+    print(f"\n1. 测试服务唯一性检查...")
+    print(f"   服务名称: {service_name}")
+    print(f"   端口: {port}")
+    
+    # 第一次检查应该通过
+    result1 = manager.check_service_uniqueness()
+    print(f"   第一次检查结果: {'✅ 通过' if result1 else '❌ 失败'}")
+    
+    if result1:
+        # 创建 PID 文件
+        manager.create_pid_file()
+        print(f"   已创建 PID 文件: {manager.pid_file}")
+        
+        # 第二次检查应该失败（因为 PID 文件存在且进程运行中）
+        manager2 = ProcessManager(service_name=service_name, port=port)
+        result2 = manager2.check_service_uniqueness()
+        print(f"   第二次检查结果: {'✅ 通过' if result2 else '❌ 失败（预期）'}")
+        
+        # 清理
+        manager.cleanup_on_exit()
+        print(f"   已清理 PID 文件")
+        
+        # 第三次检查应该通过
+        manager3 = ProcessManager(service_name=service_name, port=port)
+        result3 = manager3.check_service_uniqueness()
+        print(f"   第三次检查结果: {'✅ 通过' if result3 else '❌ 失败'}")
+
+
+def test_convenience_function():
+    """测试便捷函数"""
+    print("\n" + "=" * 60)
+    print("测试便捷函数")
+    print("=" * 60)
+    
+    service_name = "test-convenience-server"
+    port = 8082
+    
+    print(f"\n2. 测试便捷函数...")
+    print(f"   服务名称: {service_name}")
+    print(f"   端口: {port}")
+    
+    # 第一次调用应该成功
+    result1 = ensure_service_uniqueness(service_name=service_name, port=port)
+    print(f"   第一次调用结果: {'✅ 成功' if result1 else '❌ 失败'}")
+    
+    if result1:
+        # 第二次调用应该失败
+        result2 = ensure_service_uniqueness(service_name=service_name, port=port)
+        print(f"   第二次调用结果: {'✅ 成功' if result2 else '❌ 失败（预期）'}")
+        
+        # 手动清理
+        pid_file = Path(f"{service_name}.pid")
+        if pid_file.exists():
+            pid_file.unlink()
+            print(f"   已手动清理 PID 文件")
+
+
+def test_real_service():
+    """测试真实服务场景"""
+    print("\n" + "=" * 60)
+    print("测试真实服务场景")
+    print("=" * 60)
+    
+    service_name = settings.SERVICE_NAME
+    port = settings.LISTEN_PORT
+    
+    print(f"\n3. 测试真实服务场景...")
+    print(f"   服务名称: {service_name}")
+    print(f"   端口: {port}")
+    
+    # 检查当前是否有服务运行
+    manager = ProcessManager(service_name=service_name, port=port)
+    instances = manager.get_running_instances()
+    
+    if instances:
+        print(f"   发现 {len(instances)} 个运行中的实例:")
+        for instance in instances:
+            print(f"     PID: {instance['pid']}, 启动时间: {instance['start_time']}")
+    else:
+        print("   未发现运行中的实例")
+    
+    # 测试唯一性检查
+    result = manager.check_service_uniqueness()
+    print(f"   唯一性检查结果: {'✅ 可以启动' if result else '❌ 已有实例运行'}")
+
+
+def test_port_conflict():
+    """测试端口冲突检测"""
+    print("\n" + "=" * 60)
+    print("测试端口冲突检测")
+    print("=" * 60)
+    
+    print(f"\n4. 测试端口冲突检测...")
+    
+    # 尝试检测一些常用端口
+    test_ports = [80, 443, 8080, 3000, 5000]
+    
+    for port in test_ports:
+        manager = ProcessManager(service_name="test-port-check", port=port)
+        is_occupied = manager._check_port_usage()
+        print(f"   端口 {port}: {'❌ 被占用' if is_occupied else '✅ 可用'}")
+
+
+def main():
+    """主测试函数"""
+    print("🧪 Z.AI2API 服务唯一性验证测试")
+    print("=" * 60)
+    print("此测试将验证以下功能:")
+    print("1. 进程管理器基本功能")
+    print("2. 便捷函数功能")
+    print("3. 真实服务场景")
+    print("4. 端口冲突检测")
+    print("=" * 60)
+    
+    try:
+        # 运行所有测试
+        test_process_manager()
+        test_convenience_function()
+        test_real_service()
+        test_port_conflict()
+        
+        print("\n" + "=" * 60)
+        print("✅ 所有测试完成")
+        print("=" * 60)
+        
+        print("\n📋 使用说��:")
+        print("1. 启动服务时会自动进行唯一性检查")
+        print("2. 如果检测到已有实例运行，新实例将拒绝启动")
+        print("3. 可以通过环境变量 SERVICE_NAME 自定义服务名称")
+        print("4. PID 文件会在服务正常退出时自动清理")
+        print("5. 异常退出的 PID 文件会在下次启动时自动清理")
+        
+    except Exception as e:
+        logger.error(f"❌ 测试过程中发生错误: {e}")
+        import traceback
+        traceback.print_exc()
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

tests/test_sse_optimization.py

@@ -0,0 +1,131 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+测试 SSE 工具调用处理器的优化效果
+"""
+
+import json
+import time
+from app.utils.sse_tool_handler import SSEToolHandler
+from app.utils.logger import get_logger
+
+logger = get_logger()
+
+def test_tool_call_processing():
+    """测试工具调用处理的优化效果"""
+    
+    # 创建处理器
+    handler = SSEToolHandler("test_chat_id", "GLM-4.5")
+    
+    # 模拟 Z.AI 的原始响应数据（基于文档中的示例）
+    test_data_sequence = [
+        # 第一个数据块 - 工具调用开始
+        {
+            "edit_index": 22,
+            "edit_content": '\n\n<glm_block >{"type": "mcp", "data": {"metadata": {"id": "call_fyh97tn03ow", "name": "playwri-browser_navigate", "arguments": "{\\"url\\":\\"https://www.goo',
+            "phase": "tool_call"
+        },
+        # 第二个数据块 - 参数补全
+        {
+            "edit_index": 176,
+            "edit_content": 'gle.com\\"}", "result": "", "display_result": "", "duration": "...", "status": "completed", "is_error": false, "mcp_server": {"name": "mcp-server"}}, "thought": null, "ppt": null, "browser": null}}</glm_block>',
+            "phase": "tool_call"
+        },
+        # 第三个数据块 - 工具调用结束
+        {
+            "edit_index": 199,
+            "edit_content": 'null, "display_result": "", "duration": "...", "status": "completed", "is_error": false, "mcp_server": {"name": "mcp-server"}}, "thought": null, "ppt": null, "browser": null}}</glm_block>',
+            "phase": "other"
+        }
+    ]
+    
+    print("🧪 开始测试 SSE 工具调用处理器优化...")
+    
+    # 处理数据序列
+    all_chunks = []
+    for i, data in enumerate(test_data_sequence):
+        print(f"\n📦 处理数据块 {i+1}: phase={data['phase']}, edit_index={data['edit_index']}")
+        
+        if data["phase"] == "tool_call":
+            chunks = list(handler.process_tool_call_phase(data, is_stream=True))
+        else:
+            chunks = list(handler.process_other_phase(data, is_stream=True))
+        
+        all_chunks.extend(chunks)
+        
+        # 打印生成的块
+        for j, chunk in enumerate(chunks):
+            if chunk.strip():
+                print(f"  📤 输出块 {j+1}: {chunk[:100]}...")
+    
+    print(f"\n✅ 测试完成，共生成 {len(all_chunks)} 个输出块")
+    
+    # 验证工具调用是否正确解析
+    print(f"🔧 活跃工具数: {len(handler.active_tools)}")
+    print(f"✅ 完成工具数: {len(handler.completed_tools)}")
+    
+    # 打印最终的内容缓冲区
+    try:
+        final_content = handler.content_buffer.decode('utf-8', errors='ignore')
+        print(f"\n📝 最终内容缓冲区长度: {len(final_content)}")
+        print(f"📝 内容预览: {final_content[:200]}...")
+    except Exception as e:
+        print(f"❌ 内容缓冲区解析失败: {e}")
+
+def test_partial_arguments_parsing():
+    """测试部分参数解析功能"""
+    
+    handler = SSEToolHandler("test_chat_id", "GLM-4.5")
+    
+    # 测试各种不完整的参数
+    test_cases = [
+        '{"url":"https://www.goo',  # 不完整的URL
+        '{"city":"北京',  # 缺少引号和括号
+        '{"query":"test", "limit":',  # 不完整的数值
+        '{"name":"test"',  # 缺少结束括号
+        '',  # 空字符串
+        '{',  # 只有开始括号
+    ]
+    
+    print("\n🧪 测试部分参数解析...")
+    
+    for i, test_arg in enumerate(test_cases):
+        print(f"\n📦 测试用例 {i+1}: {test_arg}")
+        result = handler._parse_partial_arguments(test_arg)
+        print(f"  ✅ 解析结果: {result}")
+
+def test_performance():
+    """测试性能优化效果"""
+    
+    print("\n🚀 测试性能优化效果...")
+    
+    # 创建大量数据进行性能测试
+    handler = SSEToolHandler("test_chat_id", "GLM-4.5")
+    
+    # 模拟大量的编辑操作
+    start_time = time.time()
+    
+    for i in range(1000):
+        edit_data = {
+            "edit_index": i * 10,
+            "edit_content": f"test_content_{i}",
+            "phase": "tool_call"
+        }
+        list(handler.process_tool_call_phase(edit_data, is_stream=False))
+    
+    end_time = time.time()
+    
+    print(f"⏱️ 处理1000次编辑操作耗时: {end_time - start_time:.3f}秒")
+    print(f"📊 平均每次操作耗时: {(end_time - start_time) * 1000 / 1000:.3f}毫秒")
+
+if __name__ == "__main__":
+    try:
+        test_tool_call_processing()
+        test_partial_arguments_parsing()
+        test_performance()
+        print("\n🎉 所有测试完成！")
+    except Exception as e:
+        logger.error(f"❌ 测试失败: {e}")
+        import traceback
+        traceback.print_exc()

tests/test_tool_handler_optimized.py

@@ -0,0 +1,492 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+测试优化后的SSE工具调用处理器
+基于真实的Z.AI响应格式和日志数据进行全面测试
+"""
+
+import json
+import time
+import traceback
+from typing import List, Dict, Any
+from app.utils.sse_tool_handler import SSEToolHandler
+from app.utils.logger import get_logger
+
+logger = get_logger()
+
+
+class TestResult:
+    """测试结果类"""
+    def __init__(self, name: str):
+        self.name = name
+        self.passed = 0
+        self.failed = 0
+        self.errors = []
+
+    def add_pass(self):
+        self.passed += 1
+
+    def add_fail(self, error: str):
+        self.failed += 1
+        self.errors.append(error)
+
+    def print_summary(self):
+        total = self.passed + self.failed
+        success_rate = (self.passed / total * 100) if total > 0 else 0
+
+        print(f"\n📊 {self.name} 测试结果:")
+        print(f"  ✅ 通过: {self.passed}")
+        print(f"  ❌ 失败: {self.failed}")
+        print(f"  📈 成功率: {success_rate:.1f}%")
+
+        if self.errors:
+            print(f"  🔍 错误详情:")
+            for i, error in enumerate(self.errors, 1):
+                print(f"    {i}. {error}")
+
+
+def parse_openai_chunk(chunk_data: str) -> Dict[str, Any]:
+    """解析OpenAI格式的chunk数据"""
+    try:
+        if chunk_data.startswith("data: "):
+            chunk_data = chunk_data[6:]  # 移除 "data: " 前缀
+        if chunk_data.strip() == "[DONE]":
+            return {"type": "done"}
+        return json.loads(chunk_data)
+    except json.JSONDecodeError:
+        return {"type": "invalid", "raw": chunk_data}
+
+
+def extract_tool_calls(chunks: List[str]) -> List[Dict[str, Any]]:
+    """从chunk列表中提取工具调用信息"""
+    tools = []
+    current_tool = None
+
+    for chunk in chunks:
+        parsed = parse_openai_chunk(chunk)
+        if parsed.get("type") == "invalid":
+            continue
+
+        choices = parsed.get("choices", [])
+        if not choices:
+            continue
+
+        delta = choices[0].get("delta", {})
+        tool_calls = delta.get("tool_calls", [])
+
+        for tc in tool_calls:
+            if tc.get("function", {}).get("name"):  # 新工具开始
+                current_tool = {
+                    "id": tc.get("id"),
+                    "name": tc["function"]["name"],
+                    "arguments": ""
+                }
+                tools.append(current_tool)
+            elif tc.get("function", {}).get("arguments") and current_tool:  # 参数累积
+                current_tool["arguments"] += tc["function"]["arguments"]
+
+    # 解析最终参数
+    for tool in tools:
+        try:
+            tool["parsed_arguments"] = json.loads(tool["arguments"]) if tool["arguments"] else {}
+        except json.JSONDecodeError:
+            tool["parsed_arguments"] = {}
+
+    return tools
+
+
+def test_real_world_scenarios():
+    """测试基于真实Z.AI响应的工具调用处理"""
+
+    result = TestResult("真实场景测试")
+
+    # 基于实际日志的测试数据
+    test_scenarios = [
+        {
+            "name": "浏览器导航工具调用",
+            "description": "模拟打开Google网站的工具调用",
+            "expected_tools": [
+                {
+                    "name": "playwri-browser_navigate",
+                    "id": "call_fyh97tn03ow",
+                    "arguments": {"url": "https://www.google.com"}
+                }
+            ],
+            "data_sequence": [
+                {
+                    "edit_index": 22,
+                    "edit_content": '\n\n<glm_block >{"type": "mcp", "data": {"metadata": {"id": "call_fyh97tn03ow", "name": "playwri-browser_navigate", "arguments": "{\\"url\\":\\"https://www.goo',
+                    "phase": "tool_call"
+                },
+                {
+                    "edit_index": 176,
+                    "edit_content": 'gle.com\\"}", "result": "", "display_result": "", "duration": "...", "status": "completed", "is_error": false, "mcp_server": {"name": "mcp-server"}}, "thought": null, "ppt": null, "browser": null}}</glm_block>',
+                    "phase": "tool_call"
+                },
+                {
+                    "edit_index": 199,
+                    "edit_content": 'null, "display_result": "", "duration": "...", "status": "completed", "is_error": false, "mcp_server": {"name": "mcp-server"}}, "thought": null, "ppt": null, "browser": null}}</glm_block>',
+                    "phase": "other"
+                }
+            ]
+        },
+        {
+            "name": "天气查询工具调用",
+            "description": "模拟查询上海天气的工具调用",
+            "expected_tools": [
+                {
+                    "name": "search",
+                    "id": "call_qsn2jby8al",
+                    "arguments": {"queries": ["今天上海天气", "上海天气预报 今天"]}
+                }
+            ],
+            "data_sequence": [
+                {
+                    "edit_index": 16,
+                    "edit_content": '\n\n<glm_block >{"type": "mcp", "data": {"metadata": {"id": "call_qsn2jby8al", "name": "search", "arguments": "{\\"queries\\":[\\"今天上海天气\\", \\"',
+                    "phase": "tool_call"
+                },
+                {
+                    "edit_index": 183,
+                    "edit_content": '上海天气预报 今天\\"]}", "result": "", "display_result": "", "duration": "...", "status": "completed", "is_error": false, "mcp_server": {"name": "mcp-server"}}, "thought": null, "ppt": null, "browser": null}}</glm_block>',
+                    "phase": "tool_call"
+                }
+            ]
+        },
+        {
+            "name": "多工具调用序列",
+            "description": "模拟连续的多个工具调用",
+            "expected_tools": [
+                {
+                    "name": "search",
+                    "id": "call_001",
+                    "arguments": {"query": "北京天气"}
+                },
+                {
+                    "name": "visit_page",
+                    "id": "call_002",
+                    "arguments": {"url": "https://weather.com"}
+                }
+            ],
+            "data_sequence": [
+                {
+                    "edit_index": 0,
+                    "edit_content": '<glm_block >{"type": "mcp", "data": {"metadata": {"id": "call_001", "name": "search", "arguments": "{\\"query\\":\\"北京天气\\"}", "result": "", "status": "completed"}}, "thought": null}}</glm_block>',
+                    "phase": "tool_call"
+                },
+                {
+                    "edit_index": 200,
+                    "edit_content": '\n\n<glm_block >{"type": "mcp", "data": {"metadata": {"id": "call_002", "name": "visit_page", "arguments": "{\\"url\\":\\"https://weather.com\\"}", "result": "", "status": "completed"}}, "thought": null}}</glm_block>',
+                    "phase": "tool_call"
+                }
+            ]
+        }
+    ]
+
+    print(f"\n🧪 开始执行 {len(test_scenarios)} 个真实场景测试...")
+
+    # 执行每个测试场景
+    for i, scenario in enumerate(test_scenarios, 1):
+        print(f"\n{'='*60}")
+        print(f"测试 {i}: {scenario['name']}")
+        print(f"描述: {scenario['description']}")
+        print('='*60)
+
+        try:
+            # 创建新的处理器实例
+            handler = SSEToolHandler("test_chat_id", "GLM-4.5")
+
+            # 处理数据序列
+            all_chunks = []
+            for j, data in enumerate(scenario["data_sequence"]):
+                print(f"\n📦 处理数据块 {j+1}: phase={data['phase']}, edit_index={data['edit_index']}")
+
+                if data["phase"] == "tool_call":
+                    chunks = list(handler.process_tool_call_phase(data, is_stream=True))
+                else:
+                    chunks = list(handler.process_other_phase(data, is_stream=True))
+
+                all_chunks.extend(chunks)
+
+            # 提取工具调用信息
+            extracted_tools = extract_tool_calls(all_chunks)
+
+            # 验证结果
+            expected_tools = scenario["expected_tools"]
+
+            print(f"\n📊 验证结果:")
+            print(f"  期望工具数: {len(expected_tools)}")
+            print(f"  实际工具数: {len(extracted_tools)}")
+
+            # 详细验证每个工具
+            for k, expected_tool in enumerate(expected_tools):
+                if k < len(extracted_tools):
+                    actual_tool = extracted_tools[k]
+
+                    # 验证工具名称
+                    name_match = actual_tool["name"] == expected_tool["name"]
+                    # 验证工具ID
+                    id_match = actual_tool["id"] == expected_tool["id"]
+                    # 验证参数
+                    args_match = actual_tool["parsed_arguments"] == expected_tool["arguments"]
+
+                    if name_match and id_match and args_match:
+                        print(f"  ✅ 工具 {k+1}: {expected_tool['name']} - 验证通过")
+                        result.add_pass()
+                    else:
+                        error_details = []
+                        if not name_match:
+                            error_details.append(f"名称不匹配: 期望'{expected_tool['name']}', 实际'{actual_tool['name']}'")
+                        if not id_match:
+                            error_details.append(f"ID不匹配: 期望'{expected_tool['id']}', 实际'{actual_tool['id']}'")
+                        if not args_match:
+                            error_details.append(f"参数不匹配: 期望{expected_tool['arguments']}, 实际{actual_tool['parsed_arguments']}")
+
+                        error_msg = f"工具 {k+1} 验证失败: {'; '.join(error_details)}"
+                        print(f"  ❌ {error_msg}")
+                        result.add_fail(error_msg)
+                else:
+                    error_msg = f"缺少工具 {k+1}: {expected_tool['name']}"
+                    print(f"  ❌ {error_msg}")
+                    result.add_fail(error_msg)
+
+            # 显示提取的工具详情
+            if extracted_tools:
+                print(f"\n🔍 提取的工具详情:")
+                for tool in extracted_tools:
+                    print(f"  - {tool['name']}(id={tool['id']})")
+                    print(f"    参数: {tool['parsed_arguments']}")
+
+        except Exception as e:
+            error_msg = f"测试 {scenario['name']} 执行失败: {str(e)}"
+            print(f"❌ {error_msg}")
+            result.add_fail(error_msg)
+            logger.error(f"测试执行异常: {e}")
+
+    result.print_summary()
+    return result
+
+
+def test_edge_cases():
+    """测试边界情况和异常处理"""
+
+    result = TestResult("边界情况测试")
+
+    edge_cases = [
+        {
+            "name": "空内容处理",
+            "data": {"edit_index": 0, "edit_content": "", "phase": "tool_call"},
+            "should_pass": True
+        },
+        {
+            "name": "无效JSON处理",
+            "data": {"edit_index": 0, "edit_content": '<glm_block >{"invalid": json}}</glm_block>', "phase": "tool_call"},
+            "should_pass": True  # 应该优雅处理，不崩溃
+        },
+        {
+            "name": "不完整的glm_block",
+            "data": {"edit_index": 0, "edit_content": '<glm_block >{"type": "mcp", "data": {"metadata": {"id": "test"', "phase": "tool_call"},
+            "should_pass": True
+        },
+        {
+            "name": "超大edit_index",
+            "data": {"edit_index": 999999, "edit_content": "test", "phase": "tool_call"},
+            "should_pass": True
+        },
+        {
+            "name": "特殊字符处理",
+            "data": {"edit_index": 0, "edit_content": '<glm_block >{"type": "mcp", "data": {"metadata": {"id": "test", "name": "test", "arguments": "{\\"text\\":\\"测试\\u4e2d\\u6587\\"}"}}}</glm_block>', "phase": "tool_call"},
+            "should_pass": True
+        }
+    ]
+
+    print(f"\n🧪 开始执行 {len(edge_cases)} 个边界情况测试...")
+
+    for i, case in enumerate(edge_cases, 1):
+        print(f"\n📦 测试 {i}: {case['name']}")
+
+        try:
+            handler = SSEToolHandler("test_chat_id", "GLM-4.5")
+
+            # 处理数据
+            if case["data"]["phase"] == "tool_call":
+                chunks = list(handler.process_tool_call_phase(case["data"], is_stream=True))
+            else:
+                chunks = list(handler.process_other_phase(case["data"], is_stream=True))
+
+            # 检查是否按预期处理
+            if case["should_pass"]:
+                print(f"  ✅ 成功处理，生成 {len(chunks)} 个输出块")
+                result.add_pass()
+            else:
+                print(f"  ❌ 应该失败但成功了")
+                result.add_fail(f"{case['name']}: 应该失败但成功了")
+
+        except Exception as e:
+            if case["should_pass"]:
+                error_msg = f"{case['name']}: 意外异常 - {str(e)}"
+                print(f"  ❌ {error_msg}")
+                result.add_fail(error_msg)
+            else:
+                print(f"  ✅ 按预期失败: {str(e)}")
+                result.add_pass()
+
+    result.print_summary()
+    return result
+
+
+def test_performance():
+    """测试性能表现"""
+
+    result = TestResult("性能测试")
+
+    print(f"\n🚀 开始性能测试...")
+
+    # 测试大量小块数据的处理性能
+    handler = SSEToolHandler("test_chat_id", "GLM-4.5")
+
+    start_time = time.time()
+
+    # 模拟1000次小的编辑操作
+    for i in range(1000):
+        data = {
+            "edit_index": i * 5,
+            "edit_content": f"chunk_{i}",
+            "phase": "tool_call"
+        }
+        list(handler.process_tool_call_phase(data, is_stream=False))
+
+    end_time = time.time()
+    duration = end_time - start_time
+
+    print(f"⏱️ 处理1000次编辑操作耗时: {duration:.3f}秒")
+    print(f"📊 平均每次操作耗时: {duration * 1000 / 1000:.3f}毫秒")
+
+    # 性能基准：每次操作应该在1毫秒以内
+    if duration < 1.0:  # 1秒内完成1000次操作
+        print("✅ 性能测试通过")
+        result.add_pass()
+    else:
+        error_msg = f"性能测试失败: 耗时{duration:.3f}秒，超过1秒基准"
+        print(f"❌ {error_msg}")
+        result.add_fail(error_msg)
+
+    result.print_summary()
+    return result
+
+
+def test_argument_parsing():
+    """测试参数解析功能"""
+
+    result = TestResult("参数解析测试")
+
+    print(f"\n🧪 开始参数解析测试...")
+
+    handler = SSEToolHandler("test", "test")
+
+    test_cases = [
+        ('{"city": "北京"}', {"city": "北京"}),
+        ('{"city": "北京', {"city": "北京"}),  # 缺少闭合括号
+        ('{"city": "北京"', {"city": "北京"}),  # 缺少闭合括号但有引号
+        ('{\\"city\\": \\"北京\\"}', {"city": "北京"}),  # 转义的JSON
+        ('{}', {}),  # 空参数
+        ('null', {}),  # null参数
+        ('{"array": [1,2,3], "nested": {"key": "value"}}', {"array": [1,2,3], "nested": {"key": "value"}}),  # 复杂参数
+        ('{"url":"https://www.goo', {"url": "https://www.goo"}),  # 不完整的URL
+        ('', {}),  # 空字符串
+        ('{', {}),  # 只有开始括号
+    ]
+
+    for i, (input_str, expected) in enumerate(test_cases, 1):
+        try:
+            parsed_result = handler._parse_partial_arguments(input_str)
+            success = parsed_result == expected
+
+            if success:
+                print(f"✅ 测试 {i}: 解析成功")
+                result.add_pass()
+            else:
+                error_msg = f"测试 {i} 失败: 输入'{input_str[:30]}...', 期望{expected}, 实际{parsed_result}"
+                print(f"❌ {error_msg}")
+                result.add_fail(error_msg)
+
+        except Exception as e:
+            error_msg = f"测试 {i} 异常: 输入'{input_str[:30]}...', 错误: {str(e)}"
+            print(f"❌ {error_msg}")
+            result.add_fail(error_msg)
+
+    result.print_summary()
+    return result
+
+
+def run_all_tests():
+    """运行所有测试"""
+
+    print("🧪 SSE工具调用处理器优化测试套件")
+    print("="*60)
+
+    all_results = []
+
+    try:
+        # 运行真实场景测试
+        print("\n1️⃣ 真实场景测试")
+        all_results.append(test_real_world_scenarios())
+
+        # 运行边界情况测试
+        print("\n2️⃣ 边界情况测试")
+        all_results.append(test_edge_cases())
+
+        # 运行参数解析测试
+        print("\n3️⃣ 参数解析测试")
+        all_results.append(test_argument_parsing())
+
+        # 运行性能测试
+        print("\n4️⃣ 性能测试")
+        all_results.append(test_performance())
+
+        # 汇总结果
+        print("\n" + "="*60)
+        print("📊 测试汇总")
+        print("="*60)
+
+        total_passed = sum(r.passed for r in all_results)
+        total_failed = sum(r.failed for r in all_results)
+        total_tests = total_passed + total_failed
+
+        print(f"总测试数: {total_tests}")
+        print(f"✅ 通过: {total_passed}")
+        print(f"❌ 失败: {total_failed}")
+
+        if total_tests > 0:
+            success_rate = (total_passed / total_tests) * 100
+            print(f"📈 总体成功率: {success_rate:.1f}%")
+
+            if success_rate >= 90:
+                print("🎉 测试结果优秀！")
+            elif success_rate >= 70:
+                print("👍 测试结果良好")
+            else:
+                print("⚠️ 需要改进")
+
+        # 显示失败的测试
+        failed_tests = []
+        for result in all_results:
+            failed_tests.extend(result.errors)
+
+        if failed_tests:
+            print(f"\n🔍 失败测试详情:")
+            for i, error in enumerate(failed_tests, 1):
+                print(f"  {i}. {error}")
+
+        return total_failed == 0
+
+    except Exception as e:
+        print(f"❌ 测试套件执行失败: {e}")
+        traceback.print_exc()
+        return False
+
+
+if __name__ == "__main__":
+    success = run_all_tests()
+    exit(0 if success else 1)