Task Manager Agent

1from connectonion import Agent
2from datetime import datetime
3
4# Simple in-memory storage (use database in production)
5tasks = []
6task_counter = 1
7
8def add_task(title: str, priority: str = "medium") -> str:
9    """Add a new task to the list."""
10    global task_counter
11    task = {
12        "id": task_counter,
13        "title": title,
14        "priority": priority,
15        "completed": False,
16        "created": datetime.now().strftime("%Y-%m-%d %H:%M")
17    }
18    tasks.append(task)
19    task_counter += 1
20    return f"✅ Added task #{task['id']}: '{title}' (Priority: {priority})"
21
22def list_tasks(status: str = "all") -> str:
23    """List tasks with optional filtering."""
24    if not tasks:
25        return "📝 No tasks found."
26    
27    filtered = tasks
28    if status == "pending":
29        filtered = [t for t in tasks if not t["completed"]]
30    elif status == "completed":
31        filtered = [t for t in tasks if t["completed"]]
32    
33    result = f"📋 Tasks ({status}):\n"
34    for task in filtered:
35        icon = "✅" if task["completed"] else "⭕"
36        result += f"{icon} #{task['id']} {task['title']} [{task['priority']}]\n"
37    return result
38
39def complete_task(task_id: int) -> str:
40    """Mark a task as completed."""
41    for task in tasks:
42        if task["id"] == task_id:
43            if task["completed"]:
44                return f"Task #{task_id} is already completed!"
45            task["completed"] = True
46            return f"🎉 Completed task #{task_id}: '{task['title']}'"
47    return f"❌ Task #{task_id} not found"
48
49# Create task manager agent  
50agent = Agent(
51    name="task_manager",
52    tools=[add_task, list_tasks, complete_task]
53)

