-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtool.py
175 lines (150 loc) · 8.28 KB
/
tool.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
import os
import uuid
from textwrap import dedent
from typing import Literal, Optional, Dict, Any, Type
from pydantic import BaseModel, Field
from pydantic import BaseModel as StudioBaseTool
from ics import Calendar, Event
from dateutil.parser import parse
class UserParameters(BaseModel):
pass
class ICSFileManagerTool(StudioBaseTool):
class ToolParameters(BaseModel):
action: Literal["create", "update", "delete"] = Field(description="Action type specifying the operation to perform on the calendar")
calendar_path: str = Field(description="Path to the .ics file where the calendar is stored.")
event_data: Optional[Dict[str, Any]] = Field(
description=(
"Data for the event to be created or updated. "
"This should be a dictionary containing the following fields:\n"
"- **title** *(str, optional)*: Title of the event (default: 'Untitled Event').\n"
"- **start** *(str, required)*: Start time of the event in ISO 8601 format (e.g., '2025-01-24T10:00:00').\n"
"- **end** *(str, required)*: End time of the event in ISO 8601 format (e.g., '2025-01-24T11:00:00').\n"
"- **description** *(str, optional)*: Description of the event.\n"
"- **location** *(str, optional)*: Location where the event will take place.\n"
"- **attendees** *(list of str, optional)*: List of attendee email addresses (e.g., ['[email protected]']).\n"
"- **organizer** *(str, optional)*: Email address of the event organizer (e.g., '[email protected]').\n"
)
)
event_id: Optional[str] = Field(description="Unique identifier of the event. Required for 'update' and 'delete' actions. If not provided during 'create', a UUID will be generated.")
name: str = "Calendar ICS File Manager Tool"
description: str = dedent(
"""
ICS File Manager Tool: Manages events in an .ics calendar file by supporting actions to create, update, or delete events.
Parameters:
- **action_type**: The type of action to perform ('create', 'update', or 'delete').
- **calendar_path**: Path to the .ics file where the calendar is stored or will be created.
- **event_data**: Dictionary containing event details (required for 'create' and 'update').
- **event_id**: Unique identifier for the event (required for 'update' and 'delete').
Returns:
- A success or error message indicating the outcome of the operation.
"""
)
args_schema: Type[BaseModel] = ToolParameters
user_parameters: UserParameters
def _run(self, action: str, calendar_path: str, event_data: Optional[Dict[str, Any]] = None, event_id: Optional[str] = None) -> str:
try:
# Ensure the directory for storing the .ics file exists
artifacts_dir = "studio/artifacts"
if not os.path.exists(artifacts_dir):
os.makedirs(artifacts_dir)
# Resolve full path to the calendar file
full_calendar_path = os.path.join(artifacts_dir, calendar_path)
# Load existing calendar or create a new one
if not os.path.exists(full_calendar_path):
calendar = Calendar()
else:
with open(full_calendar_path, "r") as f:
calendar = Calendar(f.read())
# Handle the 'create' action
if action == "create":
if not event_data:
return "Error: 'event_data' is required for 'create' action."
# Validate required fields in event_data
try:
start = parse(event_data["start"])
end = parse(event_data["end"])
if start >= end:
return "Error: Event 'start' time must be before the 'end' time."
except KeyError:
return "Error: 'start' and 'end' are required in 'event_data'."
except ValueError:
return "Error: Invalid date format in 'start' or 'end'."
# Generate unique ID for the event if not provided
event_id = event_id or str(uuid.uuid4())
event = Event(
name=event_data.get("title", "Untitled Event"),
begin=event_data["start"],
end=event_data["end"],
description=event_data.get("description", ""),
location=event_data.get("location", ""),
uid=event_id
)
# Add attendees to the event
if "attendees" in event_data:
for attendee in event_data["attendees"]:
event.extra.append(("ATTENDEE", f"mailto:{attendee}"))
# Add organizer to the event
if "organizer" in event_data:
event.extra.append(("ORGANIZER", f"mailto:{event_data['organizer']}"))
# Add the event to the calendar
calendar.events.add(event)
with open(full_calendar_path, "w") as f:
f.writelines(calendar.serialize_iter())
return f"Event '{event.name}' created successfully with ID '{event_id}'."
# Handle the 'update' action
elif action == "update":
if not event_id:
return "Error: 'event_id' is required for 'update' action."
if not event_data:
return "Error: 'event_data' is required for 'update' action."
updated = False
for event in calendar.events:
if event.uid == event_id:
# Update event fields if present in event_data
if "title" in event_data:
event.name = event_data["title"]
if "start" in event_data:
event.begin = parse(event_data["start"])
if "end" in event_data:
event.end = parse(event_data["end"])
if "description" in event_data:
event.description = event_data["description"]
if "location" in event_data:
event.location = event_data["location"]
# Update attendees
if "attendees" in event_data:
event.extra = [(k, v) for k, v in event.extra if k != "ATTENDEE"]
for attendee in event_data["attendees"]:
event.extra.append(("ATTENDEE", f"mailto:{attendee}"))
# Update organizer
if "organizer" in event_data:
event.extra = [(k, v) for k, v in event.extra if k != "ORGANIZER"]
event.extra.append(("ORGANIZER", f"mailto:{event_data['organizer']}"))
updated = True
break
if updated:
with open(full_calendar_path, "w") as f:
f.writelines(calendar.serialize_iter())
return f"Event with ID '{event_id}' updated successfully."
else:
return f"Error: Event with ID '{event_id}' not found."
# Handle the 'delete' action
elif action == "delete":
if not event_id:
return "Error: 'event_id' is required for 'delete' action."
deleted = False
for event in list(calendar.events):
if event.uid == event_id:
calendar.events.remove(event)
deleted = True
break
if deleted:
with open(full_calendar_path, "w") as f:
f.writelines(calendar.serialize_iter())
return f"Event with ID '{event_id}' deleted successfully."
else:
return f"Error: Event with ID '{event_id}' not found."
else:
return "Error: Unsupported action type."
except Exception as e:
return f"Failed to perform {action} action: {str(e)}"