"""Tools that run in the brain: search, weather, reminders, memory. Blueprint 8.1 and 8.2.""" from datetime import datetime, timedelta, tzinfo from typing import Any import httpx from ..memory import service as memory from ..memory.scope import forced_scope from ..persona.types import Mode from . import weather from .registry import Tool, ToolContext, ToolOutcome, failed, ok, schema, string DEFAULT_CITY = "Kolkata" #: A reminder "in the past" by less than this is treated as "now" — the model computed the time #: a few seconds before the call landed. PAST_TOLERANCE = timedelta(minutes=2) async def _web_search(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome: """Blueprint 8.2: a separate grounded call; the sources become chips under the reply.""" answer, sources = await ctx.llm.grounded_search(str(args["query"])) if not answer: return ToolOutcome(failed("the search found nothing useful")) return ToolOutcome(ok("searched", answer=answer), sources=sources) async def _get_weather(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome: place = ( str(args.get("location") or "").strip() or str(ctx.hud.get("city") or "").strip() or str((ctx.settings.get("hud") or {}).get("city") or "").strip() or DEFAULT_CITY ) client = ctx.http or httpx.AsyncClient(timeout=10.0) try: report = await weather.current(client, place) except weather.WeatherError as exc: return ToolOutcome(failed(str(exc))) finally: if ctx.http is None: await client.aclose() return ToolOutcome(ok("weather", **report)) def parse_when(raw: str, ctx: ToolContext) -> datetime | None: """An ISO time from the model. A naive one is read in HIS timezone, never UTC (CLAUDE.md).""" try: when = datetime.fromisoformat(raw.strip().replace("Z", "+00:00")) except ValueError: return None if when.tzinfo is None: when = when.replace(tzinfo=ctx.tz) return when def local_label(when: datetime, tz: tzinfo) -> str: local = when.astimezone(tz) hour = local.hour % 12 or 12 suffix = "AM" if local.hour < 12 else "PM" return f"{local.strftime('%a %d %b')}, {hour}:{local.minute:02d} {suffix}" async def _set_reminder(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome: when = parse_when(str(args["when_iso"]), ctx) if when is None: return ToolOutcome(failed("when_iso must be an ISO 8601 date and time")) if when < ctx.now - PAST_TOLERANCE: return ToolOutcome(failed("that time has already passed")) when = max(when, ctx.now) text = str(args["text"]).strip() row = await ctx.store.add_reminder(ctx.user_id, ctx.mode, text, when) return ToolOutcome(ok("reminder set", id=str(row["id"]), due=local_label(when, ctx.tz))) async def _list_reminders(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome: rows = await ctx.store.list_reminders(ctx.user_id) # Normal mode never lists a reminder that was created in GF mode — its text may be private. visible = [row for row in rows if ctx.mode is Mode.GF or row["mode"] == Mode.NORMAL.value] return ToolOutcome( ok( f"{len(visible)} upcoming", reminders=[ { "id": str(row["id"]), "text": row["text"], "due": local_label(row["due_at"], ctx.tz), } for row in visible ], ) ) async def _cancel_reminder(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome: cancelled = await ctx.store.cancel_reminder(ctx.user_id, str(args["id"])) return ToolOutcome(ok("cancelled") if cancelled else failed("no reminder with that id")) async def _remember(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome: scope = forced_scope(args.get("scope"), ctx.mode) row, duplicate = await memory.remember( ctx.store, ctx.llm, ctx.user_id, str(args["fact"]), scope, is_manual=True ) if duplicate is not None: return ToolOutcome(ok("already remembered", existing=duplicate.content)) return ToolOutcome(ok("remembered", scope=scope.value)) async def _forget(args: dict[str, Any], ctx: ToolContext) -> ToolOutcome: """Finds the memory but never deletes it: he confirms on the device first (blueprint 8.1).""" hit = await memory.find_to_forget(ctx.store, ctx.llm, ctx.user_id, str(args["query"]), ctx.mode) if hit is None: return ToolOutcome(failed("nothing like that is in memory")) action = {"type": "confirm_forget", "memoryId": hit.id, "content": hit.content} return ToolOutcome(ok("asked him to confirm", found=hit.content), client_actions=[action]) SERVER_TOOLS: list[Tool] = [ Tool( "web_search", "Search the web and get a current answer with sources — news, scores, prices, anything " "that changes. Use it instead of guessing about recent events.", schema({"query": string("What to find out")}, ["query"]), _web_search, ), Tool( "get_weather", "Current weather and today's range. Omit location to use where he is.", schema({"location": string("A city or place name")}), _get_weather, ), Tool( "set_reminder", "Set a reminder that arrives as a notification at a specific time. Work out the exact " "time from CURRENT TIME in your instructions and his timezone.", schema( { "text": string("What to remind him about, as he should read it"), "when_iso": string("ISO 8601 local date and time, e.g. 2026-09-19T14:45:00"), }, ["text", "when_iso"], ), _set_reminder, ), Tool( "list_reminders", "List his upcoming reminders.", schema({}), _list_reminders, ), Tool( "cancel_reminder", "Cancel a reminder by its id (get ids from list_reminders).", schema({"id": string("The reminder id")}, ["id"]), _cancel_reminder, ), Tool( "remember", "Save a fact about him for the long term, when he explicitly asks you to remember " "something. One short third-person sentence in English.", schema( { "fact": string("The fact, e.g. 'His dentist appointment is on Friday.'"), "scope": string( "private for anything romantic or intimate; shared for everything else", ["shared", "private"], ), }, ["fact"], ), _remember, ), Tool( "forget", "Find a memory he wants deleted. He confirms on his device before anything is deleted.", schema({"query": string("What he wants forgotten")}, ["query"]), _forget, ), ]