# -*- coding: utf-8 -*- """ Claude Router CLI 服务适配器（统一版） 这个模块提供了与 AnyRouter/AgentRouter (Claude Code) 的集成，使其能够像 OpenAI API 一样被调用。 支持多个 Claude Code 共享平台，通过 Claude CLI 进行调用。 合并了原来的 anyrouter_service.py 和 agentrouter_service.py """ import os import subprocess import tempfile import json import time import re from typing import Dict, List, Any, Optional from pathlib import Path import logging from .stage_detector import StageDetector from .instruction_builder import InstructionBuilder from .guideline_manager import GuidelineManager logger = logging.getLogger(__name__) # 支持的路由器配置 ROUTER_CONFIGS = { 'anyrouter': { 'name': 'AnyRouter (Claude Code)', 'base_url': 'https://anyrouter.top', 'log_prefix': 'anyrouter', 'token_table': 'AnyRouterToken', 'env_token': 'ANTHROPIC_AUTH_TOKEN', 'env_base_url': 'ANTHROPIC_BASE_URL', }, 'agentrouter': { 'name': 'AgentRouter (Claude Code)', 'base_url': 'https://agentrouter.org/', 'log_prefix': 'agentrouter', 'token_table': 'AgentRouterToken', 'env_token': 'ANTHROPIC_API_KEY', 'env_base_url': 'ANTHROPIC_BASE_URL', } } class ClaudeRouterService: """Claude Router (Claude Code) CLI 服务适配器 统一的服务类，支持 AnyRouter 和 AgentRouter 实现 BaseAIProvider 接口，使其可以被 AIService 调用 """ def __init__(self, router_type: str = 'anyrouter', user_id: Optional[int] = None, verbose: bool = False, model: str = None): """初始化 Claude Router 服务 Args: router_type: 路由器类型 ('anyrouter' 或 'agentrouter') user_id: 用户ID，用于获取用户的 Token verbose: 是否显示详细输出（默认 False） model: 模型名称（可选，默认使用 claude-sonnet-4） """ # 验证路由器类型 if router_type not in ROUTER_CONFIGS: raise ValueError(f"不支持的路由器类型: {router_type}，支持: {list(ROUTER_CONFIGS.keys())}") self.router_type = router_type self.router_config = ROUTER_CONFIGS[router_type] # 项目根目录（ctf目录） # 从 claude_router_service.py 向上查找：claude_router_service.py -> core -> ai_driver -> services -> app -> ctf # 路径: ctf/app/services/ai_driver/core/claude_router_service.py # parent(1): core, parent(2): ai_driver, parent(3): services, parent(4): app, parent(5): ctf self.project_root = Path(__file__).parent.parent.parent.parent.parent # ge10 工作目录，统一使用 ctf/ge10 self.ge10_dir = self.project_root / "ge10" if not self.ge10_dir.exists(): raise Exception(f"ge10 目录不存在: {self.ge10_dir}") # 用户ID（用于获取Token） self.user_id = user_id # 日志回调函数 self.log_callback = None # 超时恢复标记 self.timeout_recovered = False self.timeout_warning = None # 是否显示详细输出 self.verbose = verbose # 任务ID（用于日志文件隔离） self.task_id = None # 当前阶段 self._current_stage = 0 # BaseAIProvider 兼容属性 self.provider_type = router_type self.provider_name = self.router_config['name'] self.model = model or 'claude-sonnet-4' self.base_url = self.router_config['base_url'] # 验证 Claude CLI 是否可用 self._verify_claude_cli() logger.info(f"{self.provider_name} 服务初始化完成") def _verify_claude_cli(self): """验证 Claude CLI 是否可用""" try: result = subprocess.run( ['claude', '--version'], capture_output=True, text=True, timeout=10 ) if result.returncode != 0: raise Exception(f"Claude CLI 不可用: {result.stderr}") print(f"✅ Claude CLI 版本: {result.stdout.strip()}") except FileNotFoundError: raise Exception("未找到 claude 命令，请确保 Claude Code 已安装: npm install -g @anthropic-ai/claude-code") except subprocess.TimeoutExpired: raise Exception("Claude CLI 验证超时") except Exception as e: raise Exception(f"Claude CLI 验证失败: {str(e)}") def set_log_callback(self, callback): """设置日志回调函数""" self.log_callback = callback def _get_user_token(self) -> Optional[str]: """获取用户的 Router Token 从 AIProviderConfig 获取（统一存储方式） """ if not self.user_id: return None try: from app.models.database import AIProviderConfig # 优先读取用户的配置 config = AIProviderConfig.get_user_config( user_id=self.user_id, provider_type=self.router_type ) if config: api_key = config.get_api_key() if api_key: print(f"🔑 从 AIProviderConfig 获取到用户 {self.user_id} 的 {self.provider_name} Token") return api_key # 如果没有用户配置，尝试读取全局配置（user_id 为 None） global_config = AIProviderConfig.query.filter_by( user_id=None, provider_type=self.router_type, is_active=True, is_default=True ).first() if global_config: api_key = global_config.get_api_key() if api_key: print(f"🔑 从 AIProviderConfig 获取到全局 {self.provider_name} Token") return api_key print(f"⚠️ 警告: 用户 {self.user_id} 未设置 {self.provider_name} Token") return None except Exception as e: logger.warning(f"获取 {self.provider_name} Token 失败: {e}") return None def _log(self, level: str, message: str): """记录日志""" if self.log_callback: self.log_callback(level, message) def generate_ctf_challenge( self, language: str, vulnerabilities: List[str], scene: Dict, difficulty: str, extra_requirements: str = '', category_id: str = 'web', form_data: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """生成 CTF 题目（Claude Router 专用接口） 这个方法直接调用 Claude CLI 来生成题目，不经过 API Args: language: 编程语言 vulnerabilities: 漏洞列表 scene: 场景 difficulty: 难度 extra_requirements: 额外要求 category_id: 方向ID form_data: 完整的表单数据（可选，用于传递所有表单字段） task_id: 任务ID（可选，用于创建独立工作目录避免多任务冲突） """ import datetime try: # 存储 task_id 用于日志文件隔离 self.task_id = task_id # 重置阶段状态 self._current_stage = 0 # 设置日志文件路径（包含 task_id 以确保唯一性） log_prefix = self.router_config['log_prefix'] log_dir = self.project_root / 'logs' log_dir.mkdir(exist_ok=True) # 日志文件名包含 task_id 以确保唯一性（如果提供了 task_id） if task_id: # 使用 task_id 的一部分（去掉时间戳前缀，只保留 UUID 部分）来创建唯一文件名 task_suffix = task_id.split('-')[-1][:8] if '-' in task_id else task_id[:8] log_file = log_dir / f"{log_prefix}_{task_suffix}_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" else: # 如果没有 task_id，使用时间戳和进程ID确保唯一性 import os log_file = log_dir / f"{log_prefix}_{os.getpid()}_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" try: # 设置日志文件路径到 generation_statuses（按 task_id 索引） from app.routes.generator.utils import generation_statuses, generation_lock if task_id: with generation_lock: if task_id not in generation_statuses: generation_statuses[task_id] = {} generation_statuses[task_id]["log_file"] = str(log_file) generation_statuses[task_id]["log_position"] = 0 else: # 向后兼容：如果没有 task_id，使用全局 generation_status from app.routes.generator.utils import generation_status generation_status["log_file"] = str(log_file) generation_status["log_position"] = 0 print(f"📝 日志文件: {log_file} (task_id: {task_id})", flush=True) except Exception as e: print(f"⚠️ 设置日志文件失败: {e}", flush=True) # 构建指令（传递 form_data 以支持动态字段） instruction = self._build_instruction(language, vulnerabilities, scene, difficulty, extra_requirements, category_id, form_data) print("\n" + "=" * 60) print(f"📝 传递给 {self.provider_name} 的指令:") print("=" * 60) print(instruction) print("=" * 60 + "\n") # 根据难度复制对应的 guideline 文件 self._setup_guideline_for_difficulty(difficulty) # 调用 Claude CLI result = self._call_claude_cli(instruction, log_file) return result except Exception as e: error_msg = f"{self.provider_name} 生成失败: {str(e)}" logger.error(error_msg) self._log('error', error_msg) return { "status": "error", "error": error_msg } def _build_instruction(self, language: str, vulnerabilities: List[str], scene: Dict, difficulty: str, extra_requirements: str = '', category_id: str = 'web', form_data: Optional[Dict[str, Any]] = None) -> str: """构建传递给 Claude CLI 的指令（委托给 InstructionBuilder）""" # 如果有 form_data，使用动态指令构建 if form_data: return InstructionBuilder.build_dynamic_instruction( category_id, form_data, language, vulnerabilities, scene, difficulty, extra_requirements ) else: # 回退到简单指令构建（向后兼容） return InstructionBuilder.build_claude_instruction( language, vulnerabilities, scene, difficulty, extra_requirements ) def _setup_guideline_for_difficulty(self, difficulty: str): """根据难度设置对应的 guideline 文件（委托给 GuidelineManager）""" GuidelineManager.setup_for_claude(self.ge10_dir, difficulty, verbose=True) def _call_claude_cli(self, instruction: str, log_file: Path = None) -> Dict[str, Any]: """调用 Claude CLI 执行指令""" # 创建临时指令文件 with tempfile.NamedTemporaryFile( mode='w', suffix='.md', delete=False, encoding='utf-8' ) as f: f.write(instruction) instruction_file = f.name try: # 获取用户 Token token = self._get_user_token() # 设置环境变量（根据路由器类型使用不同的环境变量名） env = os.environ.copy() if token: env_token_name = self.router_config.get('env_token', 'ANTHROPIC_AUTH_TOKEN') env_base_url_name = self.router_config.get('env_base_url', 'ANTHROPIC_BASE_URL') env[env_token_name] = token env[env_base_url_name] = self.base_url print(f"✅ 已设置用户 {self.user_id} 的 {self.provider_name} Token", flush=True) # 显示 Token 前后几位用于调试 if len(token) > 10: print(f"🔑 Token ({env_token_name}): {token[:4]}...{token[-4:]}", flush=True) print(f"📡 Base URL ({env_base_url_name}): {self.base_url}", flush=True) # 构建命令 cmd = [ 'claude', '-p', # print mode '--verbose', '--output-format', 'stream-json', '--model', self.model, '--tools', 'default', '--permission-mode', 'bypassPermissions', ] # 从文件读取指令 with open(instruction_file, 'r', encoding='utf-8') as f: prompt_content = f.read() print(f"🚀 执行命令: {' '.join(cmd)} ...", flush=True) print(f"📄 Claude CLI 将自动读取 CLAUDE.md 作为 guideline", flush=True) # 获取或创建日志文件 if not log_file: import datetime log_prefix = self.router_config['log_prefix'] log_dir = self.project_root / 'logs' log_dir.mkdir(exist_ok=True) log_file = log_dir / f"{log_prefix}_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" # 创建可读格式的日志文件 readable_log_file = Path(str(log_file).replace('.txt', '_readable.txt')) # 记录开始时间 start_time = time.time() print(f"🔄 正在启动 Claude CLI...", flush=True) # 执行命令 process = subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, env=env, cwd=str(self.ge10_dir), bufsize=1 ) # 发送指令 process.stdin.write(prompt_content) process.stdin.close() # 读取输出 output_lines = [] current_stage = 0 output_dir = None generation_complete = False # 标记是否检测到完成标记 accumulated_text = "" # 累积的文本，用于检测完成标记 # 使用统一的日志格式器（用于可读日志） from app.services.ai.core.log_formatter import UnifiedLogWriter, LogLevel with open(log_file, 'w', encoding='utf-8') as log_f, \ UnifiedLogWriter(str(readable_log_file), self.provider_name) as readable_writer: for line in process.stdout: output_lines.append(line) log_f.write(line) log_f.flush() # 解析 JSON 输出 try: data = json.loads(line.strip()) msg_type = data.get('type') # 提取可读内容 if msg_type == 'assistant': content = data.get('message', {}).get('content', []) for item in content: item_type = item.get('type') if item_type == 'text': text = item.get('text', '') accumulated_text += text # 累积文本用于检测完成标记 readable_writer.write_raw(text + '\n') # 实时输出到控制台 print(text, end='', flush=True) # 检测生成完成标记（优先检测，因为这是最可靠的完成信号） # 使用累积的文本检测，因为完成标记可能跨多行 if StageDetector.detect_completion(accumulated_text): print(f"\n✅ 检测到生成完成标记 [CTF_GENERATION_COMPLETE]", flush=True) readable_writer.write_success("检测到生成完成标记，所有阶段已完成") # 获取总阶段数，标记所有阶段为完成（仅记录到日志） total_stages = StageDetector.get_total_stages() if total_stages > 0: # 标记所有阶段为完成（仅记录到日志，不再上报进度） for stage_num in range(total_stages): if stage_num >= current_stage: stage_name = StageDetector.get_stage_name(stage_num) readable_writer.write_stage(stage_num, stage_name, "完成") # 设置完成标志并跳出内层循环 generation_complete = True break # 如果已检测到完成标记，跳出外层循环 if generation_complete: break # 检测阶段并使用统一的阶段转换处理逻辑 new_stage = self._detect_stage(text) if new_stage is not None and new_stage > current_stage: current_stage = StageDetector.handle_stage_transition( current_stage=current_stage, new_stage=new_stage, output_text=text, logger=None ) # 记录阶段信息（使用统一格式） stage_name = StageDetector.get_stage_name(current_stage) readable_writer.write_stage(current_stage, stage_name, "开始") # 检测输出目录 if not output_dir: output_dir = self._detect_output_dir(text) elif item_type == 'tool_use': # 工具调用（使用统一格式） tool_name = item.get('name', '') tool_input = item.get('input', {}) if tool_name == 'Bash': cmd = tool_input.get('command', '') readable_writer.write_tool(tool_name, f"命令: {cmd}") elif tool_name in ('Read', 'Write', 'Edit'): file_path = tool_input.get('file_path', '') readable_writer.write_tool(tool_name, f"文件: {file_path}") else: readable_writer.write_tool(tool_name, str(tool_input)) elif msg_type == 'user': # 工具执行结果（使用统一格式） content = data.get('message', {}).get('content', []) for item in content: if item.get('type') == 'tool_result': result_content = item.get('content', '') # 尝试从上下文获取工具名 readable_writer.write_info(f"工具执行结果: {result_content}") elif msg_type == 'result': # 工具调用结果（旧格式） tool_name = data.get('tool_name', '') if tool_name in ['write_file', 'create_file']: file_path = data.get('tool_input', {}).get('path', '') if file_path and not output_dir: output_dir = self._detect_output_dir(file_path) # 如果已检测到完成标记，跳出外层循环 if generation_complete: break except json.JSONDecodeError: # 非 JSON 行，直接写入可读日志（原始格式） readable_writer.write_raw(line) # 如果已检测到完成标记，跳出最外层循环 if generation_complete: break # 等待进程完成 process.wait(timeout=1800) # 30分钟超时 # 记录执行时间（使用统一格式） elapsed_time = time.time() - start_time readable_writer.write_success(f"执行完成，耗时: {elapsed_time:.2f} 秒") print(f"\n⏱️ 执行时间: {elapsed_time:.2f} 秒", flush=True) # 检查执行结果 if process.returncode != 0: # 检查是否有输出目录生成 if output_dir and Path(output_dir).exists(): print(f"⚠️ CLI 返回非零状态码，但检测到输出目录: {output_dir}", flush=True) else: raise Exception(f"Claude CLI 执行失败，返回码: {process.returncode}") # 查找输出目录 if not output_dir: output_dir = self._find_latest_output_dir(category_id) if output_dir: print(f"✅ 题目生成完成，输出目录: {output_dir}", flush=True) return { "status": "success", "output_dir": output_dir, "elapsed_time": elapsed_time } else: raise Exception("未找到输出目录") except subprocess.TimeoutExpired: process.kill() raise Exception("Claude CLI 执行超时（超过 30 分钟）") except Exception as e: raise Exception(f"Claude CLI 执行异常: {str(e)}") finally: # 清理临时文件 try: os.unlink(instruction_file) except: pass def _detect_stage(self, text: str) -> int: """从文本中检测当前阶段（委托给 StageDetector）""" stage = StageDetector.detect_stage(text) return stage if stage is not None else 0 def _detect_output_dir(self, text: str) -> Optional[str]: """从文本中检测输出目录（委托给 StageDetector）""" return StageDetector.detect_output_dir(text, str(self.ge10_dir)) def _find_latest_output_dir(self, category_id: str = 'web') -> Optional[str]: """查找最新的输出目录（委托给 StageDetector）""" return StageDetector.find_latest_output_dir(str(self.ge10_dir), category_id) def chat(self, messages: List[Dict], tools: List[Dict] = None, **kwargs) -> Dict: """聊天接口（兼容 BaseAIProvider） 注意：Claude CLI 模式下，这个方法用于继续对话，返回文本响应 Args: messages: 消息列表 tools: 工具定义（可选） **kwargs: 额外参数，支持 workspace_dir 指定工作目录 """ # 获取工作目录（如果提供） workspace_dir = kwargs.get('workspace_dir') # 简化实现：提取最后一条用户消息和系统提示 user_message = "" system_prompt = "" for msg in messages: if msg.get('role') == 'system': system_prompt = msg.get('content', '') elif msg.get('role') == 'user': content = msg.get('content', '') if isinstance(content, str): user_message = content elif isinstance(content, list): for item in content: if isinstance(item, dict) and item.get('type') == 'text': user_message = item.get('text', '') break if not user_message: return { "content": "", "tool_calls": None } # 调用 Claude CLI 对话模式（非流式） response_text = self._call_claude_chat(user_message, workspace_dir=workspace_dir, system_prompt=system_prompt) return { "content": response_text, "tool_calls": None } def chat_stream(self, messages: List[Dict], tools: List[Dict] = None, **kwargs): """流式聊天接口（生成器） Args: messages: 消息列表 tools: 工具定义（可选） **kwargs: 额外参数，支持 workspace_dir 指定工作目录 Yields: dict: {'type': 'content', 'content': '...'} 或 {'type': 'tool', 'name': '...'} """ # 获取工作目录（如果提供） workspace_dir = kwargs.get('workspace_dir') # 简化实现：提取最后一条用户消息和系统提示 user_message = "" system_prompt = "" for msg in messages: if msg.get('role') == 'system': system_prompt = msg.get('content', '') elif msg.get('role') == 'user': content = msg.get('content', '') if isinstance(content, str): user_message = content elif isinstance(content, list): for item in content: if isinstance(item, dict) and item.get('type') == 'text': user_message = item.get('text', '') break if not user_message: return # 调用 Claude CLI 流式对话模式 yield from self._call_claude_chat_stream(user_message, workspace_dir=workspace_dir, system_prompt=system_prompt) def _call_claude_chat(self, instruction: str, workspace_dir: str = None, system_prompt: str = None) -> str: """调用 Claude CLI 进行对话（返回文本响应） Args: instruction: 用户指令 workspace_dir: 工作目录（如果提供，将在该目录下执行） system_prompt: 系统提示词（如果提供，将添加到指令前面） """ # 确定工作目录 if workspace_dir: cwd = Path(workspace_dir) print(f"📂 使用题目工作目录: {cwd}", flush=True) else: cwd = self.ge10_dir # 构建完整指令（包含系统提示词） full_instruction = instruction if system_prompt: full_instruction = f"{system_prompt}\n\n---\n\n用户指令：{instruction}" # 创建临时指令文件 with tempfile.NamedTemporaryFile( mode='w', suffix='.md', delete=False, encoding='utf-8' ) as f: f.write(full_instruction) instruction_file = f.name try: # 获取用户 Token print(f"🔍 正在获取用户 {self.user_id} 的 {self.router_type} Token...", flush=True) token = self._get_user_token() # 设置环境变量（根据路由器类型使用不同的环境变量名） env = os.environ.copy() if token: env_token_name = self.router_config.get('env_token', 'ANTHROPIC_AUTH_TOKEN') env_base_url_name = self.router_config.get('env_base_url', 'ANTHROPIC_BASE_URL') env[env_token_name] = token env[env_base_url_name] = self.base_url # 显示 Token 前后几位用于调试 if len(token) > 10: print(f"✅ 已设置用户 {self.user_id} 的 {self.provider_name} Token: {token[:4]}...{token[-4:]}", flush=True) else: print(f"✅ 已设置用户 {self.user_id} 的 {self.provider_name} Token", flush=True) print(f"📡 Base URL ({env_base_url_name}): {self.base_url}", flush=True) else: print(f"⚠️ 未获取到用户 Token，将使用默认配置", flush=True) # 构建命令 cmd = [ 'claude', '-p', # print mode '--verbose', '--output-format', 'stream-json', '--model', self.model, '--tools', 'default', '--permission-mode', 'bypassPermissions', ] # 从文件读取指令 with open(instruction_file, 'r', encoding='utf-8') as f: prompt_content = f.read() print(f"🚀 执行命令: {' '.join(cmd)} ...", flush=True) print(f"🔄 正在启动 Claude CLI...", flush=True) # 记录开始时间 start_time = time.time() # 执行命令 process = subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, env=env, cwd=str(cwd), bufsize=1 ) # 发送指令 process.stdin.write(prompt_content) process.stdin.close() # 读取输出并提取文本响应 response_text = [] for line in process.stdout: # 解析 JSON 输出 try: data = json.loads(line.strip()) # 提取文本内容 if data.get('type') == 'assistant': content = data.get('message', {}).get('content', []) for item in content: if item.get('type') == 'text': text = item.get('text', '') response_text.append(text) print(text, end='', flush=True) # 工具调用结果也记录 elif data.get('type') == 'result': tool_name = data.get('tool_name', '') if tool_name: response_text.append(f"\n🔧 执行工具: {tool_name}\n") except json.JSONDecodeError: # 非 JSON 行 pass # 等待进程完成 process.wait(timeout=600) # 10分钟超时 elapsed_time = time.time() - start_time print(f"\n⏱️ 执行时间: {elapsed_time:.2f} 秒", flush=True) return ''.join(response_text) except subprocess.TimeoutExpired: process.kill() return "❌ Claude CLI 执行超时" except Exception as e: return f"❌ Claude CLI 执行异常: {str(e)}" finally: # 清理临时文件 try: os.unlink(instruction_file) except: pass def _call_claude_chat_stream(self, instruction: str, workspace_dir: str = None, system_prompt: str = None): """流式调用 Claude CLI 进行对话（生成器） Args: instruction: 用户指令 workspace_dir: 工作目录（如果提供，将在该目录下执行，并使用 --continue 继续对话） system_prompt: 系统提示词（如果提供，将添加到指令前面） Yields: dict: {'type': 'content', 'content': '...'} 或 {'type': 'tool', 'name': '...', 'result': '...'} """ # 确定工作目录 if workspace_dir: cwd = Path(workspace_dir) print(f"📂 使用题目工作目录: {cwd}", flush=True) else: cwd = self.ge10_dir # 检查是否有之前的对话（用于决定是否使用 --continue） # Claude CLI 会自动在 ~/.claude/projects/ 目录下按项目路径保存会话 has_previous_session = False if workspace_dir: # 检查 Claude CLI 的会话目录是否存在 # 目录名格式：路径中的 / 替换为 - project_name = str(cwd).replace('/', '-') claude_project_dir = Path.home() / '.claude' / 'projects' / project_name if claude_project_dir.exists(): # 检查是否有会话文件（排除 agent-*.jsonl） session_files = [f for f in claude_project_dir.glob('*.jsonl') if not f.name.startswith('agent-')] has_previous_session = len(session_files) > 0 if has_previous_session: print(f"📜 检测到之前的 Claude CLI 会话，将使用 --continue 继续", flush=True) else: print(f"🆕 这是新的对话", flush=True) try: # 获取用户 Token token = self._get_user_token() # 设置环境变量 env = os.environ.copy() if token: env_token_name = self.router_config.get('env_token', 'ANTHROPIC_AUTH_TOKEN') env_base_url_name = self.router_config.get('env_base_url', 'ANTHROPIC_BASE_URL') env[env_token_name] = token env[env_base_url_name] = self.base_url # 构建命令 cmd = [ 'claude', '-p', '--verbose', '--output-format', 'stream-json', '--model', self.model, '--tools', 'default', '--permission-mode', 'bypassPermissions', ] # 如果有之前的对话，使用 --continue 继续 if has_previous_session: cmd.append('--continue') # 如果有系统提示词且是新对话，添加系统提示词 if system_prompt and not has_previous_session: cmd.extend(['--system-prompt', system_prompt]) # 直接将指令作为参数传递 cmd.append(instruction) print(f"🚀 执行命令: {' '.join(cmd[:8])} ...", flush=True) # 执行命令 process = subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, env=env, cwd=str(cwd), bufsize=1 ) # 关闭 stdin（不需要输入） process.stdin.close() # 流式读取输出 for line in process.stdout: try: data = json.loads(line.strip()) # 提取文本内容 if data.get('type') == 'assistant': content = data.get('message', {}).get('content', []) for item in content: if item.get('type') == 'text': text = item.get('text', '') if text: yield {'type': 'content', 'content': text} # 工具调用 elif data.get('type') == 'result': tool_name = data.get('tool_name', '') if tool_name: yield {'type': 'tool', 'name': tool_name} except json.JSONDecodeError: pass # 等待进程完成 process.wait(timeout=600) # 发送完成信号 yield {'type': 'finish'} except subprocess.TimeoutExpired: process.kill() yield {'type': 'error', 'message': 'Claude CLI 执行超时'} except Exception as e: yield {'type': 'error', 'message': str(e)} # 兼容性别名 class AnyRouterService(ClaudeRouterService): """AnyRouter 服务（兼容性别名）""" def __init__(self, user_id: Optional[int] = None, verbose: bool = False, model: str = None): super().__init__(router_type='anyrouter', user_id=user_id, verbose=verbose, model=model) class AgentRouterService(ClaudeRouterService): """AgentRouter 服务（兼容性别名）""" def __init__(self, user_id: Optional[int] = None, verbose: bool = False, model: str = None): super().__init__(router_type='agentrouter', user_id=user_id, verbose=verbose, model=model)