| """ |
| 命令行界面模块 |
| 提供友好的用户交互界面 |
| """ |
| |
| import asyncio |
| import click |
| import logging |
| import os |
| from typing import Optional |
| from rich.console import Console |
| from rich.panel import Panel |
| from rich.text import Text |
| from rich.prompt import Prompt, Confirm |
| from rich.logging import RichHandler |
| from rich.table import Table |
| from rich.progress import Progress, SpinnerColumn, TextColumn |
| |
| from ..core.agent import AgentCore, ThinkingMode |
| from ..mcp.integration import MCPToolIntegration |
| |
| |
| class CLIInterface: |
| """命令行界面类""" |
| |
| def __init__(self): |
| self.console = Console() |
| self.agent: Optional[AgentCore] = None |
| self.mcp_integration: Optional[MCPToolIntegration] = None |
| self.setup_logging() |
| |
| def setup_logging(self): |
| """设置日志""" |
| logging.basicConfig( |
| level=logging.INFO, |
| format="%(message)s", |
| datefmt="[%X]", |
| handlers=[RichHandler(console=self.console, rich_tracebacks=True)] |
| ) |
| |
| async def initialize(self, api_key: Optional[str] = None): |
| """初始化Agent和MCP集成""" |
| try: |
| with Progress( |
| SpinnerColumn(), |
| TextColumn("[progress.description]{task.description}"), |
| console=self.console, |
| ) as progress: |
| task = progress.add_task("初始化Claude Agent...", total=None) |
| |
| # 初始化Agent核心 |
| self.agent = AgentCore(api_key=api_key) |
| progress.update(task, description="初始化MCP工具集成...") |
| |
| # 初始化MCP集成 |
| self.mcp_integration = MCPToolIntegration(self.agent) |
| await self.mcp_integration.setup_default_tools() |
| await self.mcp_integration.enhance_agent_with_tools() |
| |
| progress.update(task, description="初始化完成!", completed=True) |
| |
| self.console.print("[green]✓[/green] Claude Agent 初始化完成!") |
| |
| except Exception as e: |
| self.console.print(f"[red]✗[/red] 初始化失败: {str(e)}") |
| self.agent = None |
| self.mcp_integration = None |
| |
| def show_welcome(self): |
| """显示欢迎信息""" |
| welcome_text = Text() |
| welcome_text.append("Claude Agent", style="bold cyan") |
| welcome_text.append(" - 智能命令行助手\n", style="cyan") |
| welcome_text.append("支持交互模式和YOLO自主模式\n", style="dim") |
| welcome_text.append("输入 'help' 查看帮助，'quit' 退出", style="dim") |
| |
| panel = Panel( |
| welcome_text, |
| title="欢迎", |
| border_style="cyan", |
| padding=(1, 2) |
| ) |
| self.console.print(panel) |
| |
| def show_help(self): |
| """显示帮助信息""" |
| help_table = Table(title="命令帮助") |
| help_table.add_column("命令", style="cyan") |
| help_table.add_column("描述", style="white") |
| |
| help_table.add_row("help", "显示此帮助信息") |
| help_table.add_row("mode", "切换思考模式 (interactive/yolo)") |
| help_table.add_row("history", "显示对话历史") |
| help_table.add_row("clear", "清空对话历史") |
| help_table.add_row("tools", "显示可用的MCP工具") |
| help_table.add_row("status", "显示系统状态") |
| help_table.add_row("quit", "退出程序") |
| |
| self.console.print(help_table) |
| |
| def show_status(self): |
| """显示系统状态""" |
| status_table = Table(title="系统状态") |
| status_table.add_column("项目", style="cyan") |
| status_table.add_column("状态", style="white") |
| |
| if not self.agent: |
| status_table.add_row("思考模式", "[red]Agent未初始化[/red]") |
| status_table.add_row("对话历史", "[red]Agent未初始化[/red]") |
| else: |
| mode_text = "YOLO模式" if self.agent.thinking_mode == ThinkingMode.YOLO else "交互模式" |
| status_table.add_row("思考模式", mode_text) |
| status_table.add_row("对话历史", f"{len(self.agent.conversation_history)} 条消息") |
| |
| if not self.mcp_integration: |
| status_table.add_row("MCP工具", "[red]MCP服务未连接[/red]") |
| else: |
| status_table.add_row("MCP工具", f"{len(self.mcp_integration.tool_manager.tools)} 个可用") |
| |
| self.console.print(status_table) |
| |
| def show_tools(self): |
| """显示可用工具""" |
| if not self.mcp_integration: |
| self.console.print("[red]MCP服务未连接[/red]") |
| return |
| |
| tools = self.mcp_integration.tool_manager.get_available_tools() |
| |
| if not tools: |
| self.console.print("[yellow]暂无可用的MCP工具[/yellow]") |
| return |
| |
| tools_table = Table(title="可用的MCP工具") |
| tools_table.add_column("工具名称", style="cyan") |
| tools_table.add_column("描述", style="white") |
| |
| for tool_name in tools: |
| tool_info = self.mcp_integration.tool_manager.get_tool_info(tool_name) |
| description = tool_info['description'] if tool_info else "无描述" |
| tools_table.add_row(tool_name, description) |
| |
| self.console.print(tools_table) |
| |
| def show_history(self): |
| """显示对话历史""" |
| if not self.agent: |
| self.console.print("[red]Agent未初始化[/red]") |
| return |
| |
| history = self.agent.get_conversation_history() |
| |
| if not history: |
| self.console.print("[yellow]暂无对话历史[/yellow]") |
| return |
| |
| for i, message in enumerate(history[-10:], 1): # 只显示最后10条 |
| role_style = "blue" if message["role"] == "user" else "green" |
| role_text = "用户" if message["role"] == "user" else "助手" |
| |
| content = message["content"] |
| if len(content) > 200: |
| content = content[:200] + "..." |
| |
| self.console.print(f"[{role_style}]{i}. {role_text}:[/{role_style}] {content}") |
| |
| async def switch_mode(self): |
| """切换思考模式""" |
| if not self.agent: |
| self.console.print("[red]Agent未初始化[/red]") |
| return |
| |
| current_mode = self.agent.thinking_mode |
| |
| if current_mode == ThinkingMode.INTERACTIVE: |
| new_mode = ThinkingMode.YOLO |
| mode_text = "YOLO自主模式" |
| else: |
| new_mode = ThinkingMode.INTERACTIVE |
| mode_text = "交互模式" |
| |
| if Confirm.ask(f"确定切换到{mode_text}吗?"): |
| self.agent.set_thinking_mode(new_mode) |
| self.console.print(f"[green]已切换到{mode_text}[/green]") |
| else: |
| self.console.print("[yellow]已取消切换[/yellow]") |
| |
| async def process_command(self, command: str) -> bool: |
| """处理特殊命令，返回True表示继续，False表示退出""" |
| command = command.lower().strip() |
| |
| if command == "quit" or command == "exit": |
| return False |
| |
| elif command == "help": |
| self.show_help() |
| |
| elif command == "status": |
| self.show_status() |
| |
| elif command == "tools": |
| self.show_tools() |
| |
| elif command == "history": |
| self.show_history() |
| |
| elif command == "clear": |
| if not self.agent: |
| self.console.print("[red]Agent未初始化[/red]") |
| return True |
| |
| if Confirm.ask("确定清空对话历史吗?"): |
| self.agent.clear_history() |
| self.console.print("[green]对话历史已清空[/green]") |
| |
| elif command == "mode": |
| await self.switch_mode() |
| |
| else: |
| # 不是特殊命令，交给Agent处理 |
| await self.process_user_input(command) |
| |
| return True |
| |
| async def process_user_input(self, user_input: str): |
| """处理用户输入""" |
| if not self.agent: |
| self.console.print("[red]Agent未初始化[/red]") |
| return |
| |
| try: |
| # 根据模式显示不同的处理提示 |
| if self.agent.thinking_mode == ThinkingMode.YOLO: |
| # YOLO模式的详细进度显示 |
| with Progress( |
| SpinnerColumn(), |
| TextColumn("[progress.description]{task.description}"), |
| console=self.console, |
| ) as progress: |
| # 创建主任务 |
| main_task = progress.add_task("启动YOLO自主思考模式...", total=None) |
| |
| # 创建一个日志处理器来捕获思考过程 |
| import logging |
| from rich.logging import RichHandler |
| |
| # 临时创建一个日志捕获器 |
| class ProgressLogHandler(logging.Handler): |
| def __init__(self, progress_task, progress_obj): |
| super().__init__() |
| self.progress_task = progress_task |
| self.progress_obj = progress_obj |
| |
| def emit(self, record): |
| # 捕获所有包含思考过程标识符的日志 |
| thinking_indicators = [ |
| "🧠", "📋", "🔍", "✅", "🎯", "🔧", "📊", |
| "💡", "⚡", "🎉", "❌", "🔄", "⚠️" |
| ] |
| if any(indicator in record.msg for indicator in thinking_indicators): |
| # 清理消息，移除多余的格式 |
| clean_msg = record.msg.strip() |
| self.progress_obj.update(self.progress_task, description=clean_msg) |
| |
| # 设置日志捕获 |
| agent_logger = logging.getLogger('claude_agent.core.agent') |
| progress_handler = ProgressLogHandler(main_task, progress) |
| agent_logger.addHandler(progress_handler) |
| |
| try: |
| # 处理用户输入 |
| response = await self.agent.process_user_input(user_input) |
| |
| # 如果有MCP工具调用，处理工具调用 |
| enhanced_response = await self.mcp_integration.process_tool_calls_in_response(response) |
| |
| progress.update(main_task, description="思考完成!", completed=True) |
| |
| finally: |
| # 清理日志处理器 |
| agent_logger.removeHandler(progress_handler) |
| |
| else: |
| # 交互模式的简单显示 |
| with Progress( |
| SpinnerColumn(), |
| TextColumn("[progress.description]{task.description}"), |
| console=self.console, |
| ) as progress: |
| task = progress.add_task("处理中...", total=None) |
| |
| # 处理用户输入 |
| response = await self.agent.process_user_input(user_input) |
| |
| # 如果有MCP工具调用，处理工具调用 |
| enhanced_response = await self.mcp_integration.process_tool_calls_in_response(response) |
| |
| # 显示响应 |
| self.console.print("\n[green]助手:[/green]") |
| self.console.print(Panel(enhanced_response, border_style="green")) |
| return enhanced_response |
| |
| except Exception as e: |
| self.console.print(f"[red]处理请求时出错: {str(e)}[/red]") |
| return None |
| |
| async def run_interactive_loop(self): |
| """运行交互循环""" |
| self.show_welcome() |
| |
| while True: |
| try: |
| # 获取用户输入 |
| user_input = Prompt.ask("\n[cyan]你[/cyan]", console=self.console) |
| |
| if not user_input.strip(): |
| continue |
| |
| # 处理命令或输入 |
| should_continue = await self.process_command(user_input) |
| if not should_continue: |
| break |
| |
| except KeyboardInterrupt: |
| if Confirm.ask("\n确定要退出吗?"): |
| break |
| else: |
| self.console.print("继续...") |
| continue |
| except EOFError: |
| break |
| except Exception as e: |
| self.console.print(f"[red]发生未知错误: {str(e)}[/red]") |
| continue |
| |
| async def shutdown(self): |
| """关闭资源""" |
| if self.mcp_integration: |
| await self.mcp_integration.shutdown() |
| self.console.print("[yellow]再见![/yellow]") |
| |
| |
| @click.command() |
| @click.option('--api-key', '-k', help='Claude API密钥') |
| @click.option('--interactive/--no-interactive', default=True, help='是否启动交互模式') |
| @click.option('--mode', type=click.Choice(['interactive', 'yolo']), default='interactive', help='思考模式') |
| @click.argument('message', required=False) |
| def main(api_key: Optional[str], interactive: bool, mode: str, message: Optional[str]): |
| """Claude Agent 命令行工具""" |
| |
| async def run(): |
| cli = CLIInterface() |
| |
| try: |
| # 从环境变量获取API密钥 |
| api_key_final = api_key or os.getenv('CLAUDE_API_KEY') or os.getenv('ANTHROPIC_API_KEY') |
| if not api_key_final: |
| click.echo("警告: 未提供Claude API密钥，将使用默认配置") |
| |
| # 初始化 |
| await cli.initialize(api_key_final) |
| |
| # 设置模式 |
| thinking_mode = ThinkingMode.YOLO if mode == 'yolo' else ThinkingMode.INTERACTIVE |
| cli.agent.set_thinking_mode(thinking_mode) |
| |
| # 如果提供了消息且不是交互模式，直接处理 |
| if message and not interactive: |
| await cli.process_user_input(message) |
| else: |
| # 启动交互循环 |
| await cli.run_interactive_loop() |
| |
| except Exception as e: |
| click.echo(f"错误: {str(e)}") |
| finally: |
| await cli.shutdown() |
| |
| # 运行异步主函数 |
| asyncio.run(run()) |
| |
| |
| if __name__ == "__main__": |
| main() |