bvandeusen/FabledScribe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 50 Wiki Activity

feat(tools): confirmed guard for deletes, update_person/place, get/update_profile, calculate

Browse Source 
- delete_note / delete_task: add confirmed parameter + requires_confirmation guard
  (find the note first, then ask, consistent with create_note/task pattern)
- get_note: description now mentions notes AND tasks
- update_person / update_place: new tools to update existing entity notes in-place
- get_profile / update_profile: surface and edit the user's stored profile
  (expertise, tone, response style, job title, interests)
- calculate: eval math expressions via Python math module; solves precision issues
  on multi-step arithmetic and supports sqrt/log/trig/etc.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Van Deusen
2026-04-05 22:58:14 -04:00
parent eb92b2a976
commit a171210224
1 changed files with 275 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/fabledassistant/services/tools.py
+275 -4
View File
@@ -258,8 +258,8 @@ _CORE_TOOLS = [
"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."
"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."
),
"parameters": {
"type": "object",
@@ -316,7 +316,7 @@ _CORE_TOOLS = [
"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."
"Always confirm with the user first — this cannot be undone."
),
"parameters": {
"type": "object",
@@ -325,6 +325,10 @@ _CORE_TOOLS = [
"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.",
},
},
"required": ["query"],
},
@@ -336,7 +340,7 @@ _CORE_TOOLS = [
"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."
"Always confirm with the user first — this cannot be undone."
),
"parameters": {
"type": "object",
@@ -345,6 +349,10 @@ _CORE_TOOLS = [
"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"],
},
@@ -937,6 +945,28 @@ _RAG_TOOLS = [
},
},
},
{
"type": "function",
"function": {
"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": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "A valid Python math expression (e.g. '(450 * 1.13) / 12', 'sqrt(144)', 'log(1000, 10)')",
},
},
"required": ["expression"],
},
},
},
]
_ENTITY_TOOLS = [
@@ -964,6 +994,29 @@ _ENTITY_TOOLS = [
},
},
},
{
"type": "function",
"function": {
"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": {
"type": "object",
"properties": {
"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"],
},
},
},
{
"type": "function",
"function": {
@@ -987,6 +1040,28 @@ _ENTITY_TOOLS = [
},
},
},
{
"type": "function",
"function": {
"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": {
"type": "object",
"properties": {
"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"],
},
},
},
{
"type": "function",
"function": {
@@ -1051,12 +1126,71 @@ _ENTITY_TOOLS = [
]
_PROFILE_TOOLS = [
{
"type": "function",
"function": {
"name": "get_profile",
"description": (
"Retrieve the user's stored profile: name, job title, industry, expertise level, "
"preferred response style, tone, and interests. Use this when you need to personalise "
"a response and the user's profile hasn't already been injected into context."
),
"parameters": {
"type": "object",
"properties": {},
},
},
},
{
"type": "function",
"function": {
"name": "update_profile",
"description": (
"Update the user's stored profile when they share personal information: their name, "
"job, industry, expertise, preferred response style, tone, or interests. "
"Only set fields the user has explicitly mentioned."
),
"parameters": {
"type": "object",
"properties": {
"display_name": {"type": "string", "description": "User's preferred display name"},
"job_title": {"type": "string", "description": "Job title (e.g. 'Software Engineer')"},
"industry": {"type": "string", "description": "Industry (e.g. 'Healthcare', 'Finance')"},
"expertise_level": {
"type": "string",
"enum": ["novice", "intermediate", "expert"],
"description": "User's general expertise level",
},
"response_style": {
"type": "string",
"enum": ["concise", "balanced", "detailed"],
"description": "Preferred response length/depth",
},
"tone": {
"type": "string",
"enum": ["casual", "professional", "technical"],
"description": "Preferred communication tone",
},
"interests": {
"type": "array",
"items": {"type": "string"},
"description": "List of topics or hobbies the user is interested in",
},
},
},
},
},
]
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)
tools.extend(_URL_TOOLS)
tools.extend(_RAG_TOOLS)
tools.extend(_ENTITY_TOOLS)
tools.extend(_PROFILE_TOOLS)
if await is_caldav_configured(user_id):
tools.extend(_CALDAV_TOOLS)
if Config.searxng_enabled():
@@ -1698,6 +1832,12 @@ async def execute_tool(
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"):
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 note."}
@@ -1717,6 +1857,12 @@ async def execute_tool(
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."}
@@ -2172,6 +2318,72 @@ async def execute_tool(
_schedule_embedding(note.id, user_id, name, note.body)
return {"success": True, "type": "place", "data": {"id": note.id, "name": name}}
elif tool_name == "update_person":
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}}
elif tool_name == "update_place":
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}}
elif tool_name == "create_list":
name = str(arguments.get("name", "")).strip()
if not name:
@@ -2235,6 +2447,65 @@ async def execute_tool(
_schedule_embedding(target.id, user_id, target.title, cleaned)
return {"success": True, "type": "list_cleared", "data": {"id": target.id, "name": target.title}}
elif tool_name == "get_profile":
from fabledassistant.services.user_profile import get_profile as _get_profile
profile = await _get_profile(user_id)
return {
"success": True,
"type": "profile",
"data": {
"display_name": profile.display_name or "",
"job_title": profile.job_title or "",
"industry": profile.industry or "",
"expertise_level": profile.expertise_level or "intermediate",
"response_style": profile.response_style or "balanced",
"tone": profile.tone or "casual",
"interests": profile.interests or [],
},
}
elif tool_name == "update_profile":
from fabledassistant.services.user_profile import update_profile as _update_profile, VALID_EXPERTISE, VALID_STYLES, VALID_TONES
data: dict = {}
for field in ("display_name", "job_title", "industry"):
val = arguments.get(field)
if val is not None:
data[field] = str(val)
expertise = arguments.get("expertise_level")
if expertise in VALID_EXPERTISE:
data["expertise_level"] = expertise
style = arguments.get("response_style")
if style in VALID_STYLES:
data["response_style"] = style
tone = arguments.get("tone")
if tone in VALID_TONES:
data["tone"] = tone
interests = arguments.get("interests")
if isinstance(interests, list):
data["interests"] = [str(i) for i in interests if str(i).strip()]
if not data:
return {"success": False, "error": "No valid fields provided to update."}
profile = await _update_profile(user_id, data)
return {
"success": True,
"type": "profile_updated",
"data": {"fields_updated": list(data.keys())},
}
elif tool_name == "calculate":
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}}
else:
return {"success": False, "error": f"Unknown tool: {tool_name}"}