"""Utility tools for common operations."""
import random
import re
from datetime import datetime
from zoneinfo import ZoneInfo
from pydantic import BaseModel, Field
from ..bot import BotContext
from .base import ToolResult
from .registry import folder_bot
[docs]
class GetTimeRequest(BaseModel, frozen=True):
"""Request for getting current time."""
timezone: str = Field(
default="UTC",
description="Timezone (e.g., 'UTC', 'America/New_York', 'Europe/London')",
)
format: str = Field(
default="%A, %Y-%m-%d %H:%M:%S",
description="strftime format string",
)
[docs]
@folder_bot.tool(
name="get_time",
request_type=GetTimeRequest,
response_type=ToolResult,
)
async def get_time(
request: GetTimeRequest, _context: BotContext | None = None
) -> ToolResult:
"""Get the current date and time. Useful for knowing what day/time it is."""
try:
tz = ZoneInfo(request.timezone)
now = datetime.now(tz)
formatted = now.strftime(request.format)
return ToolResult(content=f"{formatted} ({request.timezone})")
except Exception as e:
return ToolResult(content=f"Error: {e}", is_error=True)
[docs]
class CompareNumbersRequest(BaseModel, frozen=True):
"""Request for comparing two numbers."""
a: float = Field(description="First number")
b: float = Field(description="Second number")
[docs]
@folder_bot.tool(
name="compare_numbers",
request_type=CompareNumbersRequest,
response_type=ToolResult,
)
async def compare_numbers(
request: CompareNumbersRequest, _context: BotContext | None = None
) -> ToolResult:
"""Compare two numbers. Returns which is greater, lesser, or if equal."""
if request.a > request.b:
result = f"{request.a} > {request.b} (a is greater)"
elif request.a < request.b:
result = f"{request.a} < {request.b} (b is greater)"
else:
result = f"{request.a} = {request.b} (equal)"
return ToolResult(content=result)
[docs]
class ShuffleListRequest(BaseModel, frozen=True):
"""Request for shuffling a list."""
items: list[str] = Field(description="List of items to shuffle")
[docs]
@folder_bot.tool(
name="shuffle_list",
request_type=ShuffleListRequest,
response_type=ToolResult,
)
async def shuffle_list(
request: ShuffleListRequest, _context: BotContext | None = None
) -> ToolResult:
"""Randomly shuffle a list of items."""
items = list(request.items)
random.shuffle(items)
return ToolResult(content="\n".join(items))
[docs]
class SortListRequest(BaseModel, frozen=True):
"""Request for sorting a list."""
items: list[str | float] = Field(description="List of items to sort")
reverse: bool = Field(default=False, description="Sort in descending order")
numeric: bool = Field(
default=False, description="Treat items as numbers for sorting"
)
[docs]
@folder_bot.tool(
name="sort_list",
request_type=SortListRequest,
response_type=ToolResult,
)
async def sort_list(
request: SortListRequest, _context: BotContext | None = None
) -> ToolResult:
"""Sort a list of items alphabetically or numerically."""
if request.numeric:
try:
sorted_nums = sorted(
[float(x) for x in request.items], reverse=request.reverse
)
sorted_items = [str(int(x)) if x == int(x) else str(x) for x in sorted_nums]
except ValueError as e:
return ToolResult(
content=f"Error converting to numbers: {e}", is_error=True
)
else:
sorted_items = sorted([str(x) for x in request.items], reverse=request.reverse)
return ToolResult(content="\n".join(sorted_items))
[docs]
class RandomChoiceRequest(BaseModel, frozen=True):
"""Request for picking random items."""
items: list[str] = Field(description="List of items to choose from")
count: int = Field(
default=1, description="Number of items to pick (without replacement)"
)
[docs]
@folder_bot.tool(
name="random_choice",
request_type=RandomChoiceRequest,
response_type=ToolResult,
)
async def random_choice(
request: RandomChoiceRequest, _context: BotContext | None = None
) -> ToolResult:
"""Pick random item(s) from a list."""
if request.count > len(request.items):
return ToolResult(
content=f"Cannot pick {request.count} items from list of {len(request.items)}",
is_error=True,
)
if request.count == 1:
return ToolResult(content=random.choice(request.items))
choices = random.sample(list(request.items), request.count)
return ToolResult(content="\n".join(choices))
[docs]
class RandomNumberRequest(BaseModel, frozen=True):
"""Request for generating a random number."""
min: float = Field(default=0, description="Minimum value (inclusive)")
max: float = Field(default=100, description="Maximum value (inclusive)")
integer: bool = Field(default=True, description="Return an integer")
[docs]
@folder_bot.tool(
name="random_number",
request_type=RandomNumberRequest,
response_type=ToolResult,
)
async def random_number(
request: RandomNumberRequest, _context: BotContext | None = None
) -> ToolResult:
"""Generate a random number within a range."""
if request.integer:
result: int | float = random.randint(int(request.min), int(request.max))
else:
result = random.uniform(request.min, request.max)
return ToolResult(content=str(result))
[docs]
class SendMessageRequest(BaseModel, frozen=True):
"""Request for sending a message to the user."""
message: str = Field(description="The message text to send to the user")
_TEMPLATE_PATTERN = re.compile(r"\{\{.*?\}\}")
[docs]
@folder_bot.tool(
name="send_message",
request_type=SendMessageRequest,
response_type=ToolResult,
)
async def send_message(
request: SendMessageRequest, _context: BotContext | None = None
) -> ToolResult:
"""Send a message directly to the user via Telegram.
Use this in scheduled tasks to send notifications, greetings,
reminders, or any other message to the user.
The message is sent exactly as provided — no template interpolation.
"""
if _TEMPLATE_PATTERN.search(request.message):
return ToolResult(
content=(
"Message contains template placeholders like {{...}} which will "
"be sent as literal text. There is no template engine. Compose "
"the actual message content before calling send_message."
),
is_error=True,
)
return ToolResult(content=request.message)
[docs]
class ListTopicsRequest(BaseModel, frozen=True):
"""Request for listing conversation topics."""
[docs]
@folder_bot.tool(
name="list_topics",
request_type=ListTopicsRequest,
response_type=ToolResult,
)
async def list_topics(
request: ListTopicsRequest, _context: BotContext | None = None
) -> ToolResult:
"""List active conversation topics for the current user.
Shows all topics with message counts and last activity.
Use this when the user asks about their active conversations or threads.
"""
if _context is None or "session" not in _context.services:
return ToolResult(content="Session manager not available.", is_error=True)
session_manager = _context.services["session"]
topics = session_manager.get_topics(_context.user_id)
if not topics:
return ToolResult(content="No conversation topics yet.")
lines = [f"Active topics ({len(topics)}):"]
for t in topics:
lines.append(
f" - {t['topic']}: {t['message_count']} messages"
f" (last: {t['last_activity'][:10]})"
)
return ToolResult(content="\n".join(lines))