"""Tool definitions and executor for LLM tool calling.""" import asyncio 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, search_todos, update_event, update_todo, ) from fabledassistant.config import Config from fabledassistant.services.notes import create_note, delete_note, get_note_by_title, list_notes, update_note from fabledassistant.services.tag_suggestions import suggest_tags logger = logging.getLogger(__name__) def _schedule_embedding(note_id: int, user_id: int, title: str, body: str) -> None: """Fire-and-forget: update the embedding for a note after it's created/modified.""" from fabledassistant.services.embeddings import upsert_note_embedding text = f"{title}\n{body}".strip() if body else (title or "") if text: asyncio.create_task(upsert_note_embedding(note_id, user_id, text)) # 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", }, "tags": { "type": "array", "items": {"type": "string"}, "description": "Tags for the note (without # prefix, hyphens for multi-word: [\"science-fiction\", \"story/idea\"]). Do NOT embed #tags in the note body.", }, }, "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", }, "tags": { "type": "array", "items": {"type": "string"}, "description": "Tags to apply (see tag_mode for how they're applied)", }, "tag_mode": { "type": "string", "enum": ["replace", "add", "remove"], "description": "'replace' sets tags to exactly this list, 'add' appends without duplicates, 'remove' removes listed tags (default: replace)", }, }, "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", }, "type": { "type": "string", "enum": ["note", "task"], "description": "Restrict results to only notes or only tasks. Omit to search both.", }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "get_note", "description": ( "Retrieve the full content of a specific note. Use this when the user asks to read, " "view, or check what a particular note says. Returns the complete note body." ), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Title or keyword to identify the note", }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "list_notes", "description": ( "Browse the user's notes (not tasks). Use this when the user asks to see, list, or browse " "their notes — by recency, keyword, or tag. For tasks, use list_tasks instead." ), "parameters": { "type": "object", "properties": { "q": { "type": "string", "description": "Optional keyword filter", }, "tags": { "type": "array", "items": {"type": "string"}, "description": "Filter notes that have ALL of these tags", }, "sort": { "type": "string", "enum": ["updated_at", "created_at", "title"], "description": "Sort field (default: updated_at)", }, "limit": { "type": "integer", "description": "Maximum number of notes to return (default 10)", }, }, }, }, }, { "type": "function", "function": { "name": "delete_note", "description": ( "Delete a note permanently. Use ONLY when the user explicitly asks to delete or remove a note. " "This action requires user confirmation and cannot be undone." ), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Title or keyword to find the note to delete", }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "delete_task", "description": ( "Delete a task permanently. Use ONLY when the user explicitly asks to delete or remove a task. " "This action requires user confirmation and cannot be undone." ), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Title or keyword to find the task to delete", }, }, "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"], }, }, }, { "type": "function", "function": { "name": "search_todos", "description": ( "Search CalDAV todos by keyword. Use this when the user asks to find a specific calendar todo " "by name or content, rather than listing all todos." ), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Keyword to search for in todo summary or description", }, "include_completed": { "type": "boolean", "description": "Include completed todos in search (default false)", }, "calendar_name": { "type": "string", "description": "Optional calendar name to restrict search", }, }, "required": ["query"], }, }, }, ] _SEARCH_TOOLS = [ { "type": "function", "function": { "name": "search_web", "description": ( "Search the web for quick information and answer the user's question from results. " "Use for factual lookups, current events, version numbers, quick definitions, or any question " "where a fast web answer is needed without creating a note. " "Use research_topic instead when the user explicitly wants a comprehensive note or deep research." ), "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "The search query"} }, "required": ["query"], }, }, } ] _RESEARCH_TOOLS = [ { "type": "function", "function": { "name": "research_topic", "description": ( "Research a topic by searching the web and compiling a note with cited sources. " "Use for 'research X', 'look up X', 'find information about X', " "'compile notes on X'. Takes 30–120 seconds." ), "parameters": { "type": "object", "properties": { "topic": { "type": "string", "description": "The topic or question to research", } }, "required": ["topic"], }, }, } ] _IMAGE_TOOLS = [ { "type": "function", "function": { "name": "search_images", "description": ( "Search the web for images and display them inline in the response. " "ONLY use this when the user explicitly asks to SEE, SHOW, VIEW, or DISPLAY " "an image or photo — e.g. 'show me a picture of X', 'what does X look like', " "'find a photo of X', 'display an image of X'. " "Do NOT use for general descriptions, factual questions, or when the user just " "wants information without visual content. Returns at most 2 images." ), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "The image search query", } }, "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) if Config.searxng_enabled(): tools.extend(_SEARCH_TOOLS) tools.extend(_RESEARCH_TOOLS) tools.extend(_IMAGE_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", "") if not isinstance(task_title, str): return {"success": False, "error": "title must be a string. Call create_task once per task."} if not isinstance(task_body, str): task_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) _schedule_embedding(note.id, 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_tags = arguments.get("tags", []) if not isinstance(note_title, str): return {"success": False, "error": "title must be a string. Call create_note once per note."} if not isinstance(note_body, str): note_body = "" if not isinstance(note_tags, list): note_tags = [] note = await create_note( user_id=user_id, title=note_title, body=note_body, tags=note_tags, ) suggested = await suggest_tags(user_id, note_title, note_body) _schedule_embedding(note.id, 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"]) if "tags" in arguments: new_tags = arguments["tags"] if isinstance(new_tags, list): tag_mode = arguments.get("tag_mode", "replace") if tag_mode == "add": existing = list(note.tags or []) for t in new_tags: if t not in existing: existing.append(t) update_fields["tags"] = existing elif tag_mode == "remove": existing = list(note.tags or []) update_fields["tags"] = [t for t in existing if t not in new_tags] else: update_fields["tags"] = new_tags 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 "") _schedule_embedding(updated.id, 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", "") type_filter = arguments.get("type") is_task: bool | None = None if type_filter == "note": is_task = False elif type_filter == "task": is_task = True notes, total = await list_notes(user_id=user_id, q=query, is_task=is_task, 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, } elif tool_name == "search_todos": results = await search_todos( user_id=user_id, query=arguments.get("query", ""), include_completed=bool(arguments.get("include_completed", False)), calendar_name=arguments.get("calendar_name"), ) return { "success": True, "type": "todos", "data": { "count": len(results), "todos": results, }, } elif tool_name == "get_note": query = arguments.get("query", "") note = await get_note_by_title(user_id, query) if note is None: notes, _ = await list_notes(user_id=user_id, q=query, limit=3) if not notes: return {"success": False, "error": f"No note found matching '{query}'."} note = notes[0] return { "success": True, "type": "note_content", "data": { "id": note.id, "title": note.title, "body": note.body or "", "tags": note.tags or [], }, } elif tool_name == "list_notes": tags_raw = arguments.get("tags") tags = tags_raw if isinstance(tags_raw, list) else None notes, total = await list_notes( user_id=user_id, q=arguments.get("q") or arguments.get("query"), tags=tags, is_task=False, sort=arguments.get("sort", "updated_at"), order="desc", limit=int(arguments.get("limit", 10)), ) results = [ { "id": n.id, "title": n.title, "tags": n.tags or [], "preview": (n.body[:200] + "...") if n.body and len(n.body) > 200 else (n.body or ""), } for n in notes ] return { "success": True, "type": "notes_list", "data": { "total": total, "count": len(results), "results": results, }, } elif tool_name == "delete_note": query = arguments.get("query", "") note = await get_note_by_title(user_id, query) if note is None: notes, _ = await list_notes(user_id=user_id, q=query, is_task=False, limit=3) if not notes: return {"success": False, "error": f"No note found matching '{query}'."} note = notes[0] if note.status is not None: return {"success": False, "error": f"'{note.title}' is a task. Use delete_task instead."} deleted = await delete_note(user_id, note.id) if not deleted: return {"success": False, "error": "Failed to delete note."} return { "success": True, "type": "note_deleted", "data": {"id": note.id, "title": note.title}, } elif tool_name == "delete_task": query = arguments.get("query", "") note = await get_note_by_title(user_id, query) if note is None: notes, _ = await list_notes(user_id=user_id, q=query, is_task=True, limit=3) if not notes: return {"success": False, "error": f"No task found matching '{query}'."} note = notes[0] if note.status is None: return {"success": False, "error": f"'{note.title}' is a note. Use delete_note instead."} deleted = await delete_note(user_id, note.id) if not deleted: return {"success": False, "error": "Failed to delete task."} return { "success": True, "type": "task_deleted", "data": {"id": note.id, "title": note.title}, } elif tool_name == "search_web": from fabledassistant.services.research import _search_searxng query = arguments.get("query", "") results = await _search_searxng(query) if not results: return {"success": False, "error": f"No results found for '{query}'"} return { "success": True, "type": "web_search", "data": { "query": query, "results": results, "count": len(results), }, } elif tool_name == "research_topic": # Research is always handled upstream in generation_task.py (round 0). # This fallback exists in case it somehow reaches execute_tool. topic = arguments.get("topic", "") return { "success": True, "type": "research_pending", "data": {"topic": topic}, } elif tool_name == "search_images": from fabledassistant.services.images import fetch_and_store_image from fabledassistant.services.research import _search_searxng_images query = arguments.get("query", "") if not query: return {"success": False, "error": "query is required"} raw_results = await _search_searxng_images(query) if not raw_results: return {"success": False, "error": f"No images found for '{query}'"} # Fetch and cache up to 2 images images = [] for r in raw_results[:2]: img_url = r.get("img_src", "") if not img_url: continue record = await fetch_and_store_image( url=img_url, title=r.get("title"), source_domain=r.get("source_domain"), ) if record: title = record.title or query page_url = r.get("page_url", "") source_domain = r.get("source_domain", "") images.append({ "embed": f"![{title}](/api/images/{record.id})", "citation": f"*Source: [{source_domain or 'image'}]({page_url})*", "page_url": page_url, "title": title, }) if not images: return {"success": False, "error": f"Could not retrieve images for '{query}'"} return { "success": True, "type": "image_search", "data": { "query": query, "images": images, "instructions": ( "Embed each image in your response by writing the 'embed' field verbatim, " "then the 'citation' field on the next line. Do not paraphrase or list URLs." ), }, } 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)}