"""Tool definitions and executor for LLM tool calling.""" import logging from datetime import date, datetime from fabledassistant.services.caldav import ( complete_todo, create_event, create_todo, delete_event, delete_todo, is_caldav_configured, list_calendars, list_events, list_todos, search_events, update_event, update_todo, ) from fabledassistant.services.notes import create_note, get_note_by_title, list_notes, update_note from fabledassistant.services.tag_suggestions import suggest_tags logger = logging.getLogger(__name__) # Core tools — always available _CORE_TOOLS = [ { "type": "function", "function": { "name": "create_task", "description": "Create a new task for the user. Use this when the user asks you to add a task, todo, or reminder.", "parameters": { "type": "object", "properties": { "title": { "type": "string", "description": "The task title", }, "body": { "type": "string", "description": "Optional task description or details", }, "due_date": { "type": "string", "description": "Optional due date in YYYY-MM-DD format", }, "priority": { "type": "string", "enum": ["none", "low", "medium", "high"], "description": "Task priority level", }, }, "required": ["title"], }, }, }, { "type": "function", "function": { "name": "create_note", "description": "Create a new note for the user. Use this when the user asks you to write down, save, or record something as a note.", "parameters": { "type": "object", "properties": { "title": { "type": "string", "description": "The note title", }, "body": { "type": "string", "description": "The note content in markdown", }, }, "required": ["title"], }, }, }, { "type": "function", "function": { "name": "update_note", "description": ( "Update an existing note or task — content, title, status, priority, or due date. " "Use this when the user asks to add to, edit, expand, or modify a note, " "OR to mark a task done/in-progress, change its priority, or set a due date. " "NEVER use create_note when updating an existing note — always use update_note." ), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Title or keyword to find the note or task to update", }, "body": { "type": "string", "description": "New note content in markdown (omit if only updating task fields)", }, "title": { "type": "string", "description": "Optional new title", }, "mode": { "type": "string", "enum": ["replace", "append"], "description": ( "How to apply the new body: 'replace' overwrites existing content (default), " "'append' adds after existing content" ), }, "status": { "type": "string", "enum": ["todo", "in_progress", "done"], "description": "New task status. Use to mark a task done, start it, etc.", }, "priority": { "type": "string", "enum": ["none", "low", "medium", "high"], "description": "New task priority", }, "due_date": { "type": "string", "description": "New due date in YYYY-MM-DD format", }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "search_notes", "description": "Search the user's notes and tasks. Use this when the user asks about their existing notes, tasks, or wants to find something they've written.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Search query to find notes/tasks", }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "list_tasks", "description": ( "List the user's tasks with optional filters. Use this when the user asks about their tasks " "by status (e.g. 'what's in progress'), priority (e.g. 'high priority tasks'), " "or due date (e.g. 'overdue tasks', 'due this week'). " "For 'overdue', set due_before to today's date. For 'due today', set due_after to today and due_before to tomorrow." ), "parameters": { "type": "object", "properties": { "status": { "type": "string", "enum": ["todo", "in_progress", "done"], "description": "Filter by task status", }, "priority": { "type": "string", "enum": ["none", "low", "medium", "high"], "description": "Filter by priority level", }, "due_before": { "type": "string", "description": "Return tasks due before this date (YYYY-MM-DD). Use today's date to find overdue tasks.", }, "due_after": { "type": "string", "description": "Return tasks due on or after this date (YYYY-MM-DD)", }, "limit": { "type": "integer", "description": "Maximum number of tasks to return (default 10)", }, }, }, }, }, ] # CalDAV tools — only included when user has CalDAV configured _CALDAV_TOOLS = [ { "type": "function", "function": { "name": "create_event", "description": "Create a calendar event for the user. Use this when the user asks to schedule, add, or create a meeting, appointment, or event.", "parameters": { "type": "object", "properties": { "title": { "type": "string", "description": "A descriptive event title (e.g. 'John's Birthday' not just 'Birthday')", }, "start": { "type": "string", "description": "Start date (YYYY-MM-DD for all-day) or datetime (ISO 8601, e.g. 2025-01-15T14:00:00)", }, "end": { "type": "string", "description": "Optional end date or datetime in same format as start", }, "duration": { "type": "integer", "description": "Optional duration in minutes (default 60, ignored if end is set or all_day is true)", }, "description": { "type": "string", "description": "Optional event description", }, "location": { "type": "string", "description": "Optional event location", }, "all_day": { "type": "boolean", "description": "Set to true for all-day events like birthdays, holidays, deadlines (default false)", }, "recurrence": { "type": "string", "description": "Optional iCalendar RRULE (e.g. 'FREQ=YEARLY' for annual, 'FREQ=WEEKLY' for weekly, 'FREQ=MONTHLY' for monthly)", }, "reminder_minutes": { "type": "integer", "description": "Optional reminder N minutes before the event (e.g. 30 for 30 minutes before)", }, "attendees": { "type": "array", "items": {"type": "string"}, "description": "Optional list of attendee email addresses", }, "timezone": { "type": "string", "description": "Optional IANA timezone (e.g. 'America/New_York'). Falls back to user's caldav_timezone setting.", }, "calendar_name": { "type": "string", "description": "Optional calendar name to create the event in. Falls back to default calendar.", }, }, "required": ["title", "start"], }, }, }, { "type": "function", "function": { "name": "list_events", "description": "List calendar events in a date range. Use this when the user asks what events or meetings they have.", "parameters": { "type": "object", "properties": { "date_from": { "type": "string", "description": "Start of range in ISO 8601 format (e.g. 2025-01-15T00:00:00)", }, "date_to": { "type": "string", "description": "End of range in ISO 8601 format (e.g. 2025-01-22T23:59:59)", }, }, "required": ["date_from", "date_to"], }, }, }, { "type": "function", "function": { "name": "search_events", "description": "Search calendar events by keyword. Use this when the user asks to find a specific event or meeting.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Search keyword to match against event titles, locations, and descriptions", }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "update_event", "description": "Update an existing calendar event. Use this when the user asks to change, move, reschedule, or modify an event.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Search term to find the event to update (matches against title)", }, "title": { "type": "string", "description": "New title for the event", }, "start": { "type": "string", "description": "New start datetime in ISO 8601 format", }, "end": { "type": "string", "description": "New end datetime in ISO 8601 format", }, "description": { "type": "string", "description": "New event description", }, "location": { "type": "string", "description": "New event location", }, "timezone": { "type": "string", "description": "Optional IANA timezone (e.g. 'America/New_York')", }, "calendar_name": { "type": "string", "description": "Optional calendar name to search in", }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "delete_event", "description": "Delete a calendar event. Use this when the user asks to cancel, remove, or delete an event.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Search term to find the event to delete (matches against title)", }, "calendar_name": { "type": "string", "description": "Optional calendar name to search in", }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "list_calendars", "description": "List all available calendars. Use this when the user asks which calendars they have.", "parameters": { "type": "object", "properties": {}, }, }, }, { "type": "function", "function": { "name": "create_todo", "description": "Create a CalDAV todo item on the calendar server. Use this when the user explicitly asks to create a calendar todo or CalDAV todo (NOT for regular app tasks).", "parameters": { "type": "object", "properties": { "summary": { "type": "string", "description": "The todo summary/title", }, "due": { "type": "string", "description": "Optional due date/datetime in ISO 8601 format", }, "description": { "type": "string", "description": "Optional description", }, "priority": { "type": "integer", "description": "Optional priority (1=highest, 9=lowest)", }, "reminder_minutes": { "type": "integer", "description": "Optional reminder N minutes before due date", }, "timezone": { "type": "string", "description": "Optional IANA timezone (e.g. 'America/New_York')", }, "calendar_name": { "type": "string", "description": "Optional calendar name", }, }, "required": ["summary"], }, }, }, { "type": "function", "function": { "name": "list_todos", "description": "List CalDAV todos from the calendar server. Use this when the user asks to see their calendar todos.", "parameters": { "type": "object", "properties": { "include_completed": { "type": "boolean", "description": "Whether to include completed todos (default false)", }, "calendar_name": { "type": "string", "description": "Optional calendar name", }, }, }, }, }, { "type": "function", "function": { "name": "update_todo", "description": "Update a CalDAV todo's summary, due date, description, or priority.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Search term to find the todo to update (matches against summary)", }, "summary": { "type": "string", "description": "New summary/title for the todo", }, "due": { "type": "string", "description": "New due date or datetime in ISO 8601 format", }, "description": { "type": "string", "description": "New description", }, "priority": { "type": "integer", "description": "New priority (1=highest, 9=lowest)", }, "timezone": { "type": "string", "description": "Optional IANA timezone for due datetime", }, "calendar_name": { "type": "string", "description": "Optional calendar name to search in", }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "complete_todo", "description": "Mark a CalDAV todo as completed. Use this when the user says they finished or completed a calendar todo.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Search term to find the todo to complete (matches against summary)", }, "calendar_name": { "type": "string", "description": "Optional calendar name", }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "delete_todo", "description": "Delete a CalDAV todo. Use this when the user asks to remove or delete a calendar todo.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Search term to find the todo to delete (matches against summary)", }, "calendar_name": { "type": "string", "description": "Optional calendar name", }, }, "required": ["query"], }, }, }, ] async def get_tools_for_user(user_id: int) -> list[dict]: """Build the tool list for a user based on their configured integrations.""" tools = list(_CORE_TOOLS) if await is_caldav_configured(user_id): tools.extend(_CALDAV_TOOLS) logger.debug("User %d: %d tools available", user_id, len(tools)) return tools def _parse_due_date(value: str | None) -> date | None: """Parse a due date string, returning None on failure.""" if not value: return None try: return datetime.strptime(value, "%Y-%m-%d").date() except (ValueError, TypeError): logger.warning("Invalid due_date format: %s", value) return None async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict: """Execute a tool call and return the result.""" try: if tool_name == "create_task": task_title = arguments.get("title", "Untitled Task") task_body = arguments.get("body", "") note = await create_note( user_id=user_id, title=task_title, body=task_body, status="todo", priority=arguments.get("priority", "none"), due_date=_parse_due_date(arguments.get("due_date")), ) suggested = await suggest_tags(user_id, task_title, task_body) return { "success": True, "type": "task", "data": { "id": note.id, "title": note.title, "status": note.status, "priority": note.priority, "due_date": str(note.due_date) if note.due_date else None, }, "suggested_tags": suggested, } elif tool_name == "create_note": note_title = arguments.get("title", "Untitled Note") note_body = arguments.get("body", "") note = await create_note( user_id=user_id, title=note_title, body=note_body, ) suggested = await suggest_tags(user_id, note_title, note_body) return { "success": True, "type": "note", "data": { "id": note.id, "title": note.title, }, "suggested_tags": suggested, } elif tool_name == "update_note": query = arguments.get("query", "") new_body = arguments.get("body", "") new_title = arguments.get("title") mode = arguments.get("mode", "replace") # Find the note: exact title match first, then fuzzy search note = await get_note_by_title(user_id, query) if note is None: notes, _ = await list_notes(user_id=user_id, q=query, limit=5) # Prefer notes (not tasks) when there are mixed results note_only = [n for n in notes if n.status is None] candidates = note_only or notes if not candidates: return {"success": False, "error": f"No note found matching '{query}'."} note = candidates[0] # Build the update payload update_fields: dict = {} if new_title: update_fields["title"] = new_title if new_body: if mode == "append" and note.body: update_fields["body"] = note.body + "\n\n" + new_body else: update_fields["body"] = new_body if "status" in arguments: update_fields["status"] = arguments["status"] if "priority" in arguments: update_fields["priority"] = arguments["priority"] if "due_date" in arguments: update_fields["due_date"] = _parse_due_date(arguments["due_date"]) updated = await update_note(user_id, note.id, **update_fields) if updated is None: return {"success": False, "error": "Failed to update note."} suggested = await suggest_tags(user_id, updated.title, updated.body or "") return { "success": True, "type": "note_updated", "data": { "id": updated.id, "title": updated.title, }, "suggested_tags": suggested, } elif tool_name == "list_tasks": notes, total = await list_notes( user_id=user_id, is_task=True, status=arguments.get("status"), priority=arguments.get("priority"), due_before=_parse_due_date(arguments.get("due_before")), due_after=_parse_due_date(arguments.get("due_after")), limit=int(arguments.get("limit", 10)), sort="due_date", order="asc", ) results = [] for n in notes: results.append({ "id": n.id, "title": n.title, "status": n.status, "priority": n.priority, "due_date": str(n.due_date) if n.due_date else None, "preview": (n.body[:120] + "...") if n.body and len(n.body) > 120 else (n.body or ""), }) return { "success": True, "type": "tasks", "data": { "total": total, "count": len(results), "results": results, }, } elif tool_name == "search_notes": query = arguments.get("query", "") notes, total = await list_notes(user_id=user_id, q=query, limit=5) results = [] for n in notes: results.append({ "id": n.id, "title": n.title, "type": "task" if n.status is not None else "note", "status": n.status, "preview": (n.body[:200] + "...") if n.body and len(n.body) > 200 else (n.body or ""), }) return { "success": True, "type": "search", "data": { "query": query, "total": total, "results": results, }, } elif tool_name == "create_event": result = await create_event( user_id=user_id, title=arguments.get("title", "Untitled Event"), start=arguments["start"], end=arguments.get("end"), duration=arguments.get("duration"), description=arguments.get("description"), location=arguments.get("location"), all_day=arguments.get("all_day", False), recurrence=arguments.get("recurrence"), timezone=arguments.get("timezone"), reminder_minutes=arguments.get("reminder_minutes"), attendees=arguments.get("attendees"), calendar_name=arguments.get("calendar_name"), ) return { "success": True, "type": "event", "data": result, } elif tool_name == "list_events": events = await list_events( user_id=user_id, date_from=arguments["date_from"], date_to=arguments["date_to"], ) return { "success": True, "type": "events", "data": { "count": len(events), "events": events, }, } elif tool_name == "search_events": events = await search_events( user_id=user_id, query=arguments.get("query", ""), ) return { "success": True, "type": "events", "data": { "query": arguments.get("query", ""), "count": len(events), "events": events, }, } elif tool_name == "update_event": result = await update_event( user_id=user_id, query=arguments["query"], title=arguments.get("title"), start=arguments.get("start"), end=arguments.get("end"), description=arguments.get("description"), location=arguments.get("location"), timezone=arguments.get("timezone"), calendar_name=arguments.get("calendar_name"), ) return { "success": True, "type": "event_updated", "data": result, } elif tool_name == "delete_event": result = await delete_event( user_id=user_id, query=arguments["query"], calendar_name=arguments.get("calendar_name"), ) return { "success": True, "type": "event_deleted", "data": result, } elif tool_name == "list_calendars": calendars = await list_calendars(user_id=user_id) return { "success": True, "type": "calendars", "data": { "count": len(calendars), "calendars": calendars, }, } elif tool_name == "create_todo": result = await create_todo( user_id=user_id, summary=arguments.get("summary", "Untitled Todo"), due=arguments.get("due"), description=arguments.get("description"), priority=arguments.get("priority"), reminder_minutes=arguments.get("reminder_minutes"), timezone=arguments.get("timezone"), calendar_name=arguments.get("calendar_name"), ) return { "success": True, "type": "todo", "data": result, } elif tool_name == "list_todos": todos = await list_todos( user_id=user_id, include_completed=arguments.get("include_completed", False), calendar_name=arguments.get("calendar_name"), ) return { "success": True, "type": "todos", "data": { "count": len(todos), "todos": todos, }, } elif tool_name == "update_todo": result = await update_todo( user_id=user_id, query=arguments["query"], summary=arguments.get("summary"), due=arguments.get("due"), description=arguments.get("description"), priority=arguments.get("priority"), timezone=arguments.get("timezone"), calendar_name=arguments.get("calendar_name"), ) return { "success": True, "type": "todo_updated", "data": result, } elif tool_name == "complete_todo": result = await complete_todo( user_id=user_id, query=arguments["query"], calendar_name=arguments.get("calendar_name"), ) return { "success": True, "type": "todo_completed", "data": result, } elif tool_name == "delete_todo": result = await delete_todo( user_id=user_id, query=arguments["query"], calendar_name=arguments.get("calendar_name"), ) return { "success": True, "type": "todo_deleted", "data": result, } else: return {"success": False, "error": f"Unknown tool: {tool_name}"} except Exception as e: logger.exception("Tool execution failed: %s", tool_name) return {"success": False, "error": str(e)}