1# task_manager_agent.py
2import os
3from connectonion import Agent
4from datetime import datetime, timedelta
5from typing import List, Dict, Any
6
7# Set your OpenAI API key
8os.environ['OPENAI_API_KEY'] = 'your-api-key-here'
9
10# Global task storage (in production, use a database)
11tasks_database = []
12task_id_counter = 1
13
14def add_task(title: str, priority: str = "medium", due_date: str = "") -> str:
15    """Add a new task with optional due date."""
16    global task_id_counter
17    
18    # Validate priority
19    valid_priorities = ["low", "medium", "high", "urgent"]
20    if priority.lower() not in valid_priorities:
21        priority = "medium"
22    
23    task = {
24        "id": task_id_counter,
25        "title": title.strip(),
26        "priority": priority.lower(),
27        "completed": False,
28        "created": datetime.now().isoformat(),
29        "due_date": due_date if due_date else None,
30        "completed_at": None
31    }
32    
33    tasks_database.append(task)
34    task_id_counter += 1
35    
36    due_text = f" (Due: {due_date})" if due_date else ""
37    return f"✅ Added task #{task['id']}: '{title}' [Priority: {priority.upper()}]{due_text}"
38
39def list_tasks(filter_type: str = "all") -> str:
40    """List tasks with various filtering options."""
41    if not tasks_database:
42        return "📝 No tasks found. Add some tasks to get started!"
43    
44    # Filter tasks based on type
45    filtered_tasks = []
46    if filter_type == "pending":
47        filtered_tasks = [t for t in tasks_database if not t["completed"]]
48    elif filter_type == "completed":
49        filtered_tasks = [t for t in tasks_database if t["completed"]]
50    elif filter_type == "high-priority":
51        filtered_tasks = [t for t in tasks_database if t["priority"] in ["high", "urgent"] and not t["completed"]]
52    elif filter_type == "overdue":
53        today = datetime.now().date()
54        filtered_tasks = [
55            t for t in tasks_database 
56            if not t["completed"] and t["due_date"] 
57            and datetime.fromisoformat(t["due_date"]).date() < today
58        ]
59    else:
60        filtered_tasks = tasks_database
61    
62    if not filtered_tasks:
63        return f"📝 No {filter_type} tasks found."
64    
65    # Sort by priority and due date
66    priority_order = {"urgent": 0, "high": 1, "medium": 2, "low": 3}
67    filtered_tasks.sort(key=lambda x: (priority_order.get(x["priority"], 2), x["created"]))
68    
69    result = f"📋 Tasks ({filter_type.upper()}): {len(filtered_tasks)} items\n\n"
70    
71    for task in filtered_tasks:
72        # Priority emoji
73        priority_emoji = {
74            "urgent": "🚨", "high": "🔴", "medium": "🟡", "low": "🟢"
75        }.get(task["priority"], "⚪")
76        
77        # Status emoji
78        status = "✅" if task["completed"] else "⭕"
79        
80        # Due date info
81        due_info = ""
82        if task["due_date"]:
83            due_date = datetime.fromisoformat(task["due_date"]).date()
84            today = datetime.now().date()
85            if due_date < today:
86                due_info = f" 🚨 OVERDUE ({task['due_date']})"
87            elif due_date == today:
88                due_info = f" 📅 DUE TODAY"
89            else:
90                due_info = f" 📅 Due {task['due_date']}"
91        
92        result += f"{status} {priority_emoji} #{task['id']} {task['title']}{due_info}\n"
93    
94    return result
95
96def complete_task(task_id: int) -> str:
97    """Mark a task as completed."""
98    for task in tasks_database:
99        if task["id"] == task_id:
100            if task["completed"]:
101                return f"⚠️  Task #{task_id} is already completed!"
102            
103            task["completed"] = True
104            task["completed_at"] = datetime.now().isoformat()
105            return f"🎉 Completed task #{task_id}: '{task['title']}'!"
106    
107    available_ids = [str(t["id"]) for t in tasks_database]
108    return f"❌ Task #{task_id} not found. Available task IDs: {', '.join(available_ids) if available_ids else 'None'}"
109
110def update_task_priority(task_id: int, new_priority: str) -> str:
111    """Update the priority of an existing task."""
112    valid_priorities = ["low", "medium", "high", "urgent"]
113    
114    if new_priority.lower() not in valid_priorities:
115        return f"❌ Invalid priority '{new_priority}'. Use: {', '.join(valid_priorities)}"
116    
117    for task in tasks_database:
118        if task["id"] == task_id:
119            old_priority = task["priority"]
120            task["priority"] = new_priority.lower()
121            return f"📝 Updated task #{task_id} priority: {old_priority.upper()} → {new_priority.upper()}"
122    
123    return f"❌ Task #{task_id} not found"
124
125def delete_task(task_id: int) -> str:
126    """Delete a task permanently."""
127    for i, task in enumerate(tasks_database):
128        if task["id"] == task_id:
129            deleted_task = tasks_database.pop(i)
130            return f"🗑️  Deleted task #{task_id}: '{deleted_task['title']}'"
131    
132    return f"❌ Task #{task_id} not found"
133
134def get_task_stats() -> str:
135    """Get overall task statistics."""
136    if not tasks_database:
137        return "📊 No tasks to analyze."
138    
139    total = len(tasks_database)
140    completed = len([t for t in tasks_database if t["completed"]])
141    pending = total - completed
142    
143    # Priority breakdown
144    priority_counts = {}
145    for task in tasks_database:
146        if not task["completed"]:
147            priority = task["priority"]
148            priority_counts[priority] = priority_counts.get(priority, 0) + 1
149    
150    # Overdue tasks
151    today = datetime.now().date()
152    overdue = len([
153        t for t in tasks_database 
154        if not t["completed"] and t["due_date"] 
155        and datetime.fromisoformat(t["due_date"]).date() < today
156    ])
157    
158    completion_rate = (completed / total * 100) if total > 0 else 0
159    
160    stats = f"""📊 Task Statistics
161    
162📈 Overall Progress:
163   • Total Tasks: {total}
164   • Completed: {completed} ({completion_rate:.1f}%)
165   • Pending: {pending}
166   
167🔥 Priority Breakdown (Pending):"""
168    
169    for priority in ["urgent", "high", "medium", "low"]:
170        count = priority_counts.get(priority, 0)
171        emoji = {"urgent": "🚨", "high": "🔴", "medium": "🟡", "low": "🟢"}[priority]
172        stats += f"\n   • {emoji} {priority.title()}: {count}"
173    
174    if overdue > 0:
175        stats += f"\n\n⚠️  Overdue Tasks: {overdue}"
176    
177    return stats
178
179# Create the task manager agent
180agent = Agent(
181    name="task_manager",
182    system_prompt="""You are a helpful personal task manager. You help users:
183    1. Add and organize tasks with priorities and due dates
184    2. Track task completion and progress
185    3. Filter and view tasks by status, priority, or due date
186    4. Update task priorities and delete tasks
187    5. Provide task statistics and insights
188    
189    Always be encouraging and help users stay organized and productive!""",
190    tools=[add_task, list_tasks, complete_task, update_task_priority, delete_task, get_task_stats]
191)
192
193if __name__ == "__main__":
194    print("=== Task Manager Agent Demo ===\n")
195    
196    # Demo workflow
197    commands = [
198        "Add a high priority task to review quarterly reports with due date 2024-03-15",
199        "Add a task to call dentist",  
200        "Add a low priority task to organize photos",
201        "Show me all my tasks",
202        "Complete task 1",
203        "Show me pending tasks",
204        "Update task 2 priority to high",
205        "Give me task statistics"
206    ]
207    
208    for i, command in enumerate(commands, 1):
209        print(f"Step {i}: {command}")
210        response = agent.input(command)
211        print(f"Response: {response}\n")
212        print("-" * 60)