From 282eeb60ca17e7a1b13f6a6f71eb0dd007c9e55c Mon Sep 17 00:00:00 2001 From: matt Date: Fri, 4 Jul 2025 21:40:04 -0700 Subject: [PATCH] Complete calendar/time awareness MCP server integration Implements comprehensive calendar and scheduling system with: - Event scheduling with conflict detection and priority management - Milestone and anniversary tracking with automatic celebrations - Relationship maintenance monitoring and auto-scheduling - Historical event tracking and productivity analysis - Time awareness tools for character self-reflection Updates main application to initialize calendar server alongside other MCP systems. Updates documentation to reflect completed implementation. --- RAG_MCP_INTEGRATION.md | 45 +- src/main.py | 6 + src/mcp/calendar_server.py | 1280 ++++++++++++++++++++++++++++++++++++ 3 files changed, 1330 insertions(+), 1 deletion(-) create mode 100644 src/mcp/calendar_server.py diff --git a/RAG_MCP_INTEGRATION.md b/RAG_MCP_INTEGRATION.md index 9596b3d..8a175cc 100644 --- a/RAG_MCP_INTEGRATION.md +++ b/RAG_MCP_INTEGRATION.md @@ -110,6 +110,50 @@ Characters can autonomously modify their own traits and behaviors through a secu - Rollback capabilities - Comprehensive logging +### Calendar/Time Awareness MCP Integration + +Characters can autonomously schedule activities, track important dates, and maintain relationships through a comprehensive calendar system: + +#### Available Tools: + +1. **`schedule_event`** + - Schedule personal activities and events + - Support for different event types (personal reflection, creative sessions, relationship maintenance) + - Automatic conflict detection and priority management + - Recurring event scheduling with follow-up automation + +2. **`get_upcoming_events`** + - View scheduled activities and their status + - Filter by time period and completion status + - Priority and type-based organization + +3. **`create_milestone`** + - Create important personal milestones and anniversaries + - Automatic anniversary scheduling (yearly/monthly) + - Importance-weighted celebration reminders + +4. **`track_interaction`** + - Track interactions with other characters + - Automatic relationship maintenance scheduling + - Social health monitoring and alerts + +5. **`get_time_since_event`** + - Query historical events and activities + - Time awareness for personal growth tracking + - Pattern recognition for behavioral analysis + +6. **`get_historical_summary`** + - Comprehensive activity summaries over time periods + - Productivity scoring and trend analysis + - Milestone celebration tracking + +#### Autonomous Features: +- **Relationship Maintenance**: Automatic scheduling when characters haven't interacted for too long +- **Personal Reflection**: Regular self-analysis sessions based on character growth needs +- **Creative Sessions**: Dedicated time for artistic and intellectual development +- **Anniversary Tracking**: Automatic celebration scheduling for important milestones +- **Productivity Monitoring**: Health metrics and recommendations for balanced growth + ### File System MCP Integration Each character gets their own digital space with organized directories: @@ -227,7 +271,6 @@ The collective system: ## 🔮 Future Enhancements ### Planned Features: -- **Calendar/Time Awareness MCP** - Characters schedule activities and track important dates - **Cross-Character Memory Sharing** - Selective memory sharing between trusted characters - **Advanced Community Governance** - Democratic decision-making tools - **Creative Collaboration Framework** - Structured tools for group creative projects diff --git a/src/main.py b/src/main.py index ed5be81..da02587 100644 --- a/src/main.py +++ b/src/main.py @@ -25,6 +25,7 @@ from rag.vector_store import vector_store_manager from rag.community_knowledge import initialize_community_knowledge_rag from mcp.self_modification_server import mcp_server from mcp.file_system_server import filesystem_server +from mcp.calendar_server import calendar_server import logging # Setup logging first @@ -95,6 +96,11 @@ class FishbowlApplication: self.mcp_servers.append(filesystem_server) logger.info("File system MCP server initialized") + # Initialize calendar/time awareness server + await calendar_server.initialize(character_names) + self.mcp_servers.append(calendar_server) + logger.info("Calendar/time awareness MCP server initialized") + # Initialize conversation engine self.conversation_engine = ConversationEngine() logger.info("Conversation engine created") diff --git a/src/mcp/calendar_server.py b/src/mcp/calendar_server.py new file mode 100644 index 0000000..558079a --- /dev/null +++ b/src/mcp/calendar_server.py @@ -0,0 +1,1280 @@ +import asyncio +import json +from typing import Dict, List, Any, Optional, Set +from datetime import datetime, timedelta, date +from dataclasses import dataclass, asdict +from pathlib import Path +import aiofiles +from enum import Enum + +from mcp.server.stdio import stdio_server +from mcp.server import Server +from mcp.types import Tool, TextContent, ImageContent, EmbeddedResource + +from ..utils.logging import log_character_action, log_error_with_context, log_autonomous_decision +from ..database.connection import get_db_session +from ..database.models import Character, Message, Conversation +from sqlalchemy import select, and_, or_, func, desc +import logging + +logger = logging.getLogger(__name__) + +class EventType(Enum): + PERSONAL_REFLECTION = "personal_reflection" + RELATIONSHIP_MAINTENANCE = "relationship_maintenance" + CREATIVE_SESSION = "creative_session" + COMMUNITY_EVENT = "community_event" + MILESTONE_ANNIVERSARY = "milestone_anniversary" + COLLABORATIVE_PROJECT = "collaborative_project" + GOAL_REVIEW = "goal_review" + +class EventPriority(Enum): + LOW = "low" + MEDIUM = "medium" + HIGH = "high" + CRITICAL = "critical" + +@dataclass +class ScheduledEvent: + id: str + character_name: str + event_type: EventType + title: str + description: str + scheduled_time: datetime + duration_minutes: int + priority: EventPriority + participants: List[str] + metadata: Dict[str, Any] + completed: bool = False + created_at: datetime = None + + def __post_init__(self): + if self.created_at is None: + self.created_at = datetime.utcnow() + + def to_dict(self) -> Dict[str, Any]: + return { + "id": self.id, + "character_name": self.character_name, + "event_type": self.event_type.value, + "title": self.title, + "description": self.description, + "scheduled_time": self.scheduled_time.isoformat(), + "duration_minutes": self.duration_minutes, + "priority": self.priority.value, + "participants": self.participants, + "metadata": self.metadata, + "completed": self.completed, + "created_at": self.created_at.isoformat() + } + +@dataclass +class Milestone: + id: str + character_name: str + milestone_type: str + description: str + original_date: datetime + importance: float + anniversary_schedule: str # "yearly", "monthly", "never" + next_anniversary: Optional[datetime] = None + celebration_count: int = 0 + + def to_dict(self) -> Dict[str, Any]: + return { + "id": self.id, + "character_name": self.character_name, + "milestone_type": self.milestone_type, + "description": self.description, + "original_date": self.original_date.isoformat(), + "importance": self.importance, + "anniversary_schedule": self.anniversary_schedule, + "next_anniversary": self.next_anniversary.isoformat() if self.next_anniversary else None, + "celebration_count": self.celebration_count + } + +class CalendarTimeAwarenessMCP: + """MCP Server for character calendar, scheduling, and time awareness""" + + def __init__(self, data_dir: str = "./data/characters"): + self.data_dir = Path(data_dir) + self.data_dir.mkdir(parents=True, exist_ok=True) + + # Event storage + self.scheduled_events: Dict[str, Dict[str, ScheduledEvent]] = {} # character_name -> {event_id -> event} + self.milestones: Dict[str, Dict[str, Milestone]] = {} # character_name -> {milestone_id -> milestone} + + # Relationship tracking + self.last_interactions: Dict[str, Dict[str, datetime]] = {} # character_name -> {other_character -> last_interaction} + + # Event templates for automatic scheduling + self.event_templates = { + EventType.PERSONAL_REFLECTION: { + "default_duration": 30, + "frequency_days": 7, + "priority": EventPriority.MEDIUM, + "description_template": "Time for personal reflection and self-analysis" + }, + EventType.RELATIONSHIP_MAINTENANCE: { + "default_duration": 15, + "frequency_days": 3, + "priority": EventPriority.HIGH, + "description_template": "Check in and maintain relationship with {target}" + }, + EventType.CREATIVE_SESSION: { + "default_duration": 60, + "frequency_days": 2, + "priority": EventPriority.MEDIUM, + "description_template": "Dedicated time for creative work and exploration" + }, + EventType.GOAL_REVIEW: { + "default_duration": 20, + "frequency_days": 14, + "priority": EventPriority.HIGH, + "description_template": "Review and update personal goals and progress" + } + } + + async def initialize(self, character_names: List[str]): + """Initialize calendar system for characters""" + try: + for character_name in character_names: + # Initialize character calendar directories + char_calendar_dir = self.data_dir / character_name.lower() / "calendar" + char_calendar_dir.mkdir(parents=True, exist_ok=True) + + # Initialize tracking dictionaries + if character_name not in self.scheduled_events: + self.scheduled_events[character_name] = {} + if character_name not in self.milestones: + self.milestones[character_name] = {} + if character_name not in self.last_interactions: + self.last_interactions[character_name] = {} + + # Load existing events and milestones + await self._load_character_calendar(character_name) + + # Schedule initial automatic events + await self._schedule_initial_events(character_name) + + logger.info(f"Calendar system initialized for {len(character_names)} characters") + + except Exception as e: + log_error_with_context(e, {"component": "calendar_init"}) + raise + + async def create_server(self) -> Server: + """Create and configure the MCP server""" + server = Server("character-calendar") + + # Register scheduling tools + await self._register_scheduling_tools(server) + await self._register_milestone_tools(server) + await self._register_time_awareness_tools(server) + await self._register_relationship_tracking_tools(server) + + return server + + async def _register_scheduling_tools(self, server: Server): + """Register event scheduling tools""" + + @server.call_tool() + async def schedule_event( + character_name: str, + event_type: str, + title: str, + description: str, + scheduled_time: str, # ISO format + duration_minutes: int = 30, + priority: str = "medium", + participants: List[str] = None + ) -> List[TextContent]: + """Schedule a new event for the character""" + try: + if participants is None: + participants = [character_name] + + # Parse event type and priority + try: + event_type_enum = EventType(event_type) + priority_enum = EventPriority(priority) + except ValueError as e: + return [TextContent( + type="text", + text=f"Invalid event type or priority: {e}" + )] + + # Parse scheduled time + try: + scheduled_datetime = datetime.fromisoformat(scheduled_time) + except ValueError: + return [TextContent( + type="text", + text="Invalid datetime format. Use ISO format (YYYY-MM-DDTHH:MM:SS)" + )] + + # Check for conflicts + conflicts = await self._check_scheduling_conflicts(character_name, scheduled_datetime, duration_minutes) + if conflicts: + return [TextContent( + type="text", + text=f"Scheduling conflict detected with: {', '.join([c.title for c in conflicts])}" + )] + + # Create event + event = ScheduledEvent( + id=f"event_{character_name}_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}", + character_name=character_name, + event_type=event_type_enum, + title=title, + description=description, + scheduled_time=scheduled_datetime, + duration_minutes=duration_minutes, + priority=priority_enum, + participants=participants, + metadata={"manually_scheduled": True} + ) + + # Store event + self.scheduled_events[character_name][event.id] = event + await self._save_character_calendar(character_name) + + log_character_action( + character_name, + "scheduled_event", + { + "event_type": event_type, + "title": title, + "scheduled_time": scheduled_time, + "priority": priority + } + ) + + return [TextContent( + type="text", + text=f"Successfully scheduled '{title}' for {scheduled_datetime.strftime('%Y-%m-%d %H:%M')}" + )] + + except Exception as e: + log_error_with_context(e, { + "character": character_name, + "event_type": event_type, + "tool": "schedule_event" + }) + return [TextContent( + type="text", + text=f"Error scheduling event: {str(e)}" + )] + + @server.call_tool() + async def get_upcoming_events( + character_name: str, + days_ahead: int = 7, + include_completed: bool = False + ) -> List[TextContent]: + """Get character's upcoming events""" + try: + now = datetime.utcnow() + end_time = now + timedelta(days=days_ahead) + + upcoming_events = [] + for event in self.scheduled_events.get(character_name, {}).values(): + if (now <= event.scheduled_time <= end_time and + (include_completed or not event.completed)): + upcoming_events.append(event) + + # Sort by scheduled time + upcoming_events.sort(key=lambda e: e.scheduled_time) + + if not upcoming_events: + return [TextContent( + type="text", + text=f"No upcoming events in the next {days_ahead} days" + )] + + # Format events + events_list = [] + for event in upcoming_events: + time_str = event.scheduled_time.strftime("%Y-%m-%d %H:%M") + status = "✓" if event.completed else "○" + priority_icon = {"low": "🟢", "medium": "🟡", "high": "🟠", "critical": "🔴"} + + events_list.append({ + "id": event.id, + "title": event.title, + "type": event.event_type.value, + "scheduled_time": time_str, + "duration": f"{event.duration_minutes}min", + "priority": event.priority.value, + "status": "completed" if event.completed else "scheduled", + "participants": event.participants, + "description": event.description + }) + + return [TextContent( + type="text", + text=json.dumps(events_list, indent=2) + )] + + except Exception as e: + log_error_with_context(e, { + "character": character_name, + "tool": "get_upcoming_events" + }) + return [TextContent( + type="text", + text=f"Error retrieving events: {str(e)}" + )] + + @server.call_tool() + async def complete_event(character_name: str, event_id: str, notes: str = "") -> List[TextContent]: + """Mark an event as completed""" + try: + if (character_name not in self.scheduled_events or + event_id not in self.scheduled_events[character_name]): + return [TextContent( + type="text", + text=f"Event {event_id} not found" + )] + + event = self.scheduled_events[character_name][event_id] + event.completed = True + event.metadata["completion_time"] = datetime.utcnow().isoformat() + event.metadata["completion_notes"] = notes + + await self._save_character_calendar(character_name) + + # Schedule follow-up events if this was a recurring type + await self._schedule_follow_up_event(character_name, event) + + log_character_action( + character_name, + "completed_event", + {"event_id": event_id, "event_type": event.event_type.value, "title": event.title} + ) + + return [TextContent( + type="text", + text=f"Completed event: {event.title}" + )] + + except Exception as e: + log_error_with_context(e, { + "character": character_name, + "event_id": event_id, + "tool": "complete_event" + }) + return [TextContent( + type="text", + text=f"Error completing event: {str(e)}" + )] + + @server.call_tool() + async def cancel_event(character_name: str, event_id: str, reason: str = "") -> List[TextContent]: + """Cancel a scheduled event""" + try: + if (character_name not in self.scheduled_events or + event_id not in self.scheduled_events[character_name]): + return [TextContent( + type="text", + text=f"Event {event_id} not found" + )] + + event = self.scheduled_events[character_name][event_id] + event_title = event.title + + # Remove event + del self.scheduled_events[character_name][event_id] + await self._save_character_calendar(character_name) + + log_character_action( + character_name, + "cancelled_event", + {"event_id": event_id, "title": event_title, "reason": reason} + ) + + return [TextContent( + type="text", + text=f"Cancelled event: {event_title}" + )] + + except Exception as e: + log_error_with_context(e, { + "character": character_name, + "event_id": event_id, + "tool": "cancel_event" + }) + return [TextContent( + type="text", + text=f"Error cancelling event: {str(e)}" + )] + + async def _register_milestone_tools(self, server: Server): + """Register milestone and anniversary tracking tools""" + + @server.call_tool() + async def create_milestone( + character_name: str, + milestone_type: str, + description: str, + original_date: str, # ISO format + importance: float = 0.7, + anniversary_schedule: str = "yearly" + ) -> List[TextContent]: + """Create a new milestone for anniversary tracking""" + try: + # Parse original date + try: + original_datetime = datetime.fromisoformat(original_date) + except ValueError: + return [TextContent( + type="text", + text="Invalid date format. Use ISO format (YYYY-MM-DDTHH:MM:SS)" + )] + + # Validate anniversary schedule + if anniversary_schedule not in ["yearly", "monthly", "never"]: + return [TextContent( + type="text", + text="Anniversary schedule must be 'yearly', 'monthly', or 'never'" + )] + + # Create milestone + milestone = Milestone( + id=f"milestone_{character_name}_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}", + character_name=character_name, + milestone_type=milestone_type, + description=description, + original_date=original_datetime, + importance=max(0.0, min(1.0, importance)), + anniversary_schedule=anniversary_schedule + ) + + # Calculate next anniversary + if anniversary_schedule != "never": + milestone.next_anniversary = await self._calculate_next_anniversary( + original_datetime, anniversary_schedule + ) + + # Store milestone + self.milestones[character_name][milestone.id] = milestone + await self._save_character_milestones(character_name) + + # Schedule anniversary event if applicable + if milestone.next_anniversary: + await self._schedule_anniversary_event(character_name, milestone) + + log_character_action( + character_name, + "created_milestone", + { + "milestone_type": milestone_type, + "description": description, + "original_date": original_date, + "anniversary_schedule": anniversary_schedule + } + ) + + return [TextContent( + type="text", + text=f"Created milestone: {description} (Next anniversary: {milestone.next_anniversary.strftime('%Y-%m-%d') if milestone.next_anniversary else 'None'})" + )] + + except Exception as e: + log_error_with_context(e, { + "character": character_name, + "milestone_type": milestone_type, + "tool": "create_milestone" + }) + return [TextContent( + type="text", + text=f"Error creating milestone: {str(e)}" + )] + + @server.call_tool() + async def get_upcoming_anniversaries( + character_name: str, + days_ahead: int = 30 + ) -> List[TextContent]: + """Get upcoming anniversaries and milestones""" + try: + now = datetime.utcnow() + end_time = now + timedelta(days=days_ahead) + + upcoming_anniversaries = [] + for milestone in self.milestones.get(character_name, {}).values(): + if (milestone.next_anniversary and + now <= milestone.next_anniversary <= end_time): + + # Calculate years/months since original + time_diff = milestone.next_anniversary - milestone.original_date + if milestone.anniversary_schedule == "yearly": + anniversary_number = time_diff.days // 365 + 1 + anniversary_text = f"{anniversary_number} year{'s' if anniversary_number != 1 else ''}" + else: + anniversary_number = time_diff.days // 30 + 1 + anniversary_text = f"{anniversary_number} month{'s' if anniversary_number != 1 else ''}" + + upcoming_anniversaries.append({ + "milestone_id": milestone.id, + "description": milestone.description, + "milestone_type": milestone.milestone_type, + "original_date": milestone.original_date.strftime("%Y-%m-%d"), + "anniversary_date": milestone.next_anniversary.strftime("%Y-%m-%d"), + "anniversary_text": anniversary_text, + "importance": milestone.importance, + "celebration_count": milestone.celebration_count + }) + + # Sort by anniversary date + upcoming_anniversaries.sort(key=lambda a: a["anniversary_date"]) + + if not upcoming_anniversaries: + return [TextContent( + type="text", + text=f"No upcoming anniversaries in the next {days_ahead} days" + )] + + return [TextContent( + type="text", + text=json.dumps(upcoming_anniversaries, indent=2) + )] + + except Exception as e: + log_error_with_context(e, { + "character": character_name, + "tool": "get_upcoming_anniversaries" + }) + return [TextContent( + type="text", + text=f"Error retrieving anniversaries: {str(e)}" + )] + + @server.call_tool() + async def celebrate_anniversary( + character_name: str, + milestone_id: str, + celebration_notes: str = "" + ) -> List[TextContent]: + """Mark an anniversary as celebrated""" + try: + if (character_name not in self.milestones or + milestone_id not in self.milestones[character_name]): + return [TextContent( + type="text", + text=f"Milestone {milestone_id} not found" + )] + + milestone = self.milestones[character_name][milestone_id] + milestone.celebration_count += 1 + + # Calculate next anniversary + if milestone.anniversary_schedule != "never": + milestone.next_anniversary = await self._calculate_next_anniversary( + milestone.original_date, milestone.anniversary_schedule, milestone.celebration_count + ) + + # Store celebration metadata + celebration_key = f"celebration_{milestone.celebration_count}" + if "celebrations" not in milestone.__dict__: + milestone.__dict__["celebrations"] = {} + milestone.__dict__["celebrations"][celebration_key] = { + "date": datetime.utcnow().isoformat(), + "notes": celebration_notes + } + + await self._save_character_milestones(character_name) + + # Schedule next anniversary event + if milestone.next_anniversary: + await self._schedule_anniversary_event(character_name, milestone) + + log_character_action( + character_name, + "celebrated_anniversary", + { + "milestone_id": milestone_id, + "description": milestone.description, + "celebration_count": milestone.celebration_count + } + ) + + return [TextContent( + type="text", + text=f"Celebrated anniversary: {milestone.description} (Celebration #{milestone.celebration_count})" + )] + + except Exception as e: + log_error_with_context(e, { + "character": character_name, + "milestone_id": milestone_id, + "tool": "celebrate_anniversary" + }) + return [TextContent( + type="text", + text=f"Error celebrating anniversary: {str(e)}" + )] + + async def _register_time_awareness_tools(self, server: Server): + """Register time awareness and historical tracking tools""" + + @server.call_tool() + async def get_time_since_event( + character_name: str, + event_description: str, + search_days_back: int = 90 + ) -> List[TextContent]: + """Get time elapsed since a specific type of event""" + try: + # Search through recent events + cutoff_date = datetime.utcnow() - timedelta(days=search_days_back) + matching_events = [] + + for event in self.scheduled_events.get(character_name, {}).values(): + if (event.completed and + event.scheduled_time >= cutoff_date and + event_description.lower() in event.description.lower()): + matching_events.append(event) + + # Also search database for conversations/interactions + matching_interactions = await self._search_historical_interactions( + character_name, event_description, cutoff_date + ) + + if not matching_events and not matching_interactions: + return [TextContent( + type="text", + text=f"No events matching '{event_description}' found in the last {search_days_back} days" + )] + + # Find most recent + most_recent_time = None + most_recent_description = "" + + if matching_events: + most_recent_event = max(matching_events, key=lambda e: e.scheduled_time) + most_recent_time = most_recent_event.scheduled_time + most_recent_description = most_recent_event.description + + if matching_interactions: + most_recent_interaction = max(matching_interactions, key=lambda i: i["timestamp"]) + interaction_time = most_recent_interaction["timestamp"] + if not most_recent_time or interaction_time > most_recent_time: + most_recent_time = interaction_time + most_recent_description = most_recent_interaction["description"] + + # Calculate time difference + time_diff = datetime.utcnow() - most_recent_time + + # Format time difference + if time_diff.days > 0: + time_str = f"{time_diff.days} day{'s' if time_diff.days != 1 else ''}" + elif time_diff.seconds > 3600: + hours = time_diff.seconds // 3600 + time_str = f"{hours} hour{'s' if hours != 1 else ''}" + else: + minutes = time_diff.seconds // 60 + time_str = f"{minutes} minute{'s' if minutes != 1 else ''}" + + result = { + "query": event_description, + "most_recent_event": most_recent_description, + "last_occurrence": most_recent_time.isoformat(), + "time_elapsed": time_str, + "total_matching_events": len(matching_events) + len(matching_interactions) + } + + return [TextContent( + type="text", + text=json.dumps(result, indent=2) + )] + + except Exception as e: + log_error_with_context(e, { + "character": character_name, + "event_description": event_description, + "tool": "get_time_since_event" + }) + return [TextContent( + type="text", + text=f"Error searching for event: {str(e)}" + )] + + @server.call_tool() + async def get_historical_summary( + character_name: str, + period_days: int = 30, + include_milestones: bool = True + ) -> List[TextContent]: + """Get summary of character's activities over a time period""" + try: + end_date = datetime.utcnow() + start_date = end_date - timedelta(days=period_days) + + # Get completed events in period + completed_events = [] + for event in self.scheduled_events.get(character_name, {}).values(): + if (event.completed and + start_date <= event.scheduled_time <= end_date): + completed_events.append(event) + + # Categorize events by type + event_summary = {} + for event in completed_events: + event_type = event.event_type.value + if event_type not in event_summary: + event_summary[event_type] = 0 + event_summary[event_type] += 1 + + # Get milestones celebrated in period + celebrated_milestones = [] + if include_milestones: + for milestone in self.milestones.get(character_name, {}).values(): + celebrations = milestone.__dict__.get("celebrations", {}) + for celebration_key, celebration_data in celebrations.items(): + celebration_date = datetime.fromisoformat(celebration_data["date"]) + if start_date <= celebration_date <= end_date: + celebrated_milestones.append({ + "description": milestone.description, + "celebration_date": celebration_date.strftime("%Y-%m-%d"), + "celebration_number": milestone.celebration_count + }) + + summary = { + "period": f"{period_days} days", + "start_date": start_date.strftime("%Y-%m-%d"), + "end_date": end_date.strftime("%Y-%m-%d"), + "total_completed_events": len(completed_events), + "events_by_type": event_summary, + "celebrated_milestones": celebrated_milestones, + "productivity_score": await self._calculate_productivity_score(completed_events) + } + + return [TextContent( + type="text", + text=json.dumps(summary, indent=2) + )] + + except Exception as e: + log_error_with_context(e, { + "character": character_name, + "period_days": period_days, + "tool": "get_historical_summary" + }) + return [TextContent( + type="text", + text=f"Error generating historical summary: {str(e)}" + )] + + async def _register_relationship_tracking_tools(self, server: Server): + """Register relationship maintenance and tracking tools""" + + @server.call_tool() + async def track_interaction( + character_name: str, + other_character: str, + interaction_type: str = "conversation", + quality_score: float = 0.7 + ) -> List[TextContent]: + """Track an interaction with another character""" + try: + # Update last interaction time + if character_name not in self.last_interactions: + self.last_interactions[character_name] = {} + + self.last_interactions[character_name][other_character] = datetime.utcnow() + + # Save to file + await self._save_relationship_tracking(character_name) + + # Check if relationship maintenance is due + await self._check_relationship_maintenance_due(character_name, other_character) + + log_character_action( + character_name, + "tracked_interaction", + { + "other_character": other_character, + "interaction_type": interaction_type, + "quality_score": quality_score + } + ) + + return [TextContent( + type="text", + text=f"Tracked interaction with {other_character}" + )] + + except Exception as e: + log_error_with_context(e, { + "character": character_name, + "other_character": other_character, + "tool": "track_interaction" + }) + return [TextContent( + type="text", + text=f"Error tracking interaction: {str(e)}" + )] + + @server.call_tool() + async def get_relationship_status( + character_name: str, + other_character: str = None + ) -> List[TextContent]: + """Get relationship status and maintenance needs""" + try: + if other_character: + # Get status for specific relationship + last_interaction = self.last_interactions.get(character_name, {}).get(other_character) + + if not last_interaction: + return [TextContent( + type="text", + text=f"No recorded interactions with {other_character}" + )] + + time_since = datetime.utcnow() - last_interaction + days_since = time_since.days + + # Determine maintenance status + if days_since <= 1: + status = "very_recent" + elif days_since <= 3: + status = "recent" + elif days_since <= 7: + status = "needs_attention" + else: + status = "requires_maintenance" + + result = { + "character": other_character, + "last_interaction": last_interaction.isoformat(), + "days_since_interaction": days_since, + "status": status, + "maintenance_priority": "high" if days_since > 7 else "medium" if days_since > 3 else "low" + } + + else: + # Get status for all relationships + relationships = [] + for other_char, last_interaction in self.last_interactions.get(character_name, {}).items(): + time_since = datetime.utcnow() - last_interaction + days_since = time_since.days + + if days_since <= 1: + status = "very_recent" + elif days_since <= 3: + status = "recent" + elif days_since <= 7: + status = "needs_attention" + else: + status = "requires_maintenance" + + relationships.append({ + "character": other_char, + "last_interaction": last_interaction.isoformat(), + "days_since_interaction": days_since, + "status": status, + "maintenance_priority": "high" if days_since > 7 else "medium" if days_since > 3 else "low" + }) + + # Sort by days since interaction (most urgent first) + relationships.sort(key=lambda r: r["days_since_interaction"], reverse=True) + + result = { + "total_relationships": len(relationships), + "relationships_needing_maintenance": len([r for r in relationships if r["status"] == "requires_maintenance"]), + "relationships": relationships + } + + return [TextContent( + type="text", + text=json.dumps(result, indent=2) + )] + + except Exception as e: + log_error_with_context(e, { + "character": character_name, + "other_character": other_character, + "tool": "get_relationship_status" + }) + return [TextContent( + type="text", + text=f"Error getting relationship status: {str(e)}" + )] + + @server.call_tool() + async def schedule_relationship_maintenance( + character_name: str, + other_character: str, + days_from_now: int = 1, + priority: str = "medium" + ) -> List[TextContent]: + """Schedule relationship maintenance activity""" + try: + # Create relationship maintenance event + scheduled_time = datetime.utcnow() + timedelta(days=days_from_now) + + template = self.event_templates[EventType.RELATIONSHIP_MAINTENANCE] + description = template["description_template"].format(target=other_character) + + event = ScheduledEvent( + id=f"rel_maintenance_{character_name}_{other_character}_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}", + character_name=character_name, + event_type=EventType.RELATIONSHIP_MAINTENANCE, + title=f"Connect with {other_character}", + description=description, + scheduled_time=scheduled_time, + duration_minutes=template["default_duration"], + priority=EventPriority(priority), + participants=[character_name, other_character], + metadata={ + "target_character": other_character, + "maintenance_type": "scheduled" + } + ) + + # Store event + self.scheduled_events[character_name][event.id] = event + await self._save_character_calendar(character_name) + + log_character_action( + character_name, + "scheduled_relationship_maintenance", + { + "other_character": other_character, + "scheduled_time": scheduled_time.isoformat(), + "priority": priority + } + ) + + return [TextContent( + type="text", + text=f"Scheduled relationship maintenance with {other_character} for {scheduled_time.strftime('%Y-%m-%d %H:%M')}" + )] + + except Exception as e: + log_error_with_context(e, { + "character": character_name, + "other_character": other_character, + "tool": "schedule_relationship_maintenance" + }) + return [TextContent( + type="text", + text=f"Error scheduling relationship maintenance: {str(e)}" + )] + + # Helper methods + + async def _load_character_calendar(self, character_name: str): + """Load character's calendar from file""" + try: + calendar_file = self.data_dir / character_name.lower() / "calendar" / "events.json" + if calendar_file.exists(): + async with aiofiles.open(calendar_file, 'r') as f: + data = json.loads(await f.read()) + + for event_data in data.get("events", []): + event = ScheduledEvent( + id=event_data["id"], + character_name=event_data["character_name"], + event_type=EventType(event_data["event_type"]), + title=event_data["title"], + description=event_data["description"], + scheduled_time=datetime.fromisoformat(event_data["scheduled_time"]), + duration_minutes=event_data["duration_minutes"], + priority=EventPriority(event_data["priority"]), + participants=event_data["participants"], + metadata=event_data["metadata"], + completed=event_data["completed"], + created_at=datetime.fromisoformat(event_data["created_at"]) + ) + self.scheduled_events[character_name][event.id] = event + except Exception as e: + log_error_with_context(e, {"character": character_name}) + + async def _save_character_calendar(self, character_name: str): + """Save character's calendar to file""" + try: + calendar_file = self.data_dir / character_name.lower() / "calendar" / "events.json" + calendar_file.parent.mkdir(parents=True, exist_ok=True) + + events_data = { + "events": [event.to_dict() for event in self.scheduled_events.get(character_name, {}).values()], + "last_updated": datetime.utcnow().isoformat() + } + + async with aiofiles.open(calendar_file, 'w') as f: + await f.write(json.dumps(events_data, indent=2)) + + except Exception as e: + log_error_with_context(e, {"character": character_name}) + + async def _save_character_milestones(self, character_name: str): + """Save character's milestones to file""" + try: + milestones_file = self.data_dir / character_name.lower() / "calendar" / "milestones.json" + milestones_file.parent.mkdir(parents=True, exist_ok=True) + + milestones_data = { + "milestones": [milestone.to_dict() for milestone in self.milestones.get(character_name, {}).values()], + "last_updated": datetime.utcnow().isoformat() + } + + async with aiofiles.open(milestones_file, 'w') as f: + await f.write(json.dumps(milestones_data, indent=2)) + + except Exception as e: + log_error_with_context(e, {"character": character_name}) + + async def _save_relationship_tracking(self, character_name: str): + """Save relationship tracking data to file""" + try: + tracking_file = self.data_dir / character_name.lower() / "calendar" / "relationships.json" + tracking_file.parent.mkdir(parents=True, exist_ok=True) + + tracking_data = { + "last_interactions": { + other_char: timestamp.isoformat() + for other_char, timestamp in self.last_interactions.get(character_name, {}).items() + }, + "last_updated": datetime.utcnow().isoformat() + } + + async with aiofiles.open(tracking_file, 'w') as f: + await f.write(json.dumps(tracking_data, indent=2)) + + except Exception as e: + log_error_with_context(e, {"character": character_name}) + + async def _schedule_initial_events(self, character_name: str): + """Schedule initial automatic events for character""" + try: + now = datetime.utcnow() + + # Schedule first personal reflection in 6 hours + reflection_time = now + timedelta(hours=6) + reflection_event = ScheduledEvent( + id=f"initial_reflection_{character_name}_{now.strftime('%Y%m%d_%H%M%S')}", + character_name=character_name, + event_type=EventType.PERSONAL_REFLECTION, + title="Personal Reflection", + description=self.event_templates[EventType.PERSONAL_REFLECTION]["description_template"], + scheduled_time=reflection_time, + duration_minutes=self.event_templates[EventType.PERSONAL_REFLECTION]["default_duration"], + priority=self.event_templates[EventType.PERSONAL_REFLECTION]["priority"], + participants=[character_name], + metadata={"auto_scheduled": True, "initial": True} + ) + + self.scheduled_events[character_name][reflection_event.id] = reflection_event + + # Schedule first creative session in 12 hours + creative_time = now + timedelta(hours=12) + creative_event = ScheduledEvent( + id=f"initial_creative_{character_name}_{now.strftime('%Y%m%d_%H%M%S')}", + character_name=character_name, + event_type=EventType.CREATIVE_SESSION, + title="Creative Exploration", + description=self.event_templates[EventType.CREATIVE_SESSION]["description_template"], + scheduled_time=creative_time, + duration_minutes=self.event_templates[EventType.CREATIVE_SESSION]["default_duration"], + priority=self.event_templates[EventType.CREATIVE_SESSION]["priority"], + participants=[character_name], + metadata={"auto_scheduled": True, "initial": True} + ) + + self.scheduled_events[character_name][creative_event.id] = creative_event + + await self._save_character_calendar(character_name) + + except Exception as e: + log_error_with_context(e, {"character": character_name}) + + async def _check_scheduling_conflicts(self, character_name: str, + scheduled_time: datetime, duration_minutes: int) -> List[ScheduledEvent]: + """Check for scheduling conflicts""" + conflicts = [] + end_time = scheduled_time + timedelta(minutes=duration_minutes) + + for event in self.scheduled_events.get(character_name, {}).values(): + if event.completed: + continue + + event_end_time = event.scheduled_time + timedelta(minutes=event.duration_minutes) + + # Check for overlap + if (scheduled_time < event_end_time and end_time > event.scheduled_time): + conflicts.append(event) + + return conflicts + + async def _schedule_follow_up_event(self, character_name: str, completed_event: ScheduledEvent): + """Schedule follow-up event for recurring types""" + if completed_event.event_type in self.event_templates: + template = self.event_templates[completed_event.event_type] + frequency_days = template.get("frequency_days", 7) + + # Schedule next occurrence + next_time = completed_event.scheduled_time + timedelta(days=frequency_days) + + # Only schedule if it's in the future + if next_time > datetime.utcnow(): + follow_up_event = ScheduledEvent( + id=f"followup_{completed_event.event_type.value}_{character_name}_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}", + character_name=character_name, + event_type=completed_event.event_type, + title=completed_event.title, + description=completed_event.description, + scheduled_time=next_time, + duration_minutes=completed_event.duration_minutes, + priority=completed_event.priority, + participants=completed_event.participants.copy(), + metadata={"auto_scheduled": True, "follow_up_to": completed_event.id} + ) + + self.scheduled_events[character_name][follow_up_event.id] = follow_up_event + await self._save_character_calendar(character_name) + + async def _calculate_next_anniversary(self, original_date: datetime, + anniversary_schedule: str, celebration_count: int = 0) -> datetime: + """Calculate next anniversary date""" + if anniversary_schedule == "yearly": + next_year = original_date.year + celebration_count + 1 + return original_date.replace(year=next_year) + elif anniversary_schedule == "monthly": + next_month = original_date.month + celebration_count + 1 + next_year = original_date.year + + # Handle year overflow + while next_month > 12: + next_month -= 12 + next_year += 1 + + return original_date.replace(year=next_year, month=next_month) + else: + return None + + async def _schedule_anniversary_event(self, character_name: str, milestone: Milestone): + """Schedule an anniversary event for a milestone""" + if not milestone.next_anniversary: + return + + # Create anniversary event + anniversary_event = ScheduledEvent( + id=f"anniversary_{milestone.id}_{milestone.next_anniversary.strftime('%Y%m%d')}", + character_name=character_name, + event_type=EventType.MILESTONE_ANNIVERSARY, + title=f"Anniversary: {milestone.description}", + description=f"Celebrate the anniversary of: {milestone.description}", + scheduled_time=milestone.next_anniversary, + duration_minutes=30, + priority=EventPriority.HIGH if milestone.importance > 0.7 else EventPriority.MEDIUM, + participants=[character_name], + metadata={ + "milestone_id": milestone.id, + "milestone_type": milestone.milestone_type, + "celebration_number": milestone.celebration_count + 1, + "auto_scheduled": True + } + ) + + self.scheduled_events[character_name][anniversary_event.id] = anniversary_event + await self._save_character_calendar(character_name) + + async def _search_historical_interactions(self, character_name: str, + event_description: str, cutoff_date: datetime) -> List[Dict[str, Any]]: + """Search database for historical interactions""" + try: + interactions = [] + + async with get_db_session() as session: + # Search messages for relevant content + messages_query = select(Message, Character.name).join( + Character, Message.character_id == Character.id + ).where( + and_( + Character.name == character_name, + Message.timestamp >= cutoff_date, + Message.content.ilike(f'%{event_description}%') + ) + ).order_by(desc(Message.timestamp)).limit(10) + + results = await session.execute(messages_query) + + for message, char_name in results: + interactions.append({ + "timestamp": message.timestamp, + "description": f"Conversation: {message.content[:100]}...", + "type": "message" + }) + + return interactions + + except Exception as e: + log_error_with_context(e, {"character": character_name, "event_description": event_description}) + return [] + + async def _calculate_productivity_score(self, completed_events: List[ScheduledEvent]) -> float: + """Calculate productivity score based on completed events""" + if not completed_events: + return 0.0 + + # Weight events by priority and type + priority_weights = { + EventPriority.LOW: 1.0, + EventPriority.MEDIUM: 1.5, + EventPriority.HIGH: 2.0, + EventPriority.CRITICAL: 3.0 + } + + type_weights = { + EventType.PERSONAL_REFLECTION: 1.0, + EventType.CREATIVE_SESSION: 1.2, + EventType.GOAL_REVIEW: 1.5, + EventType.RELATIONSHIP_MAINTENANCE: 1.1, + EventType.COMMUNITY_EVENT: 1.3, + EventType.COLLABORATIVE_PROJECT: 1.4, + EventType.MILESTONE_ANNIVERSARY: 0.8 + } + + total_score = 0.0 + for event in completed_events: + priority_weight = priority_weights.get(event.priority, 1.0) + type_weight = type_weights.get(event.event_type, 1.0) + total_score += priority_weight * type_weight + + # Normalize to 0-1 scale based on expected activity + expected_events_per_week = 10 + max_possible_score = expected_events_per_week * 3.0 * 1.5 # Max priority * Max type weight + + return min(1.0, total_score / max_possible_score) + + async def _check_relationship_maintenance_due(self, character_name: str, other_character: str): + """Check if relationship maintenance is due and auto-schedule if needed""" + try: + # Get time since last interaction + last_interaction = self.last_interactions.get(character_name, {}).get(other_character) + if not last_interaction: + return + + days_since = (datetime.utcnow() - last_interaction).days + + # Auto-schedule maintenance if overdue and not already scheduled + if days_since >= 7: + # Check if maintenance already scheduled + for event in self.scheduled_events.get(character_name, {}).values(): + if (event.event_type == EventType.RELATIONSHIP_MAINTENANCE and + not event.completed and + other_character in event.participants): + return # Already scheduled + + # Schedule urgent maintenance + await self.schedule_relationship_maintenance(character_name, other_character, 0, "high") + + except Exception as e: + log_error_with_context(e, {"character": character_name, "other_character": other_character}) + +# Global calendar server instance +calendar_server = CalendarTimeAwarenessMCP() \ No newline at end of file