refactor(tools): consolidate LLM tools from 42 to 38

Merge create_task into create_note (set status='todo' for tasks, omit
for notes), merge delete_task into delete_note, consolidate entity
tools (create/update_person → save_person, create/update_place →
save_place), rename get_note → read_note with clearer descriptions,
move calculate out of rag.py into utility.py, and extract shared
duplicate detection into check_duplicate() helper.

Updates all downstream references in generation_task.py, quick_capture.py,
ToolCallCard.vue, and WorkspaceView.vue.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-12 13:38:45 -04:00
parent e95ad90055
commit 77339d5c58
12 changed files with 278 additions and 378 deletions
+1 -1
View File
@@ -218,7 +218,7 @@ async function confirmDuplicate() {
if (confirmState.value !== "idle") return;
confirmState.value = "creating";
const args = props.toolCall.arguments as Record<string, unknown>;
const isTask = props.toolCall.function === "create_task";
const isTask = !!(args.status);
const endpoint = isTask ? "/api/tasks" : "/api/notes";
const payload: Record<string, unknown> = {
title: args.title,
+5 -2
View File
@@ -75,8 +75,11 @@ watch(
noteEditorRef.value?.reload();
}
if (
["create_task", "update_task", "create_milestone", "update_milestone"].includes(tc.function) &&
tc.status === "success"
tc.status === "success" && (
["create_milestone", "update_milestone"].includes(tc.function) ||
(tc.function === "create_note" && tc.result?.data?.status) ||
(tc.function === "update_note" && tc.result?.data?.status !== undefined)
)
) {
taskPanelRef.value?.reload();
}
+1 -1
View File
@@ -21,7 +21,7 @@ quick_capture_bp = Blueprint("quick_capture", __name__, url_prefix="/api/quick-c
# Tools offered to the quick-capture endpoint. Excludes destructive ops,
# read-only queries, and conversational-only tools.
_CAPTURE_TOOL_NAMES = {"create_note", "create_task", "create_event", "update_note", "research_topic"}
_CAPTURE_TOOL_NAMES = {"create_note", "create_event", "update_note", "research_topic"}
_SYSTEM_PROMPT = """\
Today is {today}. You are a quick-capture assistant. The user has sent a short \
@@ -97,12 +97,10 @@ def _should_think(user_content: str, think_requested: bool) -> bool:
# Human-readable labels for each tool, shown in the status indicator
_TOOL_LABELS: dict[str, str] = {
"create_task": "Creating task",
"create_note": "Creating note",
"update_note": "Updating note",
"delete_note": "Deleting note",
"delete_task": "Deleting task",
"get_note": "Reading note",
"create_note": "Creating note/task",
"update_note": "Updating note/task",
"delete_note": "Deleting note/task",
"read_note": "Reading note",
"list_notes": "Listing notes",
"list_tasks": "Searching tasks",
"search_notes": "Searching notes (semantic)",
@@ -16,6 +16,7 @@ from fabledassistant.services.tools import ( # noqa: F401
rag,
rss,
tasks,
utility,
weather,
web,
)
@@ -75,3 +75,50 @@ def fuzzy_title_match(title: str, candidates, threshold: float = 0.82):
if r >= threshold and r > best_r:
best, best_r = c, r
return best, best_r
async def check_duplicate(
user_id: int,
title: str,
body: str,
is_task: bool,
confirmed: bool,
) -> dict | None:
"""Check for exact, fuzzy, and semantic duplicates. Returns error dict or None."""
from fabledassistant.services.notes import list_notes
item_label = "task" if is_task else "note"
existing, _ = await list_notes(user_id=user_id, q=title, is_task=is_task, limit=1)
exact = next((n for n in existing if n.title.lower() == title.lower()), None)
if exact is not None:
return {
"success": False,
"error": f"A {item_label} titled '{title}' already exists (id: {exact.id}). Use update_note to modify it instead of creating a duplicate.",
}
clean_q = _PUNCT_RE.sub(" ", title).strip()
candidates, _ = await list_notes(user_id=user_id, q=clean_q, is_task=is_task, limit=20)
near, ratio = fuzzy_title_match(title, candidates)
if near is not None:
return {
"success": False,
"requires_confirmation": True,
"similar_note": {"id": near.id, "title": near.title},
"error": f"A {item_label} with a very similar title '{near.title}' already exists (similarity: {ratio:.0%}). Ask the user to confirm before creating a separate entry.",
}
if not confirmed and len(body.strip()) >= 80:
from fabledassistant.services.embeddings import semantic_search_notes as _ssn
sem_query = f"{title}\n{body}".strip()
sem_hits = await _ssn(user_id, sem_query, limit=3, threshold=0.90, is_task=is_task)
if sem_hits:
best_score, best_note = sem_hits[0]
return {
"success": False,
"requires_confirmation": True,
"similar_note": {"id": best_note.id, "title": best_note.title},
"error": f"A {item_label} with very similar content exists: '{best_note.title}' (semantic similarity: {best_score:.0%}). Ask the user to confirm before creating a separate entry.",
}
return None
+82 -135
View File
@@ -7,15 +7,41 @@ from fabledassistant.services.tools._helpers import schedule_embedding
from fabledassistant.services.tools._registry import tool
def _build_person_body(meta: dict, extra_notes: str | None = None) -> str:
"""Build markdown body from person metadata."""
lines = []
for key, label in (("relationship", "Relationship"), ("phone", "Phone"), ("email", "Email"), ("birthday", "Birthday"), ("address", "Address")):
if meta.get(key):
lines.append(f"**{label}:** {meta[key]}")
if extra_notes:
lines.append(f"\n{extra_notes}")
return "\n".join(lines)
def _build_place_body(meta: dict, extra_notes: str | None = None) -> str:
"""Build markdown body from place metadata."""
lines = []
for key, label in (("address", "Address"), ("phone", "Phone"), ("hours", "Hours"), ("url", "Website")):
if meta.get(key):
lines.append(f"**{label}:** {meta[key]}")
if extra_notes:
lines.append(f"\n{extra_notes}")
return "\n".join(lines)
_PERSON_FIELDS = ("relationship", "phone", "email", "birthday", "address")
_PLACE_FIELDS = ("address", "phone", "hours", "url")
@tool(
name="create_person",
name="save_person",
description=(
"Save a person to the user's knowledge base. Use this when the user introduces "
"someone by name or asks to remember details about a person (family member, friend, "
"contact, colleague). Stored people are used to resolve names in future prompts."
"Save or update a person in the user's knowledge base. Creates a new entry if the "
"person doesn't exist, or updates an existing one. Use when the user introduces "
"someone by name or adds/corrects details about a known person."
),
parameters={
"name": {"type": "string", "description": "Full name of the person"},
"name": {"type": "string", "description": "Full name of the person (used to find existing entry or as new entry title)"},
"relationship": {"type": "string", "description": "Relationship to the user (e.g. daughter, dentist, colleague)"},
"phone": {"type": "string", "description": "Phone number"},
"email": {"type": "string", "description": "Email address"},
@@ -25,31 +51,36 @@ from fabledassistant.services.tools._registry import tool
},
required=["name"],
)
async def create_person_tool(*, user_id, arguments, **_ctx):
async def save_person_tool(*, user_id, arguments, **_ctx):
name = str(arguments.get("name", "")).strip()
if not name:
return {"success": False, "error": "name is required"}
meta: dict = {}
for field in ("relationship", "phone", "email", "birthday", "address"):
existing, _ = await list_notes(user_id=user_id, q=name, is_task=False, limit=10)
target = next((n for n in existing if n.note_type == "person"), None)
if target is not None:
meta = dict(target.entity_meta or {})
for field in _PERSON_FIELDS:
val = arguments.get(field)
if val is not None:
meta[field] = str(val)
extra_notes = arguments.get("notes")
body = _build_person_body(meta, extra_notes)
updated = await update_note(user_id=user_id, note_id=target.id, body=body, entity_meta=meta)
if updated is None:
return {"success": False, "error": "Failed to update person."}
schedule_embedding(target.id, user_id, target.title, body)
return {"success": True, "type": "person_updated", "data": {"id": target.id, "name": target.title}}
meta = {}
for field in _PERSON_FIELDS:
val = arguments.get(field)
if val:
meta[field] = str(val)
extra_notes = arguments.get("notes", "")
body_lines = []
if meta.get("relationship"):
body_lines.append(f"**Relationship:** {meta['relationship']}")
if meta.get("phone"):
body_lines.append(f"**Phone:** {meta['phone']}")
if meta.get("email"):
body_lines.append(f"**Email:** {meta['email']}")
if meta.get("birthday"):
body_lines.append(f"**Birthday:** {meta['birthday']}")
if meta.get("address"):
body_lines.append(f"**Address:** {meta['address']}")
if extra_notes:
body_lines.append(f"\n{extra_notes}")
body = _build_person_body(meta, arguments.get("notes", ""))
note = await create_note(
user_id=user_id, title=name, body="\n".join(body_lines),
user_id=user_id, title=name, body=body,
tags=["person"], note_type="person", entity_meta=meta,
)
schedule_embedding(note.id, user_id, name, note.body)
@@ -57,57 +88,12 @@ async def create_person_tool(*, user_id, arguments, **_ctx):
@tool(
name="update_person",
description="Update details about a saved person. Use this when the user corrects or adds new information about someone already in the knowledge base.",
parameters={
"query": {"type": "string", "description": "Name or keyword to find the person"},
"relationship": {"type": "string", "description": "Updated relationship to the user"},
"phone": {"type": "string", "description": "Updated phone number"},
"email": {"type": "string", "description": "Updated email address"},
"birthday": {"type": "string", "description": "Updated birthday in YYYY-MM-DD format"},
"address": {"type": "string", "description": "Updated home or mailing address"},
"notes": {"type": "string", "description": "Updated free-form notes about this person"},
},
required=["query"],
)
async def update_person_tool(*, user_id, arguments, **_ctx):
query = str(arguments.get("query", "")).strip()
if not query:
return {"success": False, "error": "query is required"}
existing, _ = await list_notes(user_id=user_id, q=query, is_task=False, limit=10)
target = next((n for n in existing if n.note_type == "person"), None)
if target is None:
return {"success": False, "error": f"No person found matching '{query}'. Use create_person to add them."}
meta = dict(target.entity_meta or {})
for field in ("relationship", "phone", "email", "birthday", "address"):
val = arguments.get(field)
if val is not None:
meta[field] = str(val)
body_lines = []
if meta.get("relationship"):
body_lines.append(f"**Relationship:** {meta['relationship']}")
if meta.get("phone"):
body_lines.append(f"**Phone:** {meta['phone']}")
if meta.get("email"):
body_lines.append(f"**Email:** {meta['email']}")
if meta.get("birthday"):
body_lines.append(f"**Birthday:** {meta['birthday']}")
if meta.get("address"):
body_lines.append(f"**Address:** {meta['address']}")
extra_notes = arguments.get("notes")
if extra_notes is not None:
body_lines.append(f"\n{extra_notes}")
new_body = "\n".join(body_lines)
updated = await update_note(user_id=user_id, note_id=target.id, body=new_body, entity_meta=meta)
if updated is None:
return {"success": False, "error": "Failed to update person."}
schedule_embedding(target.id, user_id, target.title, new_body)
return {"success": True, "type": "person_updated", "data": {"id": target.id, "name": target.title}}
@tool(
name="create_place",
description="Save a named location to the user's knowledge base. Use this when the user wants to remember a place (business, venue, address). Stored places are used to auto-fill locations in calendar events and other prompts.",
name="save_place",
description=(
"Save or update a named location in the user's knowledge base. Creates a new entry "
"if the place doesn't exist, or updates an existing one. Use when the user wants to "
"remember or correct details about a place (business, venue, address)."
),
parameters={
"name": {"type": "string", "description": "Name of the place (e.g. 'Dr. Smith Dental', 'Costco')"},
"address": {"type": "string", "description": "Street address"},
@@ -118,81 +104,42 @@ async def update_person_tool(*, user_id, arguments, **_ctx):
},
required=["name"],
)
async def create_place_tool(*, user_id, arguments, **_ctx):
async def save_place_tool(*, user_id, arguments, **_ctx):
name = str(arguments.get("name", "")).strip()
if not name:
return {"success": False, "error": "name is required"}
existing, _ = await list_notes(user_id=user_id, q=name, is_task=False, limit=10)
target = next((n for n in existing if n.note_type == "place"), None)
if target is not None:
meta = dict(target.entity_meta or {})
for field in _PLACE_FIELDS:
val = arguments.get(field)
if val is not None:
meta[field] = str(val)
extra_notes = arguments.get("notes")
body = _build_place_body(meta, extra_notes)
updated = await update_note(user_id=user_id, note_id=target.id, body=body, entity_meta=meta)
if updated is None:
return {"success": False, "error": "Failed to update place."}
schedule_embedding(target.id, user_id, target.title, body)
return {"success": True, "type": "place_updated", "data": {"id": target.id, "name": target.title}}
meta = {}
for field in ("address", "phone", "hours", "url"):
for field in _PLACE_FIELDS:
val = arguments.get(field)
if val:
meta[field] = str(val)
extra_notes = arguments.get("notes", "")
body_lines = []
if meta.get("address"):
body_lines.append(f"**Address:** {meta['address']}")
if meta.get("phone"):
body_lines.append(f"**Phone:** {meta['phone']}")
if meta.get("hours"):
body_lines.append(f"**Hours:** {meta['hours']}")
if meta.get("url"):
body_lines.append(f"**Website:** {meta['url']}")
if extra_notes:
body_lines.append(f"\n{extra_notes}")
body = _build_place_body(meta, arguments.get("notes", ""))
note = await create_note(
user_id=user_id, title=name, body="\n".join(body_lines),
user_id=user_id, title=name, body=body,
tags=["place"], note_type="place", entity_meta=meta,
)
schedule_embedding(note.id, user_id, name, note.body)
return {"success": True, "type": "place", "data": {"id": note.id, "name": name}}
@tool(
name="update_place",
description="Update details about a saved place. Use this when the user corrects or adds new information about a location already in the knowledge base.",
parameters={
"query": {"type": "string", "description": "Name or keyword to find the place"},
"address": {"type": "string", "description": "Updated street address"},
"phone": {"type": "string", "description": "Updated phone number"},
"hours": {"type": "string", "description": "Updated opening hours"},
"url": {"type": "string", "description": "Updated website URL"},
"notes": {"type": "string", "description": "Updated free-form notes about this place"},
},
required=["query"],
)
async def update_place_tool(*, user_id, arguments, **_ctx):
query = str(arguments.get("query", "")).strip()
if not query:
return {"success": False, "error": "query is required"}
existing, _ = await list_notes(user_id=user_id, q=query, is_task=False, limit=10)
target = next((n for n in existing if n.note_type == "place"), None)
if target is None:
return {"success": False, "error": f"No place found matching '{query}'. Use create_place to add it."}
meta = dict(target.entity_meta or {})
for field in ("address", "phone", "hours", "url"):
val = arguments.get(field)
if val is not None:
meta[field] = str(val)
body_lines = []
if meta.get("address"):
body_lines.append(f"**Address:** {meta['address']}")
if meta.get("phone"):
body_lines.append(f"**Phone:** {meta['phone']}")
if meta.get("hours"):
body_lines.append(f"**Hours:** {meta['hours']}")
if meta.get("url"):
body_lines.append(f"**Website:** {meta['url']}")
extra_notes = arguments.get("notes")
if extra_notes is not None:
body_lines.append(f"\n{extra_notes}")
new_body = "\n".join(body_lines)
updated = await update_note(user_id=user_id, note_id=target.id, body=new_body, entity_meta=meta)
if updated is None:
return {"success": False, "error": "Failed to update place."}
schedule_embedding(target.id, user_id, target.title, new_body)
return {"success": True, "type": "place_updated", "data": {"id": target.id, "name": target.title}}
@tool(
name="create_list",
description="Create a named list (shopping list, to-do list, packing list, etc.). Items are stored as a markdown task list. Use add_to_list to add items to an existing list.",
+92 -56
View File
@@ -7,8 +7,7 @@ import logging
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
from fabledassistant.services.tools._helpers import (
_PUNCT_RE,
fuzzy_title_match,
check_duplicate,
parse_due_date,
resolve_project,
schedule_embedding,
@@ -20,63 +19,100 @@ logger = logging.getLogger(__name__)
@tool(
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.",
description=(
"Create a new note or task. "
"For a knowledge note, omit the status field. "
"For an actionable task (todo, reminder, action item), set status to 'todo'. "
"Use this whenever the user asks to write down, save, record, or add a task/todo."
),
parameters={
"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.'},
"project": {"type": "string", "description": "Optional project name to assign this note to. Only set this if the user explicitly named a project. Do NOT infer a project from the note content or context."},
"confirmed": {"type": "boolean", "description": "Set to true only when the user has explicitly confirmed creation after being warned about a similar existing note."},
"title": {"type": "string", "description": "The title"},
"body": {"type": "string", "description": "Content in markdown"},
"tags": {"type": "array", "items": {"type": "string"}, "description": 'Tags (without # prefix, hyphens for multi-word: ["science-fiction", "story/idea"]). Do NOT embed #tags in the body.'},
"project": {"type": "string", "description": "Optional project name. Only set this if the user explicitly named a project. Do NOT infer a project from the content or context."},
"status": {"type": "string", "enum": ["todo", "in_progress", "done", "cancelled"], "description": "Set to 'todo' to create a task. Omit entirely for a knowledge note."},
"due_date": {"type": "string", "description": "Due date in YYYY-MM-DD format (tasks only)"},
"priority": {"type": "string", "enum": ["none", "low", "medium", "high"], "description": "Priority level (tasks only, default: none)"},
"parent_task": {"type": "string", "description": "Title of a parent task to make this a sub-task of"},
"milestone": {"type": "string", "description": "Milestone title within the project to assign to"},
"recurrence_rule": {"type": "object", "description": 'Recurrence rule (tasks only). Interval form: {"type":"interval","every":3,"unit":"month"} (units: day/week/month/year). Calendar form: {"type":"calendar","unit":"month","day_of_month":1}. Annual: add "month":6.'},
"confirmed": {"type": "boolean", "description": "Set to true only when the user has explicitly confirmed creation after being warned about a similar existing item."},
},
required=["title"],
)
async def create_note_tool(*, user_id, arguments, **_ctx):
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 = []
title = arguments.get("title", "Untitled")
body = arguments.get("body", "")
tags = arguments.get("tags", [])
if not isinstance(title, str):
return {"success": False, "error": "title must be a string. Call create_note once per item."}
if not isinstance(body, str):
body = ""
if not isinstance(tags, list):
tags = []
is_task = "status" in arguments and arguments["status"] is not None
status = arguments.get("status", "todo") if is_task else None
project_name = arguments.get("project")
milestone_name = arguments.get("milestone")
parent_task_name = arguments.get("parent_task")
project_id = None
milestone_id = None
if project_name:
proj = await resolve_project(user_id, project_name)
if proj is None:
return {"success": False, "error": f"Project '{project_name}' not found. Use list_projects to find the correct project name and retry with the exact name."}
project_id = proj.id
if milestone_name:
from fabledassistant.services.milestones import get_milestone_by_title as _gmbt
ms = await _gmbt(user_id, proj.id, milestone_name)
if ms is None:
return {"success": False, "error": f"Milestone '{milestone_name}' not found in project '{proj.title}'. Use list_milestones to see available milestones."}
milestone_id = ms.id
existing_notes, _ = await list_notes(user_id=user_id, q=note_title, is_task=False, limit=1)
exact = next((n for n in existing_notes if n.title.lower() == note_title.lower()), None)
if exact is not None:
return {"success": False, "error": f"A note titled '{note_title}' already exists (id: {exact.id}). Use update_note to modify it instead of creating a duplicate."}
clean_q = _PUNCT_RE.sub(" ", note_title).strip()
candidates, _ = await list_notes(user_id=user_id, q=clean_q, is_task=False, limit=20)
near, ratio = fuzzy_title_match(note_title, candidates)
if near is not None:
return {"success": False, "requires_confirmation": True, "similar_note": {"id": near.id, "title": near.title}, "error": f"A note with a very similar title '{near.title}' already exists (similarity: {ratio:.0%}). Ask the user to confirm before creating a separate entry."}
if not arguments.get("confirmed") and len(note_body.strip()) >= 80:
from fabledassistant.services.embeddings import semantic_search_notes as _ssn
sem_query = f"{note_title}\n{note_body}".strip()
sem_hits = await _ssn(user_id, sem_query, limit=3, threshold=0.90, is_task=False)
if sem_hits:
best_score, best_note = sem_hits[0]
return {"success": False, "requires_confirmation": True, "similar_note": {"id": best_note.id, "title": best_note.title}, "error": f"A note with very similar content exists: '{best_note.title}' (semantic similarity: {best_score:.0%}). Ask the user to confirm before creating a separate entry."}
dup = await check_duplicate(user_id, title, body, is_task=is_task, confirmed=bool(arguments.get("confirmed")))
if dup is not None:
return dup
note = await create_note(
user_id=user_id,
title=note_title,
body=note_body,
tags=note_tags,
title=title,
body=body,
tags=tags,
status=status,
priority=arguments.get("priority", "none") if is_task else None,
due_date=parse_due_date(arguments.get("due_date")) if is_task else None,
recurrence_rule=arguments.get("recurrence_rule") if is_task else None,
project_id=project_id,
milestone_id=milestone_id,
)
suggested = await suggest_tags(user_id, note_title, note_body)
schedule_embedding(note.id, user_id, note_title, note_body)
if parent_task_name:
parent_notes, _ = await list_notes(user_id=user_id, q=parent_task_name, is_task=True, limit=1)
if parent_notes:
note = await update_note(user_id, note.id, parent_id=parent_notes[0].id)
suggested = await suggest_tags(user_id, title, body)
schedule_embedding(note.id, user_id, title, body)
if is_task:
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,
"project_id": note.project_id,
"milestone_id": note.milestone_id,
"parent_id": note.parent_id,
},
"suggested_tags": suggested,
}
return {
"success": True,
"type": "note",
@@ -100,7 +136,7 @@ async def create_note_tool(*, user_id, arguments, **_ctx):
"tag_mode": {"type": "string", "enum": ["add", "remove", "replace"], "description": "'add' appends tags without duplicates (default), 'remove' removes listed tags, 'replace' sets tags to exactly this list (destructive — avoid unless explicitly requested)"},
"project": {"type": "string", "description": "Optional project name to assign this note/task to (set to empty string to remove project)"},
"milestone": {"type": "string", "description": "Optional milestone title within the project to assign this task to (set to empty string to remove milestone)"},
"recurrence_rule": {"type": "object", "description": "Set or update recurrence (same format as create_task). Pass null to remove."},
"recurrence_rule": {"type": "object", "description": 'Set or update recurrence. Interval form: {"type":"interval","every":3,"unit":"month"}. Calendar form: {"type":"calendar","unit":"month","day_of_month":1}. Pass null to remove.'},
},
required=["query"],
)
@@ -199,7 +235,7 @@ async def update_note_tool(*, user_id, arguments, **_ctx):
@tool(
name="search_notes",
description="Search notes and tasks by semantic meaning — finds content by concept or theme even when exact keywords don't appear.",
description="Find notes or tasks by meaning. Returns a ranked list of matches with short previews. Use this when looking for items on a topic but you don't know the exact title. For the full body of a known item, use read_note instead.",
parameters={
"query": {"type": "string", "description": "A natural-language description of what you're looking for — concepts, themes, topics, or keywords"},
"type": {"type": "string", "enum": ["note", "task"], "description": "Restrict results to only notes or only tasks. Omit to search both."},
@@ -279,10 +315,10 @@ async def search_notes_tool(*, user_id, arguments, **_ctx):
@tool(
name="get_note",
description="Retrieve the full content of a specific note or task. Use this when the user asks to read, view, or check what a particular note or task says. Returns the complete body.",
name="read_note",
description="Read the full content of one specific note or task. Use when you know which item you want and need its complete body text. For discovering items by topic, use search_notes instead.",
parameters={
"query": {"type": "string", "description": "Title or keyword to identify the note"},
"query": {"type": "string", "description": "Title or keyword to identify the note or task"},
},
required=["query"],
read_only=True,
@@ -305,7 +341,7 @@ async def get_note_tool(*, user_id, arguments, **_ctx):
@tool(
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, tag, or project. For tasks, use list_tasks instead.",
description="Browse recent notes (not tasks) with optional filters. Returns titles and short previews. Use when the user asks 'show my notes' or wants to browse by tag, project, or recency. For tasks use list_tasks; for topic search use search_notes.",
parameters={
"q": {"type": "string", "description": "Optional keyword filter"},
"tags": {"type": "array", "items": {"type": "string"}, "description": "Filter notes that have ALL of these tags"},
@@ -355,10 +391,10 @@ async def list_notes_tool(*, user_id, arguments, **_ctx):
@tool(
name="delete_note",
description="Delete a note permanently. Use ONLY when the user explicitly asks to delete or remove a note. Always confirm with the user first — this cannot be undone.",
description="Delete a note or task permanently. Use ONLY when the user explicitly asks to delete or remove an item. Always confirm with the user first — this cannot be undone.",
parameters={
"query": {"type": "string", "description": "Title or keyword to find the note to delete"},
"confirmed": {"type": "boolean", "description": "Must be true — only set after the user has explicitly confirmed they want this note deleted."},
"query": {"type": "string", "description": "Title or keyword to find the note or task to delete"},
"confirmed": {"type": "boolean", "description": "Must be true — only set after the user has explicitly confirmed they want this item deleted."},
},
required=["query"],
)
@@ -366,19 +402,19 @@ async def delete_note_tool(*, user_id, arguments, **_ctx):
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)
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}'."}
return {"success": False, "error": f"No note or task 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."}
if not arguments.get("confirmed"):
item_type = "task" if note.status is not None else "note"
return {
"success": False,
"requires_confirmation": True,
"error": f"Deleting '{note.title}' is permanent. Ask the user to confirm, then retry with confirmed=true.",
"error": f"Deleting {item_type} '{note.title}' is permanent. Ask the user to confirm, then retry with confirmed=true.",
}
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}}
return {"success": False, "error": "Failed to delete."}
item_type = "task" if note.status is not None else "note"
return {"success": True, "type": f"{item_type}_deleted", "data": {"id": note.id, "title": note.title}}
+1 -30
View File
@@ -1,4 +1,4 @@
"""RAG scope and calculation tools."""
"""RAG scope tool."""
from __future__ import annotations
@@ -32,32 +32,3 @@ async def set_rag_scope_tool(*, user_id, arguments, conv_id=None, workspace_proj
from fabledassistant.services.chat import update_conversation
await update_conversation(user_id, conv_id, rag_project_id=project_id)
return {"success": True, "type": "rag_scope_set", "scope_label": scope_label}
@tool(
name="calculate",
description=(
"Evaluate a mathematical expression and return the exact result. Use this for any "
"arithmetic, percentages, unit conversions, or multi-step calculations where precision "
"matters. Supports standard math operators (+, -, *, /, **, %) and all Python math "
"module functions (sqrt, log, sin, cos, floor, ceil, etc.)."
),
parameters={
"expression": {"type": "string", "description": "A valid Python math expression (e.g. '(450 * 1.13) / 12', 'sqrt(144)', 'log(1000, 10)')"},
},
required=["expression"],
)
async def calculate_tool(*, user_id, arguments, **_ctx):
import math as _math
expr = str(arguments.get("expression", "")).strip()
if not expr:
return {"success": False, "error": "expression is required"}
allowed_names = {k: v for k, v in vars(_math).items() if not k.startswith("_")}
allowed_names["abs"] = abs
allowed_names["round"] = round
try:
result = eval(expr, {"__builtins__": {}}, allowed_names) # noqa: S307
except Exception as calc_err:
return {"success": False, "error": f"Could not evaluate expression: {calc_err}"}
return {"success": True, "type": "calculation", "data": {"expression": expr, "result": result}}
+2 -139
View File
@@ -1,121 +1,15 @@
"""Task tools: create, list, delete, log work."""
"""Task tools: list, log work."""
from __future__ import annotations
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
from fabledassistant.services.notes import get_note_by_title, list_notes
from fabledassistant.services.tools._helpers import (
_PUNCT_RE,
fuzzy_title_match,
parse_due_date,
resolve_project,
schedule_embedding,
)
from fabledassistant.services.tools._registry import tool
@tool(
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={
"title": {"type": "string", "description": "The task title"},
"body": {"type": "string", "description": "Optional task description or details"},
"status": {"type": "string", "enum": ["todo", "in_progress", "done", "cancelled"], "description": "Initial task status (default: todo)"},
"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"},
"tags": {"type": "array", "items": {"type": "string"}, "description": "Tags for the task (without # prefix)"},
"project": {"type": "string", "description": "Optional project name to assign this task to. Only set this if the user explicitly named a project. Do NOT infer a project from the task content or context."},
"parent_task": {"type": "string", "description": "Optional title of a parent task to make this a sub-task of"},
"milestone": {"type": "string", "description": "Optional milestone title within the project to assign this task to"},
"confirmed": {"type": "boolean", "description": "Set to true only when the user has explicitly confirmed creation after being warned about a similar existing task."},
"recurrence_rule": {"type": "object", "description": 'Optional recurrence. Interval form: {"type":"interval","every":3,"unit":"month"} (units: day/week/month/year). Calendar form: {"type":"calendar","unit":"month","day_of_month":1}. Annual: add "month":6.'},
},
required=["title"],
)
async def create_task_tool(*, user_id, arguments, **_ctx):
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 = ""
project_name = arguments.get("project")
milestone_name = arguments.get("milestone")
parent_task_name = arguments.get("parent_task")
pre_fields: dict = {}
if project_name:
proj = await resolve_project(user_id, project_name)
if proj is None:
return {"success": False, "error": f"Project '{project_name}' not found. Use list_projects to find the correct project name and retry with the exact name."}
pre_fields["project_id"] = proj.id
if milestone_name:
from fabledassistant.services.milestones import get_milestone_by_title as _gmbt
ms = await _gmbt(user_id, proj.id, milestone_name)
if ms is None:
return {"success": False, "error": f"Milestone '{milestone_name}' not found in project '{proj.title}'. Use list_milestones to see available milestones."}
pre_fields["milestone_id"] = ms.id
existing_tasks, _ = await list_notes(user_id=user_id, q=task_title, is_task=True, limit=1)
exact = next((t for t in existing_tasks if t.title.lower() == task_title.lower()), None)
if exact is not None:
return {"success": False, "error": f"A task titled '{task_title}' already exists (id: {exact.id}). Use update_note to modify it instead of creating a duplicate."}
clean_q = _PUNCT_RE.sub(" ", task_title).strip()
candidates, _ = await list_notes(user_id=user_id, q=clean_q, is_task=True, limit=20)
near, ratio = fuzzy_title_match(task_title, candidates)
if near is not None:
return {"success": False, "requires_confirmation": True, "similar_note": {"id": near.id, "title": near.title}, "error": f"A task with a very similar title '{near.title}' already exists (similarity: {ratio:.0%}). Ask the user to confirm before creating a separate entry."}
if not arguments.get("confirmed") and len(task_body.strip()) >= 80:
from fabledassistant.services.embeddings import semantic_search_notes as _ssn
sem_query = f"{task_title}\n{task_body}".strip()
sem_hits = await _ssn(user_id, sem_query, limit=3, threshold=0.90, is_task=True)
if sem_hits:
best_score, best_note = sem_hits[0]
return {"success": False, "requires_confirmation": True, "similar_note": {"id": best_note.id, "title": best_note.title}, "error": f"A task with very similar content exists: '{best_note.title}' (semantic similarity: {best_score:.0%}). Ask the user to confirm before creating a separate entry."}
task_tags = arguments.get("tags", [])
if not isinstance(task_tags, list):
task_tags = []
task_status = arguments.get("status", "todo")
note = await create_note(
user_id=user_id,
title=task_title,
body=task_body,
tags=task_tags,
status=task_status,
priority=arguments.get("priority", "none"),
due_date=parse_due_date(arguments.get("due_date")),
project_id=pre_fields.get("project_id"),
milestone_id=pre_fields.get("milestone_id"),
)
suggested = await suggest_tags(user_id, task_title, task_body)
schedule_embedding(note.id, user_id, task_title, task_body)
update_fields: dict = {}
if parent_task_name:
parent_notes, _ = await list_notes(user_id=user_id, q=parent_task_name, is_task=True, limit=1)
if parent_notes:
update_fields["parent_id"] = parent_notes[0].id
if update_fields:
note = await update_note(user_id, note.id, **update_fields)
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,
"project_id": note.project_id,
"milestone_id": note.milestone_id,
"parent_id": note.parent_id,
},
"suggested_tags": suggested,
}
@tool(
name="list_tasks",
description="List tasks with optional filters. For overdue tasks, set due_before to today's date.",
@@ -184,37 +78,6 @@ async def list_tasks_tool(*, user_id, arguments, **_ctx):
}
@tool(
name="delete_task",
description="Delete a task permanently. Use ONLY when the user explicitly asks to delete or remove a task. Always confirm with the user first — this cannot be undone.",
parameters={
"query": {"type": "string", "description": "Title or keyword to find the task to delete"},
"confirmed": {"type": "boolean", "description": "Must be true — only set after the user has explicitly confirmed they want this task deleted."},
},
required=["query"],
)
async def delete_task_tool(*, user_id, arguments, **_ctx):
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."}
if not arguments.get("confirmed"):
return {
"success": False,
"requires_confirmation": True,
"error": f"Deleting '{note.title}' is permanent. Ask the user to confirm, then retry with confirmed=true.",
}
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}}
@tool(
name="log_work",
description="Add a work log entry to a task to record progress, work done, or time spent. Use this when the user says they worked on, completed, or spent time on a task.",
@@ -0,0 +1,34 @@
"""General-purpose utility tools."""
from __future__ import annotations
from fabledassistant.services.tools._registry import tool
@tool(
name="calculate",
description=(
"Evaluate a mathematical expression and return the exact result. Use this for any "
"arithmetic, percentages, unit conversions, or multi-step calculations where precision "
"matters. Supports standard math operators (+, -, *, /, **, %) and all Python math "
"module functions (sqrt, log, sin, cos, floor, ceil, etc.)."
),
parameters={
"expression": {"type": "string", "description": "A valid Python math expression (e.g. '(450 * 1.13) / 12', 'sqrt(144)', 'log(1000, 10)')"},
},
required=["expression"],
)
async def calculate_tool(*, user_id, arguments, **_ctx):
import math as _math
expr = str(arguments.get("expression", "")).strip()
if not expr:
return {"success": False, "error": "expression is required"}
allowed_names = {k: v for k, v in vars(_math).items() if not k.startswith("_")}
allowed_names["abs"] = abs
allowed_names["round"] = round
try:
result = eval(expr, {"__builtins__": {}}, allowed_names) # noqa: S307
except Exception as calc_err:
return {"success": False, "error": f"Could not evaluate expression: {calc_err}"}
return {"success": True, "type": "calculation", "data": {"expression": expr, "result": result}}
+8 -8
View File
@@ -12,10 +12,9 @@ logger = logging.getLogger(__name__)
@tool(
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."
"Quick web lookup — returns search result snippets for factual questions, current events, "
"definitions, or version numbers. No note is saved. Use research_topic when the user wants "
"a comprehensive written report saved as a note."
),
parameters={
"query": {"type": "string", "description": "The search query"},
@@ -40,9 +39,10 @@ async def search_web_tool(*, user_id, arguments, **_ctx):
@tool(
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 30120 seconds."
"Deep web research — searches multiple sources, synthesizes findings, and saves the result "
"as a structured note with sections and citations. Use when the user says 'research', "
"'look into', or wants a comprehensive write-up. Takes 30120 seconds. "
"For a quick factual answer without saving a note, use search_web."
),
parameters={
"topic": {"type": "string", "description": "The topic or question to research"},
@@ -64,7 +64,7 @@ async def research_topic_tool(*, user_id, arguments, **_ctx):
@tool(
name="search_images",
description="Search and display images inline. Use ONLY when user explicitly asks to see or show an image or photo. Not for factual questions.",
description="Search and display images inline. Use ONLY when the user explicitly asks to see, show, or find an image or photo. Not for factual questions — use search_web for those.",
parameters={
"query": {"type": "string", "description": "The image search query"},
},