""" Enhanced chat handling with tool visibility and proper streaming """ import time import json from typing import List, Dict, Any, Tuple from datetime import datetime import os def format_tool_message(tool_name: str, details: str = "", status: str = "running") -> str: """Format a tool call as HTML content""" icons = { "analyzing_emails": "📊", "extracting_patterns": "🔍", "creating_rules": "📋", "thinking": "🤔" } # Create a styled div that looks like a tool call tool_content = f"""
{icons.get(tool_name, '🔧')} {tool_name.replace('_', ' ').title()} {'⏳ Running...' if status == 'running' else '✓ Complete'}
{details or 'Processing...'}
""" return tool_content def process_chat_with_tools( user_message: str, chat_history: List[Dict[str, str]], mcp_client: Any, current_emails: List[Dict[str, Any]], pending_rules: List[Dict[str, Any]], applied_rules: List[Dict[str, Any]], rule_counter: int ) -> Tuple: """Process chat messages with tool visibility""" from .simple_agent import process_chat_message as process_agent_message from .ui_utils import create_interactive_rule_cards if not user_message.strip(): return (chat_history, "", create_interactive_rule_cards(pending_rules), "Ready...", pending_rules, rule_counter) # Check if API key is available if not os.getenv('OPENROUTER_API_KEY'): chat_history.append({"role": "user", "content": user_message}) error_msg = "⚠️ OpenRouter API key not configured. Please set OPENROUTER_API_KEY in HuggingFace Spaces secrets to enable AI features." chat_history.append({"role": "assistant", "content": error_msg}) return (chat_history, "", create_interactive_rule_cards(pending_rules), "API key missing", pending_rules, rule_counter) try: # Add user message chat_history.append({"role": "user", "content": user_message}) # Check if we should analyze emails analyze_keywords = ['analyze', 'suggest', 'organize', 'rules', 'help me organize', 'create rules', 'inbox', 'patterns'] should_analyze = any(keyword in user_message.lower() for keyword in analyze_keywords) if should_analyze and current_emails: # Add tool message for email analysis analysis_details = f"""Analyzing {len(current_emails)} emails... Looking for patterns in: • Sender domains and addresses • Subject line keywords • Email frequency by sender • Common phrases and topics • Time patterns Email categories found: • Newsletters: {len([e for e in current_emails if 'newsletter' in e.get('from_email', '').lower()])} • Work emails: {len([e for e in current_emails if any(domain in e.get('from_email', '') for domain in ['company.com', 'work.com'])])} • Personal: {len([e for e in current_emails if not any(keyword in e.get('from_email', '').lower() for keyword in ['newsletter', 'promo', 'noreply'])])} """ # Create a combined message with tool output tool_html = format_tool_message("analyzing_emails", analysis_details, "complete") # Add as assistant message with special formatting chat_history.append({ "role": "assistant", "content": tool_html }) # Get current rule state rule_state = { 'proposedRules': [r for r in pending_rules if r['status'] == 'pending'], 'activeRules': applied_rules, 'rejectedRules': [r for r in pending_rules if r['status'] == 'rejected'] } # Call the agent response_data = process_agent_message( user_message=user_message, emails=current_emails, conversation_history=chat_history[:-2] if should_analyze else chat_history[:-1], # Exclude tool message rule_state=rule_state ) # Extract response and rules response_text = response_data.get('response', '') extracted_rules = response_data.get('rules', []) # If rules were extracted, add a tool message if extracted_rules: rules_details = f"""Created {len(extracted_rules)} rules: """ for rule in extracted_rules: rules_details += f"📋 {rule['name']}\n" rules_details += f" {rule['description']}\n" rules_details += f" Confidence: {rule.get('confidence', 0.8)*100:.0f}%\n\n" tool_html = format_tool_message("creating_rules", rules_details, "complete") # Update rules updated_pending_rules = pending_rules.copy() for rule in extracted_rules: if not any(r.get('rule_id') == rule.get('rule_id') for r in updated_pending_rules): rule['status'] = 'pending' rule['id'] = rule.get('rule_id', f'rule_{rule_counter}') updated_pending_rules.append(rule) rule_counter += 1 # Add assistant response (clean it of hidden JSON) clean_response = response_text if "