bvandeusen
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>
259 lines
12 KiB
Python
259 lines
12 KiB
Python
"""Project and milestone tools."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fabledassistant.services.tools._helpers import fuzzy_title_match, resolve_project
|
|
from fabledassistant.services.tools._registry import tool
|
|
|
|
|
|
@tool(
|
|
name="create_project",
|
|
description="Create a new project. Use list_projects first to check for duplicates. Only call after user has explicitly confirmed.",
|
|
parameters={
|
|
"title": {"type": "string", "description": "Project title"},
|
|
"description": {"type": "string", "description": "Brief description"},
|
|
"goal": {"type": "string", "description": "The goal or desired outcome"},
|
|
"color": {"type": "string", "description": "Optional hex color for the project (e.g. '#6366f1')"},
|
|
"confirmed": {"type": "boolean", "description": "Must be true — only set after the user has explicitly confirmed they want this project created."},
|
|
},
|
|
required=["title", "confirmed"],
|
|
)
|
|
async def create_project_tool(*, user_id, arguments, **_ctx):
|
|
from fabledassistant.services.projects import create_project as _create_project, get_project_by_title as _gpbt, list_projects as _lp
|
|
|
|
proj_title = arguments.get("title", "New Project")
|
|
existing_proj = await _gpbt(user_id, proj_title)
|
|
if existing_proj is not None:
|
|
return {"success": False, "error": f"A project titled '{existing_proj.title}' already exists (id: {existing_proj.id}). Use that project instead of creating a duplicate."}
|
|
all_projects = await _lp(user_id)
|
|
near_proj, ratio = fuzzy_title_match(proj_title, all_projects, threshold=0.55)
|
|
if near_proj is not None:
|
|
return {"success": False, "requires_confirmation": True, "error": f"A project with a similar name '{near_proj.title}' already exists (id: {near_proj.id}, similarity: {ratio:.0%}). This is likely the project you meant — use that project's name instead. Only call create_project again if you are certain this should be a completely new, separate project with a different confirmed=true."}
|
|
if not arguments.get("confirmed"):
|
|
return {"success": False, "requires_confirmation": True, "error": f"Project creation requires explicit user confirmation. Ask the user: 'Shall I create a new project titled \"{proj_title}\"?' Then retry with confirmed=true if they agree."}
|
|
project = await _create_project(
|
|
user_id,
|
|
title=proj_title,
|
|
description=arguments.get("description", ""),
|
|
goal=arguments.get("goal", ""),
|
|
color=arguments.get("color") or None,
|
|
)
|
|
return {"success": True, "type": "project", "data": project.to_dict()}
|
|
|
|
|
|
@tool(
|
|
name="list_projects",
|
|
description="List the user's projects. Use when asked about projects, initiatives, or campaigns.",
|
|
parameters={
|
|
"status": {"type": "string", "enum": ["active", "completed", "archived"], "description": "Filter by status (omit for all)"},
|
|
},
|
|
read_only=True,
|
|
briefing=True,
|
|
)
|
|
async def list_projects_tool(*, user_id, arguments, **_ctx):
|
|
from fabledassistant.services.projects import list_projects as _list_projects
|
|
|
|
projects = await _list_projects(user_id, status=arguments.get("status"))
|
|
return {
|
|
"success": True,
|
|
"type": "projects_list",
|
|
"data": {"count": len(projects), "projects": [p.to_dict() for p in projects]},
|
|
}
|
|
|
|
|
|
@tool(
|
|
name="get_project",
|
|
description="Get details and summary of a specific project.",
|
|
parameters={
|
|
"query": {"type": "string", "description": "Project name or keyword to find it"},
|
|
},
|
|
required=["query"],
|
|
read_only=True,
|
|
briefing=True,
|
|
)
|
|
async def get_project_tool(*, user_id, arguments, **_ctx):
|
|
from fabledassistant.services.projects import get_project_by_title as _gpbt, get_project_summary as _gps, list_projects as _lp
|
|
|
|
query = arguments.get("query", "")
|
|
project = await _gpbt(user_id, query)
|
|
if project is None:
|
|
all_projects = await _lp(user_id)
|
|
matches = [p for p in all_projects if query.lower() in p.title.lower()]
|
|
if matches:
|
|
project = matches[0]
|
|
if project is None:
|
|
return {"success": False, "error": f"No project found matching '{query}'"}
|
|
summary = await _gps(user_id, project.id)
|
|
data = project.to_dict()
|
|
data["summary"] = summary
|
|
return {"success": True, "type": "project_detail", "data": data}
|
|
|
|
|
|
@tool(
|
|
name="update_project",
|
|
description="Update a project's title, status, description, goal, or color.",
|
|
parameters={
|
|
"query": {"type": "string", "description": "Project name to find"},
|
|
"title": {"type": "string", "description": "New project title"},
|
|
"status": {"type": "string", "enum": ["active", "completed", "archived"]},
|
|
"description": {"type": "string"},
|
|
"goal": {"type": "string"},
|
|
"color": {"type": "string", "description": "Hex color (e.g. '#6366f1'), or empty string to clear"},
|
|
},
|
|
required=["query"],
|
|
)
|
|
async def update_project_tool(*, user_id, arguments, **_ctx):
|
|
from fabledassistant.services.projects import get_project_by_title as _gpbt, update_project as _up, list_projects as _lp
|
|
|
|
query = arguments.get("query", "")
|
|
project = await _gpbt(user_id, query)
|
|
if project is None:
|
|
all_projects = await _lp(user_id)
|
|
matches = [p for p in all_projects if query.lower() in p.title.lower()]
|
|
if matches:
|
|
project = matches[0]
|
|
if project is None:
|
|
return {"success": False, "error": f"No project found matching '{query}'"}
|
|
fields = {}
|
|
for k in ("title", "status", "description", "goal"):
|
|
if k in arguments:
|
|
fields[k] = arguments[k]
|
|
if "color" in arguments:
|
|
fields["color"] = arguments["color"] or None
|
|
updated = await _up(user_id, project.id, **fields)
|
|
return {"success": True, "type": "project_updated", "data": updated.to_dict()}
|
|
|
|
|
|
@tool(
|
|
name="search_projects",
|
|
description="Search for projects by name, description, or theme. Use this to find which project a user is referring to and get its ID for set_rag_scope.",
|
|
parameters={
|
|
"query": {"type": "string", "description": "Search query — project name, topic, or theme"},
|
|
},
|
|
required=["query"],
|
|
read_only=True,
|
|
briefing=True,
|
|
)
|
|
async def search_projects_tool(*, user_id, arguments, **_ctx):
|
|
from difflib import SequenceMatcher
|
|
|
|
from fabledassistant.services.projects import list_projects
|
|
|
|
query = str(arguments.get("query", "")).lower()
|
|
projects = await list_projects(user_id)
|
|
scored: list[tuple[float, object]] = []
|
|
for p in projects:
|
|
combined = f"{p.title} {p.description or ''} {p.auto_summary or ''}".lower()
|
|
base_score = SequenceMatcher(None, query, combined).ratio()
|
|
query_words = set(query.split())
|
|
overlap = sum(1 for w in query_words if w in combined)
|
|
score = base_score + overlap * 0.05
|
|
scored.append((score, p))
|
|
scored.sort(key=lambda x: x[0], reverse=True)
|
|
results = []
|
|
for score, p in scored[:5]:
|
|
results.append({
|
|
"id": p.id,
|
|
"title": p.title,
|
|
"summary_snippet": (p.auto_summary or p.description or "")[:200],
|
|
"score": round(score, 3),
|
|
})
|
|
return {"type": "projects_list", "data": {"projects": results}}
|
|
|
|
|
|
@tool(
|
|
name="create_milestone",
|
|
description="Create a milestone inside a project. Confirm name and project with user before calling.",
|
|
parameters={
|
|
"project": {"type": "string", "description": "Project title"},
|
|
"title": {"type": "string", "description": "Milestone title"},
|
|
"description": {"type": "string", "description": "Optional description"},
|
|
"confirmed": {"type": "boolean", "description": "Must be true — only set after the user has explicitly confirmed they want this milestone created."},
|
|
},
|
|
required=["project", "title", "confirmed"],
|
|
)
|
|
async def create_milestone_tool(*, user_id, arguments, **_ctx):
|
|
from fabledassistant.services.milestones import create_milestone as _cm, get_milestone_by_title as _gmbt2, list_milestones as _lms
|
|
|
|
project_name = arguments.get("project", "")
|
|
ms_title = arguments.get("title", "Untitled Milestone")
|
|
if not project_name:
|
|
return {"success": False, "error": "project is required"}
|
|
if not arguments.get("confirmed"):
|
|
return {"success": False, "requires_confirmation": True, "error": f"Milestone creation requires explicit user confirmation. Ask the user: 'Shall I create a milestone titled \"{ms_title}\" in project \"{project_name}\"?' Then retry with confirmed=true if they agree."}
|
|
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."}
|
|
existing_ms = await _gmbt2(user_id, proj.id, ms_title)
|
|
if existing_ms is not None:
|
|
return {"success": False, "error": f"A milestone titled '{existing_ms.title}' already exists in '{proj.title}' (id: {existing_ms.id}). Use update_milestone to modify it instead."}
|
|
all_ms = await _lms(user_id, proj.id)
|
|
near_ms, ratio = fuzzy_title_match(ms_title, all_ms)
|
|
if near_ms is not None:
|
|
return {"success": False, "error": f"A milestone with a very similar title '{near_ms.title}' already exists in '{proj.title}' (id: {near_ms.id}, similarity: {ratio:.0%}). Use update_milestone to modify it instead."}
|
|
ms = await _cm(user_id, proj.id, title=ms_title, description=arguments.get("description"))
|
|
return {"success": True, "type": "milestone", "data": ms.to_dict()}
|
|
|
|
|
|
@tool(
|
|
name="update_milestone",
|
|
description="Update the title, description, or status of an existing milestone.",
|
|
parameters={
|
|
"project": {"type": "string", "description": "Project title the milestone belongs to"},
|
|
"milestone": {"type": "string", "description": "Current milestone title to look up"},
|
|
"title": {"type": "string", "description": "New title (omit to keep current)"},
|
|
"description": {"type": "string", "description": "New description (omit to keep current)"},
|
|
"status": {"type": "string", "enum": ["active", "done"], "description": "New status: 'active' (in progress) or 'done' (completed)"},
|
|
},
|
|
required=["project", "milestone"],
|
|
)
|
|
async def update_milestone_tool(*, user_id, arguments, **_ctx):
|
|
from fabledassistant.services.milestones import get_milestone_by_title as _gmbt, update_milestone as _um
|
|
|
|
project_name = arguments.get("project", "")
|
|
milestone_name = arguments.get("milestone", "")
|
|
if not project_name or not milestone_name:
|
|
return {"success": False, "error": "Both project and milestone are required"}
|
|
proj = await resolve_project(user_id, project_name)
|
|
if proj is None:
|
|
return {"success": False, "error": f"Project '{project_name}' not found"}
|
|
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."}
|
|
fields: dict[str, object] = {}
|
|
if "title" in arguments:
|
|
fields["title"] = arguments["title"]
|
|
if "description" in arguments:
|
|
fields["description"] = arguments["description"]
|
|
if "status" in arguments:
|
|
fields["status"] = arguments["status"]
|
|
if not fields:
|
|
return {"success": False, "error": "No fields to update — provide at least one of: title, description, status"}
|
|
updated = await _um(user_id, ms.id, **fields)
|
|
return {"success": True, "type": "milestone", "data": updated.to_dict()}
|
|
|
|
|
|
@tool(
|
|
name="list_milestones",
|
|
description="List milestones for a project with their progress percentages.",
|
|
parameters={
|
|
"project": {"type": "string", "description": "Project name to look up"},
|
|
},
|
|
required=["project"],
|
|
read_only=True,
|
|
briefing=True,
|
|
)
|
|
async def list_milestones_tool(*, user_id, arguments, **_ctx):
|
|
from fabledassistant.services.milestones import get_project_milestone_summary
|
|
|
|
project_name = arguments.get("project", "")
|
|
proj = await resolve_project(user_id, project_name)
|
|
if proj is None:
|
|
return {"success": False, "error": f"No project found matching '{project_name}'"}
|
|
summary = await get_project_milestone_summary(user_id, proj.id)
|
|
return {
|
|
"success": True,
|
|
"type": "milestones_list",
|
|
"data": {"project": proj.title, "count": len(summary), "milestones": summary},
|
|
}
|