bvandeusen
6961144c3a
Two cross-type bootstrap tools: - fable_list_tags: tag vocabulary with usage counts, top-N by count. Aggregation in Python (not SQL UNNEST) — trivial perf cost at personal scale, much easier to test. - fable_get_recent: most-recently-touched items across notes, tasks, projects, events. Useful for Claude to ask 'what was I working on recently' at the start of a conversation. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
82 lines
3.0 KiB
Python
82 lines
3.0 KiB
Python
"""fable_get_recent — cross-type recent-activity tool.
|
|
|
|
Returns the most-recently-touched notes, tasks, projects, and events for the
|
|
user, ordered by updated_at descending. Useful for Claude to bootstrap context
|
|
at the start of a conversation ("what was I working on?").
|
|
|
|
Aggregation is Python-side after three small per-table queries — simpler than
|
|
a UNION ALL with type-discriminating columns, and fine for personal-scale data.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
|
|
from fabledassistant.mcp._context import current_user_id
|
|
from fabledassistant.models import async_session
|
|
from fabledassistant.models.event import Event
|
|
from fabledassistant.models.note import Note
|
|
from fabledassistant.models.project import Project
|
|
|
|
|
|
async def fable_get_recent(days: int = 7, limit: int = 25) -> dict:
|
|
"""Return recently-touched items across notes, tasks, projects, events.
|
|
|
|
Args:
|
|
days: Look-back window in days (1-90).
|
|
limit: Maximum number of items returned (1-100).
|
|
|
|
Returns:
|
|
{"items": [{"id", "type", "title", "updated_at"}], "total": int}
|
|
Sorted by updated_at descending.
|
|
"""
|
|
uid = current_user_id()
|
|
days = max(1, min(days, 90))
|
|
limit = max(1, min(limit, 100))
|
|
since = datetime.now(timezone.utc) - timedelta(days=days)
|
|
items: list[dict] = []
|
|
async with async_session() as session:
|
|
notes = (await session.execute(
|
|
select(Note).where(Note.user_id == uid, Note.updated_at >= since)
|
|
.order_by(Note.updated_at.desc()).limit(limit)
|
|
)).scalars().all()
|
|
for n in notes:
|
|
items.append({
|
|
"id": n.id,
|
|
"type": "task" if n.is_task else "note",
|
|
"title": n.title,
|
|
"updated_at": n.updated_at.isoformat(),
|
|
})
|
|
projects = (await session.execute(
|
|
select(Project).where(Project.user_id == uid,
|
|
Project.updated_at >= since)
|
|
.order_by(Project.updated_at.desc()).limit(limit)
|
|
)).scalars().all()
|
|
for p in projects:
|
|
items.append({
|
|
"id": p.id,
|
|
"type": "project",
|
|
"title": p.title,
|
|
"updated_at": p.updated_at.isoformat(),
|
|
})
|
|
events = (await session.execute(
|
|
select(Event).where(Event.user_id == uid,
|
|
Event.updated_at >= since)
|
|
.order_by(Event.updated_at.desc()).limit(limit)
|
|
)).scalars().all()
|
|
for e in events:
|
|
items.append({
|
|
"id": e.id,
|
|
"type": "event",
|
|
"title": e.title,
|
|
"updated_at": e.updated_at.isoformat(),
|
|
})
|
|
items.sort(key=lambda r: r["updated_at"], reverse=True)
|
|
items = items[:limit]
|
|
return {"items": items, "total": len(items)}
|
|
|
|
|
|
def register(mcp) -> None:
|
|
mcp.tool(name="fable_get_recent")(fable_get_recent)
|