""" Simple command handlers """ import os from conversation import ConversationManager from mcp_tools import MCPTools import config class Commands: """Simple command handlers""" def __init__(self, conversation: ConversationManager, mcp: MCPTools): self.conversation = conversation self.mcp = mcp self.temperature = config.DEFAULT_TEMPERATURE self.max_tokens = config.DEFAULT_MAX_TOKENS self.current_conversation_id = None def handle_command(self, user_input: str) -> bool: """Handle user command, return False to quit""" parts = user_input.split() cmd = parts[0].lower() if cmd in ['/quit', '/exit']: return False elif cmd == '/help': self._show_help() elif cmd == '/history': self._show_history() elif cmd == '/clear': self._clear_conversation() elif cmd == '/clear-history': self._clear_all_conversations() elif cmd == '/temp': self._set_temperature(parts) elif cmd == '/max-tokens': self._set_max_tokens(parts) elif cmd == '/token': self._set_token() elif cmd == '/token-file': self._set_token_file(parts) elif cmd == '/tools': self.mcp.list_tools() elif cmd == '/conv': self._set_conversation_id(parts) else: print(f"āš ļø Unknown command: {cmd}") return True def _show_help(self): print("\nšŸ“‹ Commands:") print(" /quit, /exit - Quit") print(" /help - Show this help") print(" /conv - Set conversation ID") print(" /history - List all conversations") print(" /clear - Clear current conversation") print(" /clear-history - Clear all conversations") print(f" /temp <0.0-1.0> - Set temperature (current: {self.temperature})") print(f" /max-tokens <1-8192> - Set max tokens (current: {self.max_tokens})") print(" /token - Enter Okta token") print(" /token-file - Load token from file") print(" /tools - List MCP tools") print() def _show_history(self): print("šŸ“œ Fetching conversation list...") result = self.conversation.list_conversations() if "error" in result: print(f"āŒ Error: {result['error']}") return conversations = result.get("conversations", []) if not conversations: print("šŸ“­ No conversations found") return print(f"šŸ“Š Found {len(conversations)} conversations:") for conv in conversations: conv_id = conv.get('conversation_id', 'unknown') updated = conv.get('updated_at', 'unknown') count = conv.get('message_count', 0) print(f" • {conv_id} ({count} messages, {updated})") def _clear_conversation(self): if not self.current_conversation_id: print("āš ļø No conversation ID set. Use /conv first") return print(f"šŸ—‘ļø Clearing conversation {self.current_conversation_id}...") result = self.conversation.clear_conversation(self.current_conversation_id) if "error" in result: print(f"āŒ Error: {result['error']}") else: print("āœ… Conversation cleared") def _clear_all_conversations(self): print("šŸ—‘ļø Clearing ALL conversations...") result = self.conversation.clear_all_conversations() if "error" in result: print(f"āŒ Error: {result['error']}") else: count = result.get('deleted_count', 0) print(f"āœ… Cleared {count} conversations") def _set_temperature(self, parts): if len(parts) < 2: print(f"šŸŒ”ļø Current temperature: {self.temperature}") return try: temp = float(parts[1]) if 0.0 <= temp <= 1.0: self.temperature = temp print(f"šŸŒ”ļø Temperature set to {temp}") else: print("āš ļø Temperature must be 0.0-1.0") except ValueError: print("āš ļø Invalid temperature") def _set_max_tokens(self, parts): if len(parts) < 2: print(f"šŸ”¢ Current max tokens: {self.max_tokens}") return try: tokens = int(parts[1]) if 1 <= tokens <= 8192: self.max_tokens = tokens print(f"šŸ”¢ Max tokens set to {tokens}") else: print("āš ļø Max tokens must be 1-8192") except ValueError: print("āš ļø Invalid max tokens") def _set_token(self): print("šŸ”‘ Enter Okta token:") try: token = input().strip() if token: self.mcp.set_token(token) else: print("āš ļø No token provided") except (EOFError, KeyboardInterrupt): print("āš ļø Cancelled") def _set_token_file(self, parts): if len(parts) < 2: print(f"āš ļø Usage: /token-file ") return try: with open(parts[1], 'r') as f: token = f.read().strip() if token: self.mcp.set_token(token) else: print("āš ļø Empty token file") except Exception as e: print(f"āŒ Error reading file: {e}") def _set_conversation_id(self, parts): if len(parts) < 2: print(f"šŸ’¬ Current conversation ID: {self.current_conversation_id or 'None'}") print(" Usage: /conv ") return self.current_conversation_id = parts[1] print(f"šŸ’¬ Conversation ID set to: {self.current_conversation_id}") def get_settings(self): """Get current settings for message sending""" return { 'temperature': self.temperature, 'max_tokens': self.max_tokens, 'conversation_id': self.current_conversation_id, 'okta_token': self.mcp.okta_token }