bvandeusenandClaude Opus 4.6
77339d5c58
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>
232 lines
10 KiB
Python
232 lines
10 KiB
Python
"""Entity tools: people, places, and lists."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fabledassistant.services.notes import create_note, list_notes, update_note
|
|
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="save_person",
|
|
description=(
|
|
"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 (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"},
|
|
"birthday": {"type": "string", "description": "Birthday in YYYY-MM-DD format"},
|
|
"address": {"type": "string", "description": "Home or mailing address"},
|
|
"notes": {"type": "string", "description": "Any additional free-form notes about this person"},
|
|
},
|
|
required=["name"],
|
|
)
|
|
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"}
|
|
|
|
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)
|
|
body = _build_person_body(meta, arguments.get("notes", ""))
|
|
note = await create_note(
|
|
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)
|
|
return {"success": True, "type": "person", "data": {"id": note.id, "name": name}}
|
|
|
|
|
|
@tool(
|
|
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"},
|
|
"phone": {"type": "string", "description": "Phone number"},
|
|
"hours": {"type": "string", "description": "Opening hours (e.g. 'Mon-Fri 9am-5pm')"},
|
|
"url": {"type": "string", "description": "Website URL"},
|
|
"notes": {"type": "string", "description": "Any additional free-form notes about this place"},
|
|
},
|
|
required=["name"],
|
|
)
|
|
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 _PLACE_FIELDS:
|
|
val = arguments.get(field)
|
|
if val:
|
|
meta[field] = str(val)
|
|
body = _build_place_body(meta, arguments.get("notes", ""))
|
|
note = await create_note(
|
|
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="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.",
|
|
parameters={
|
|
"name": {"type": "string", "description": "List name (e.g. 'Grocery List', 'Hardware Store')"},
|
|
"items": {"type": "array", "items": {"type": "string"}, "description": "Initial list items (unchecked)"},
|
|
"store": {"type": "string", "description": "Optional associated store or place name"},
|
|
"tags": {"type": "array", "items": {"type": "string"}, "description": "Tags for the list"},
|
|
},
|
|
required=["name"],
|
|
)
|
|
async def create_list_tool(*, user_id, arguments, **_ctx):
|
|
name = str(arguments.get("name", "")).strip()
|
|
if not name:
|
|
return {"success": False, "error": "name is required"}
|
|
items = arguments.get("items") or []
|
|
if not isinstance(items, list):
|
|
items = []
|
|
body = "\n".join(f"- [ ] {item}" for item in items if str(item).strip())
|
|
meta = {}
|
|
store = arguments.get("store")
|
|
if store:
|
|
meta["store"] = str(store)
|
|
tags = arguments.get("tags") or []
|
|
if not isinstance(tags, list):
|
|
tags = []
|
|
note = await create_note(
|
|
user_id=user_id, title=name, body=body,
|
|
tags=tags, note_type="list", entity_meta=meta,
|
|
)
|
|
schedule_embedding(note.id, user_id, name, body)
|
|
return {"success": True, "type": "list", "data": {"id": note.id, "name": name, "item_count": len(items)}}
|
|
|
|
|
|
@tool(
|
|
name="add_to_list",
|
|
description="Add one or more items to an existing list. Items are appended as unchecked entries.",
|
|
parameters={
|
|
"list_name": {"type": "string", "description": "Name of the list to add items to"},
|
|
"items": {"type": "array", "items": {"type": "string"}, "description": "Items to add"},
|
|
},
|
|
required=["list_name", "items"],
|
|
)
|
|
async def add_to_list_tool(*, user_id, arguments, **_ctx):
|
|
list_name = str(arguments.get("list_name", "")).strip()
|
|
items = arguments.get("items") or []
|
|
if not list_name:
|
|
return {"success": False, "error": "list_name is required"}
|
|
if not isinstance(items, list) or not items:
|
|
return {"success": False, "error": "items must be a non-empty array"}
|
|
existing, _ = await list_notes(user_id=user_id, q=list_name, is_task=False, limit=10)
|
|
target = next((n for n in existing if n.note_type == "list" and n.title.lower() == list_name.lower()), None)
|
|
if target is None:
|
|
target = next((n for n in existing if n.note_type == "list"), None)
|
|
if target is None:
|
|
return {"success": False, "error": f"No list named '{list_name}' found. Use create_list to create it first."}
|
|
new_lines = "\n".join(f"- [ ] {item}" for item in items if str(item).strip())
|
|
existing_body = (target.body or "").rstrip()
|
|
new_body = f"{existing_body}\n{new_lines}".lstrip()
|
|
await update_note(user_id=user_id, note_id=target.id, body=new_body)
|
|
schedule_embedding(target.id, user_id, target.title, new_body)
|
|
return {"success": True, "type": "list_updated", "data": {"id": target.id, "name": target.title, "added": len(items)}}
|
|
|
|
|
|
@tool(
|
|
name="clear_checked_items",
|
|
description="Remove all checked/completed items from a list, keeping only unchecked items.",
|
|
parameters={
|
|
"list_name": {"type": "string", "description": "Name of the list to clear"},
|
|
},
|
|
required=["list_name"],
|
|
)
|
|
async def clear_checked_items_tool(*, user_id, arguments, **_ctx):
|
|
import re as _re
|
|
|
|
list_name = str(arguments.get("list_name", "")).strip()
|
|
if not list_name:
|
|
return {"success": False, "error": "list_name is required"}
|
|
existing, _ = await list_notes(user_id=user_id, q=list_name, is_task=False, limit=10)
|
|
target = next((n for n in existing if n.note_type == "list" and n.title.lower() == list_name.lower()), None)
|
|
if target is None:
|
|
target = next((n for n in existing if n.note_type == "list"), None)
|
|
if target is None:
|
|
return {"success": False, "error": f"No list named '{list_name}' found."}
|
|
body = target.body or ""
|
|
cleaned = _re.sub(r"^- \[[xX]\] .+\n?", "", body, flags=_re.MULTILINE).strip()
|
|
await update_note(user_id=user_id, note_id=target.id, body=cleaned)
|
|
schedule_embedding(target.id, user_id, target.title, cleaned)
|
|
return {"success": True, "type": "list_cleared", "data": {"id": target.id, "name": target.title}}
|