bvandeusenandClaude Opus 4.6
ce2d76447c
Split 2566-line tools.py into a tools/ package with @tool decorator registration. Each tool's schema, metadata, and implementation live together. Briefing eligibility is now a briefing=True flag instead of a separate frozenset allowlist. Conditional inclusion (CalDAV, SearXNG) uses requires= metadata. Public API (get_tools_for_user, execute_tool) unchanged. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
285 lines
14 KiB
Python
285 lines
14 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
|
|
|
|
|
|
@tool(
|
|
name="create_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."
|
|
),
|
|
parameters={
|
|
"name": {"type": "string", "description": "Full name of the person"},
|
|
"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 create_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"):
|
|
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}")
|
|
note = await create_note(
|
|
user_id=user_id, title=name, body="\n".join(body_lines),
|
|
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="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.",
|
|
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 create_place_tool(*, user_id, arguments, **_ctx):
|
|
name = str(arguments.get("name", "")).strip()
|
|
if not name:
|
|
return {"success": False, "error": "name is required"}
|
|
meta = {}
|
|
for field in ("address", "phone", "hours", "url"):
|
|
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}")
|
|
note = await create_note(
|
|
user_id=user_id, title=name, body="\n".join(body_lines),
|
|
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.",
|
|
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}}
|