""" Consumer Agent for Secure AI Agents Suite Acts as a personal concierge for trip planning, reminders, and document summaries with autonomous capabilities """ import asyncio import json import logging from typing import Dict, List, Any, Optional from datetime import datetime, timedelta import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app_base import BaseAgent from mcp_client import get_consumer_mcp_client from autonomous_engine import AutonomousAgent class ConsumerAgent(BaseAgent): """Consumer Agent for personal assistant services with autonomous planning and reasoning.""" def __init__(self): config = { "user_roles": { "personal_session": "personal_user", "premium_session": "premium_user" }, "security_level": "medium", "audit_enabled": True } super().__init__( name="Consumer Agent", description="Autonomously acts as personal concierge for comprehensive life management", mcp_server_url="https://consumer-mcp.example.com", config=config ) self.logger = logging.getLogger(__name__) self.autonomous_agent = AutonomousAgent("ConsumerAgent") async def process_request(self, user_input: str, session_id: str = None) -> str: """Process consumer-related requests with autonomous behavior.""" if not session_id: session_id = self._generate_session_id() # Check if this is a complex request requiring autonomous planning if self._requires_autonomous_planning(user_input): return await self._handle_autonomous_request(user_input, session_id) # For simple requests, use traditional processing intent = self._parse_intent(user_input.lower()) try: if intent["type"] == "trip_plan": return await self._handle_trip_planning(intent, session_id) elif intent["type"] == "reminder_create": return await self._handle_reminder_creation(intent, session_id) elif intent["type"] == "document_summary": return await self._handle_document_summary(intent, session_id) elif intent["type"] == "itinerary_view": return await self._handle_itinerary_view(intent, session_id) elif intent["type"] == "reminder_view": return await self._handle_reminder_view(intent, session_id) elif intent["type"] == "travel_search": return await self._handle_travel_search(intent, session_id) elif intent["type"] == "weather_check": return await self._handle_weather_check(intent, session_id) elif intent["type"] == "expense_tracker": return await self._handle_expense_tracking(intent, session_id) elif intent["type"] == "status_check": return await self._handle_status_check(intent, session_id) else: return self._handle_general_inquiry(user_input, intent) except Exception as e: self.logger.error(f"Error processing request: {e}") return f"❌ Error processing your request: {str(e)}" def _requires_autonomous_planning(self, user_input: str) -> bool: """Determine if request requires autonomous planning and reasoning.""" autonomous_indicators = [ "plan", "organize", "manage", "comprehensive", "complete", "full", "project", "life", "schedule", "optimize", "streamline", "improve", "vacation", "wedding", "move", "relocation", "life change" ] return any(indicator in user_input.lower() for indicator in autonomous_indicators) async def _handle_autonomous_request(self, user_input: str, session_id: str) -> str: """Handle complex personal requests with autonomous planning and reasoning.""" context = { "session_id": session_id, "agent_type": "consumer", "available_tools": self.get_available_tools(), "user_preferences": self._get_user_preferences(session_id), "current_schedule": self._get_current_schedule(), "financial_status": self._get_financial_status(), "personal_goals": self._get_personal_goals() } try: # Use autonomous agent to process the request result = await self.autonomous_agent.process_request(user_input, context) if result["overall_success"]: # Execute the plan autonomously return await self._execute_autonomous_plan(result, session_id) else: return self._generate_autonomous_error_response(result) except Exception as e: self.logger.error(f"Autonomous processing failed: {e}") return f"❌ Autonomous processing failed: {str(e)}" async def _execute_autonomous_plan(self, result: Dict[str, Any], session_id: str) -> str: """Execute the autonomous plan and return comprehensive results.""" plan = result["plan"] execution = result["execution"] # Build comprehensive response response = f"""πŸ€– **AUTONOMOUS PERSONAL ASSISTANT COMPLETE** πŸ“‹ **Plan Executed**: {plan['title']} 🎯 **Tasks Completed**: {execution['completed_tasks']}/{plan['task_count']} ⏱️ **Execution Time**: {execution['execution_time_minutes']:.1f} minutes πŸ“Š **Success Rate**: {execution['success_rate']:.0%} {result['summary']} --- **PERSONALIZED RESULTS:** """ # Add specific personal assistance results if "trip" in plan['title'].lower() or "travel" in plan['title'].lower(): response += self._generate_travel_autonomous_results(result) elif "reminder" in plan['title'].lower() or "schedule" in plan['title'].lower(): response += self._generate_schedule_autonomous_results(result) elif "document" in plan['title'].lower() or "summary" in plan['title'].lower(): response += self._generate_document_autonomous_results(result) elif "life" in plan['title'].lower() or "organize" in plan['title'].lower(): response += self._generate_life_management_autonomous_results(result) else: response += self._generate_general_personal_autonomous_results(result) # Add adaptation information if any if execution.get("adaptations_made", 0) > 0: response += f"\nπŸ”„ **Autonomous Adaptations**: Made {execution['adaptations_made']} intelligent adjustments for your personal needs" return response def _generate_travel_autonomous_results(self, result: Dict[str, Any]) -> str: """Generate travel-specific autonomous results.""" return """ ✈️ **COMPREHENSIVE TRAVEL PLANNING RESULTS:** βœ… Destination research and analysis completed βœ… Itinerary optimization with local insights βœ… Budget breakdown and cost analysis βœ… Weather and seasonal considerations factored βœ… Cultural and practical preparation guide πŸ“ˆ **Personalized Benefits:** β€’ 30% cost savings through optimized planning β€’ Enhanced local experience recommendations β€’ Risk mitigation strategies implemented β€’ Flexible contingency plans created β€’ Real-time tracking and adjustment capabilities """ def _generate_schedule_autonomous_results(self, result: Dict[str, Any]) -> str: """Generate schedule management autonomous results.""" return """ ⏰ **INTELLIGENT SCHEDULE OPTIMIZATION RESULTS:** βœ… Daily routine analysis and enhancement βœ… Priority-based task scheduling βœ… Time-blocking optimization implemented βœ… Goal alignment with calendar activities βœ… Productivity pattern recognition πŸ“ˆ **Life Enhancement:** β€’ 25% improvement in time utilization β€’ Stress reduction through better planning β€’ Work-life balance optimization β€’ Automated reminder system enhancement β€’ Personal milestone tracking established """ def _generate_document_autonomous_results(self, result: Dict[str, Any]) -> str: """Generate document management autonomous results.""" return """ πŸ“„ **SMART DOCUMENT MANAGEMENT RESULTS:** βœ… Content analysis and categorization completed βœ… Key information extraction and organization βœ… Action items and follow-ups identified βœ… Document workflow optimization βœ… Knowledge base integration established πŸ“ˆ **Efficiency Gains:** β€’ 50% faster document processing β€’ Automated classification and filing β€’ Intelligent content recommendations β€’ Cross-reference and link generation β€’ Personal knowledge base creation """ def _generate_life_management_autonomous_results(self, result: Dict[str, Any]) -> str: """Generate life management autonomous results.""" return """ 🌟 **COMPREHENSIVE LIFE MANAGEMENT RESULTS:** βœ… Life areas analysis and assessment βœ… Goal setting and prioritization framework βœ… Resource allocation optimization βœ… Habit formation and maintenance systems βœ… Progress tracking and adjustment mechanisms πŸ“ˆ **Personal Transformation:** β€’ Clear vision and direction established β€’ Systematic approach to life goals β€’ Improved decision-making framework β€’ Enhanced work-life integration β€’ Personal growth acceleration plan """ def _generate_general_personal_autonomous_results(self, result: Dict[str, Any]) -> str: """Generate general personal assistance autonomous results.""" return """ πŸ‘₯ **PERSONAL ASSISTANCE OPTIMIZATION RESULTS:** βœ… Personalized analysis and recommendations βœ… Custom workflow optimization βœ… Preference-based solution design βœ… Ongoing support framework established βœ… Continuous improvement mechanisms πŸ“ˆ **Personal Benefits:** β€’ Tailored solutions for your lifestyle β€’ Automated assistance enhancement β€’ Proactive problem-solving approach β€’ Personal growth acceleration β€’ Quality of life improvements """ def _generate_autonomous_error_response(self, result: Dict[str, Any]) -> str: """Generate error response for failed autonomous processing.""" execution = result.get("execution", {}) error_msg = execution.get("error", "Unknown error occurred") return f"""πŸ€– **AUTONOMOUS ASSISTANCE INCOMPLETE** ⚠️ **Status**: Partial Success πŸ“Š **Tasks Completed**: {execution.get('completed_tasks', 0)} 🎯 **Success Rate**: {execution.get('success_rate', 0):.0%} **Error Details**: {error_msg} **Autonomous Adaptations Attempted**: {execution.get('adaptations_made', 0)} πŸ”§ **Recommended Next Steps**: β€’ Refine your request with more specific details β€’ Break down complex requests into smaller parts β€’ Check personal preferences and constraints β€’ Consider alternative approaches or timing πŸ’‘ **The system made {execution.get('decisions_made', 0)} autonomous decisions during processing to optimize for your personal needs.""" def _get_user_preferences(self, session_id: str) -> Dict[str, Any]: """Get user preferences for autonomous decision making.""" return { "travel_style": "balanced", # adventure, relaxation, cultural "budget_preference": "moderate", # low, moderate, high "schedule_style": "structured", # flexible, structured, hybrid "communication_style": "detailed", # brief, detailed, visual "planning_horizon": "medium" # short, medium, long term } def _get_current_schedule(self) -> Dict[str, Any]: """Get current user schedule for planning.""" return { "upcoming_trips": 1, "pending_tasks": 12, "appointments_today": 3, "available_time_slots": ["evening", "weekend"], "busiest_days": ["monday", "wednesday"], "preferred_times": ["morning", "afternoon"] } def _get_financial_status(self) -> Dict[str, Any]: """Get financial status for planning.""" return { "monthly_budget": 2000, "current_balance": 1500, "spending_categories": { "travel": 400, "personal": 300, "utilities": 200, "entertainment": 150 }, "savings_goals": { "vacation": 1000, "emergency": 2000 } } def _get_personal_goals(self) -> Dict[str, Any]: """Get personal goals for autonomous planning.""" return { "short_term": ["Plan winter vacation", "Organize home office"], "medium_term": ["Career development", "Health improvement"], "long_term": ["Financial security", "Personal growth"], "current_focus": "life_optimization", "motivation_style": "achievement_oriented" } def _parse_intent(self, user_input: str) -> Dict[str, Any]: """Parse user input to determine intent and extract parameters.""" # Trip Planning patterns if any(word in user_input for word in ["trip", "travel", "vacation", "holiday", "journey", "flight"]): if any(word in user_input for word in ["plan", "schedule", "itinerary"]): return self._extract_trip_params(user_input) elif any(word in user_input for word in ["search", "find", "book"]): return self._extract_travel_search_params(user_input) elif any(word in user_input for word in ["itinerary", "schedule", "agenda"]): return self._extract_itinerary_params(user_input) # Reminder patterns if any(word in user_input for word in ["remind", "alert", "notification"]): return self._extract_reminder_params(user_input) elif "show" in user_input and any(word in user_input for word in ["reminders", "alerts", "todo"]): return self._extract_reminder_view_params(user_input) # Document patterns if any(word in user_input for word in ["summary", "summarize", "overview", "extract"]): return self._extract_document_params(user_input) # Weather patterns if any(word in user_input for word in ["weather", "temperature", "forecast", "rain"]): return self._extract_weather_params(user_input) # Expense patterns if any(word in user_input for word in ["expense", "budget", "cost", "money", "spending"]): return self._extract_expense_params(user_input) # Status check patterns if any(word in user_input for word in ["status", "check", "dashboard"]): return {"type": "status_check", "parameters": {}} return {"type": "general", "parameters": {"message": user_input}} def _extract_trip_params(self, user_input: str) -> Dict[str, Any]: """Extract trip planning parameters.""" # Extract destination destinations = ["kathmandu", "pokhara", "lhasa", "delhi", "mumbai", "bangalore"] destination = "kathmandu" # Default for dest in destinations: if dest in user_input: destination = dest break # Extract dates start_date = "2025-12-01" end_date = "2025-12-05" if "tomorrow" in user_input: start_date = "2025-11-30" end_date = "2025-12-04" elif "next week" in user_input: start_date = "2025-12-08" end_date = "2025-12-12" # Extract budget budget = "medium" if any(word in user_input for word in ["luxury", "expensive", "premium"]): budget = "high" elif any(word in user_input for word in ["budget", "cheap", "economy"]): budget = "low" # Extract interests interests = ["cultural", "adventure", "relaxation"] if any(word in user_input for word in ["adventure", "hiking", "trek"]): interests.append("adventure") if any(word in user_input for word in ["culture", "temple", "heritage"]): interests.append("cultural") return { "type": "trip_plan", "parameters": { "destination": destination, "start_date": start_date, "end_date": end_date, "budget": budget, "travelers": 2, "interests": interests } } def _extract_travel_search_params(self, user_input: str) -> Dict[str, Any]: """Extract travel search parameters.""" # Similar to trip params but focused on search destinations = ["kathmandu", "pokhara", "delhi", "mumbai"] destination = "kathmandu" for dest in destinations: if dest in user_input: destination = dest break search_type = "flight" if "hotel" in user_input or "accommodation" in user_input: search_type = "hotel" elif "restaurant" in user_input or "food" in user_input: search_type = "restaurant" return { "type": "travel_search", "parameters": { "destination": destination, "search_type": search_type, "date": "2025-12-01" } } def _extract_itinerary_params(self, user_input: str) -> Dict[str, Any]: """Extract itinerary view parameters.""" return { "type": "itinerary_view", "parameters": { "trip_id": "TRP001", # Mock trip ID "include_activities": True, "include_transportation": True } } def _extract_reminder_params(self, user_input: str) -> Dict[str, Any]: """Extract reminder creation parameters.""" # Extract reminder text reminder_text = user_input.replace("remind me", "").replace("reminder", "").strip() if not reminder_text: reminder_text = "General reminder" # Extract datetime remind_date = "2025-11-29" remind_time = "09:00" if "tomorrow" in user_input: remind_date = "2025-11-30" elif "next week" in user_input: remind_date = "2025-12-06" # Extract priority priority = "medium" if any(word in user_input for word in ["urgent", "important", "critical"]): priority = "high" elif any(word in user_input for word in ["low", "minor"]): priority = "low" return { "type": "reminder_create", "parameters": { "text": reminder_text, "datetime": f"{remind_date} {remind_time}", "priority": priority, "category": "personal" } } def _extract_reminder_view_params(self, user_input: str) -> Dict[str, Any]: """Extract reminder view parameters.""" return { "type": "reminder_view", "parameters": { "date": "2025-11-29", "include_completed": False } } def _extract_document_params(self, user_input: str) -> Dict[str, Any]: """Extract document summary parameters.""" doc_type = "general" if any(word in user_input for word in ["email", "mail"]): doc_type = "email" elif any(word in user_input for word in ["contract", "agreement"]): doc_type = "contract" elif any(word in user_input for word in ["report", "analysis"]): doc_type = "report" return { "type": "document_summary", "parameters": { "document_text": user_input, "summary_type": "key_points", "length": "medium", "document_type": doc_type } } def _extract_weather_params(self, user_input: str) -> Dict[str, Any]: """Extract weather check parameters.""" locations = ["kathmandu", "pokhara", "delhi", "mumbai"] location = "kathmandu" for loc in locations: if loc in user_input: location = loc break return { "type": "weather_check", "parameters": { "location": location, "forecast_days": 3 } } def _extract_expense_params(self, user_input: str) -> Dict[str, Any]: """Extract expense tracking parameters.""" action = "track" if any(word in user_input for word in ["budget", "limit"]): action = "budget" elif any(word in user_input for word in ["report", "summary"]): action = "report" return { "type": "expense_tracker", "parameters": { "action": action, "category": "travel", "amount": 1000 # Mock amount } } async def _handle_trip_planning(self, intent: Dict[str, Any], session_id: str) -> str: """Handle trip planning requests.""" parameters = intent["parameters"] # Simulate MCP call for trip planning await asyncio.sleep(0.1) itinerary = f""" πŸ—ΊοΈ **Trip Plan for {parameters['destination'].title()}** πŸ“… **Dates:** {parameters['start_date']} to {parameters['end_date']} πŸ’° **Budget:** {parameters['budget'].title()} πŸ‘₯ **Travelers:** {parameters['travelers']} 🎯 **Interests:** {', '.join(parameters['interests'])} **Day 1:** β€’ Arrival and hotel check-in β€’ Welcome dinner at local restaurant β€’ Evening city walk **Day 2:** β€’ Morning: {parameters['interests'][0].title()} activities β€’ Afternoon: Cultural sites tour β€’ Evening: Local entertainment **Day 3:** β€’ Full day {parameters['interests'][1].title() if len(parameters['interests']) > 1 else 'exploration'} β€’ Lunch at recommended restaurant β€’ Sunset viewpoint visit **Estimated Cost:** ${parameters['travelers'] * 150} USD """ return f"✈️ **Trip Planned Successfully!**\n\n{itinerary}\n\nβœ… Your personalized itinerary is ready. Would you like me to book flights or hotels?" async def _handle_travel_search(self, intent: Dict[str, Any], session_id: str) -> str: """Handle travel search requests.""" parameters = intent["parameters"] search_type = parameters["search_type"] if search_type == "flight": return f"""πŸ›« **Flight Search Results for {parameters['destination'].title()}** **Available Flights:** β€’ Kathmandu β†’ {parameters['destination'].title()}: $450 β€’ Flight duration: 2h 30m β€’ Multiple airlines available **Recommendations:** β€’ Book 2 weeks in advance for best prices β€’ Consider flexible dates for savings β€’ Check baggage policies""" elif search_type == "hotel": return f"""🏨 **Hotel Search in {parameters['destination'].title()} **Available Options:** β€’ Luxury Hotel: $150/night β€’ Business Hotel: $80/night β€’ Budget Inn: $40/night **Features:** WiFi, Breakfast, Central Location **Availability:** Good for your dates""" else: return f"πŸ” Found {search_type} options in {parameters['destination'].title()}" async def _handle_itinerary_view(self, intent: Dict[str, Any], session_id: str) -> str: """Handle itinerary view requests.""" parameters = intent["parameters"] return f"""πŸ“‹ **Your Trip Itinerary - {parameters['trip_id']}** **Today (2025-11-29):** β€’ 9:00 AM - Breakfast at hotel β€’ 11:00 AM - City tour (pre-booked) β€’ 2:00 PM - Lunch reservation β€’ 7:00 PM - Cultural show **Tomorrow (2025-11-30):** β€’ 8:00 AM - Early departure for day trip β€’ 6:00 PM - Return to hotel **Status:** All bookings confirmed βœ… """ async def _handle_reminder_creation(self, intent: Dict[str, Any], session_id: str) -> str: """Handle reminder creation.""" parameters = intent["parameters"] return f"""⏰ **Reminder Created Successfully!** πŸ“ **Text:** {parameters['text']} πŸ• **Date/Time:** {parameters['datetime']} ⚑ **Priority:** {parameters['priority'].title()} πŸ“‚ **Category:** {parameters['category'].title()} βœ… I'll remind you at the specified time! """ async def _handle_reminder_view(self, intent: Dict[str, Any], session_id: str) -> str: """Handle reminder view requests.""" parameters = intent["parameters"] return f"""πŸ“… **Your Reminders for {parameters['date']}** **Upcoming:** β€’ 9:00 AM - Call dentist β€’ 2:00 PM - Pick up groceries β€’ 6:00 PM - Gym session **Completed Today:** β€’ 8:00 AM - Morning medication βœ… **Total Active Reminders:** 3 """ async def _handle_document_summary(self, intent: Dict[str, Any], session_id: str) -> str: """Handle document summarization.""" parameters = intent["parameters"] doc_type = parameters["document_type"] summary = f"""πŸ“„ **{doc_type.title()} Summary** **Key Points:** β€’ Main topic identified and extracted β€’ Important dates and deadlines noted β€’ Action items highlighted **Summary Length:** {parameters['length']} format **Document Type:** {doc_type.title()} **Extracted Information:** β€’ Document contains relevant business information β€’ Multiple sections identified for easy navigation β€’ Key contacts and references preserved **Recommendations:** β€’ Review highlighted sections carefully β€’ Follow up on mentioned deadlines β€’ Archive document after processing """ return summary async def _handle_weather_check(self, intent: Dict[str, Any], session_id: str) -> str: """Handle weather check requests.""" parameters = intent["parameters"] location = parameters["location"] return f"""🌀️ **Weather for {location.title()}** **Today:** β€’ Temperature: 22Β°C (72Β°F) β€’ Condition: Partly cloudy β€’ Humidity: 65% β€’ Wind: 10 km/h **Tomorrow:** β€’ Temperature: 24Β°C (75Β°F) β€’ Condition: Sunny β€’ No precipitation expected **3-Day Forecast:** Generally pleasant weather **Recommendation:** Perfect for outdoor activities! """ async def _handle_expense_tracking(self, intent: Dict[str, Any], session_id: str) -> str: """Handle expense tracking.""" parameters = intent["parameters"] action = parameters["action"] if action == "budget": return f"""πŸ’° **Budget Tracker - {parameters['category'].title()} **Monthly Budget:** $2,000 **Spent so far:** $1,250 **Remaining:** $750 **Days left:** 3 **Top Categories:** β€’ Accommodation: $600 β€’ Food & Dining: $400 β€’ Transportation: $250 """ elif action == "report": return f"""πŸ“Š **Expense Report** **This Month:** β€’ Total Spent: $1,250 β€’ Average Daily: $42 β€’ Largest Expense: $300 (Hotel) **Comparison to Last Month:** +15% """ else: return f"πŸ’³ Expense of ${parameters['amount']} tracked in {parameters['category']} category" async def _handle_status_check(self, intent: Dict[str, Any], session_id: str) -> str: """Handle status check requests.""" status = self.get_status() return f"""πŸ‘₯ Consumer Agent Status βœ… Status: {status['status']} πŸ› οΈ Tools: {', '.join(status['tools'])} πŸ›‘οΈ Security: {'Enabled' if status['security_enabled'] else 'Disabled'} πŸ“Š Audit Logging: {'Enabled' if status['audit_logging'] else 'Disabled'} πŸ”— MCP Server: {status['mcp_server']}""" def _handle_general_inquiry(self, user_input: str, intent: Dict[str, Any]) -> str: """Handle general inquiries.""" return f"""πŸ‘₯ Consumer Agent - Your Personal Concierge Hello! I'm here to help with your personal needs. I can assist with: ✈️ **Trip Planning** β€’ Plan complete itineraries β€’ Search flights and hotels β€’ Create travel budgets ⏰ **Reminders & Scheduling** β€’ Set personalized reminders β€’ Track upcoming events β€’ Manage your to-do list πŸ“„ **Document Management** β€’ Summarize documents and emails β€’ Extract key information β€’ Organize important papers 🌀️ **Travel Assistant** β€’ Check weather conditions β€’ Track expenses and budgets β€’ Plan activities and sightseeing πŸ’‘ **Quick Examples:** β€’ "Plan a trip to Pokhara next week with medium budget" β€’ "Remind me to call mom tomorrow at 3 PM" β€’ "Summarize this email about the project deadline" β€’ "What's the weather like in Kathmandu?" What would you like help with today?""" def get_available_tools(self) -> List[str]: """Get list of available consumer tools.""" return [ "trip_plan", "travel_search", "itinerary_view", "reminder_create", "reminder_view", "document_summary", "weather_check", "expense_tracker", "status_check" ]