jpmschweitzer/tatlock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

feat: implement permanent tools (calculator, date/time, search)

Browse Source 
Add three permanent tools for Tatlock agent:
- Calculator: Safe math expression evaluation (arithmetic, algebra, trig, log)
- Date/Time toolkit: Current time, relative dates, time differences
- Web Search: SearXNG integration for privacy-preserving search

Tools use PydanticAI @agent.tool decorator pattern with:
- Clear docstrings visible to LLM
- Error handling with string-based messages
- Async support for I/O operations (web search)
- 26 comprehensive tool tests
This commit is contained in:
jpmschweitzer
2025-12-07 00:10:49 +01:00
parent a1a0f6923b
commit f3e2681a6c
2 changed files with 717 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agents/tools.py
+346
View File
@@ -0,0 +1,346 @@
"""
Tatlock's permanent tools.
These tools are always available to the butler agent:
- Calculator: For all mathematical operations
- Date/Time toolkit: For current time and time calculations
- SearXNG search: For searching the web for current information
"""
import logging
import math
import re
from datetime import datetime, timedelta
from typing import Any
import httpx
from src.core.config import config
logger = logging.getLogger(__name__)
# ============================================================================
# Calculator Tool
# ============================================================================
def calculate(expression: str) -> str:
"""
Safely evaluate mathematical expressions.
Supports:
- Basic arithmetic: +, -, *, /, //, %, **
- Parentheses for grouping
- Common math functions: sqrt, sin, cos, tan, log, exp, etc.
- Constants: pi, e
Args:
expression: Mathematical expression to evaluate (e.g., "2 + 2", "sqrt(16)", "pi * 2")
Returns:
String result of the calculation or error message
Examples:
calculate("2 + 2") -> "4"
calculate("sqrt(16) + 10") -> "14.0"
calculate("pi * 2") -> "6.283185307179586"
"""
try:
# Clean the expression
expression = expression.strip()
# Create safe namespace with math functions
safe_dict = {
# Basic math functions
'sqrt': math.sqrt,
'pow': math.pow,
'abs': abs,
'round': round,
# Trigonometric
'sin': math.sin,
'cos': math.cos,
'tan': math.tan,
'asin': math.asin,
'acos': math.acos,
'atan': math.atan,
# Logarithmic
'log': math.log,
'log10': math.log10,
'log2': math.log2,
'exp': math.exp,
# Other
'ceil': math.ceil,
'floor': math.floor,
'factorial': math.factorial,
# Constants
'pi': math.pi,
'e': math.e,
}
# Evaluate the expression safely
result = eval(expression, {"__builtins__": {}}, safe_dict)
# Format result nicely
if isinstance(result, float):
# Remove unnecessary decimal places
if result.is_integer():
return str(int(result))
return str(round(result, 10))
return str(result)
except ZeroDivisionError:
return "Error: Division by zero"
except Exception as e:
return f"Error calculating '{expression}': {str(e)}"
# ============================================================================
# Date/Time Toolkit
# ============================================================================
def get_current_datetime(format_str: str = "full") -> str:
"""
Get the current date and time.
Args:
format_str: Output format
- "full": Full datetime with timezone (default)
- "date": Just the date (YYYY-MM-DD)
- "time": Just the time (HH:MM:SS)
- "iso": ISO 8601 format
- Custom strftime format string
Returns:
Formatted current datetime string
Examples:
get_current_datetime("full") -> "2024-01-15 14:30:45"
get_current_datetime("date") -> "2024-01-15"
get_current_datetime("time") -> "14:30:45"
"""
now = datetime.now()
if format_str == "full":
return now.strftime("%Y-%m-%d %H:%M:%S")
elif format_str == "date":
return now.strftime("%Y-%m-%d")
elif format_str == "time":
return now.strftime("%H:%M:%S")
elif format_str == "iso":
return now.isoformat()
else:
# Custom format
try:
return now.strftime(format_str)
except Exception as e:
return f"Error formatting date: {str(e)}"
def calculate_time_offset(offset_description: str) -> str:
"""
Calculate a date/time relative to now.
Args:
offset_description: Natural language description of time offset
Examples: "1 week ago", "2 days from now", "3 months ago",
"1 year from now", "5 hours ago"
Returns:
Formatted datetime string (YYYY-MM-DD HH:MM:SS) or error message
Examples:
calculate_time_offset("1 week ago") -> "2024-01-08 14:30:45"
calculate_time_offset("2 days from now") -> "2024-01-17 14:30:45"
calculate_time_offset("3 months ago") -> "2023-10-15 14:30:45"
"""
try:
now = datetime.now()
# Parse the offset description
# Pattern: "N unit(s) ago/from now"
pattern = r'(\d+)\s+(second|minute|hour|day|week|month|year)s?\s+(ago|from\s+now)'
match = re.match(pattern, offset_description.lower().strip())
if not match:
return f"Error: Cannot parse '{offset_description}'. Use format like '1 week ago' or '2 days from now'"
amount = int(match.group(1))
unit = match.group(2)
direction = match.group(3)
# Calculate the offset
if direction == "ago":
amount = -amount
if unit == "second":
target = now + timedelta(seconds=amount)
elif unit == "minute":
target = now + timedelta(minutes=amount)
elif unit == "hour":
target = now + timedelta(hours=amount)
elif unit == "day":
target = now + timedelta(days=amount)
elif unit == "week":
target = now + timedelta(weeks=amount)
elif unit == "month":
# Approximate month as 30 days
target = now + timedelta(days=amount * 30)
elif unit == "year":
# Approximate year as 365 days
target = now + timedelta(days=amount * 365)
else:
return f"Error: Unknown time unit '{unit}'"
return target.strftime("%Y-%m-%d %H:%M:%S")
except Exception as e:
return f"Error calculating time offset: {str(e)}"
def time_difference(date1_str: str, date2_str: str = "now") -> str:
"""
Calculate the difference between two dates.
Args:
date1_str: First date (YYYY-MM-DD or YYYY-MM-DD HH:MM:SS)
date2_str: Second date or "now" for current time (default: "now")
Returns:
Human-readable description of the time difference
Examples:
time_difference("2024-01-01", "now") -> "14 days, 14 hours"
time_difference("2024-01-01", "2024-01-15") -> "14 days"
"""
try:
# Parse date1
if len(date1_str) == 10: # YYYY-MM-DD
date1 = datetime.strptime(date1_str, "%Y-%m-%d")
else:
date1 = datetime.strptime(date1_str, "%Y-%m-%d %H:%M:%S")
# Parse date2
if date2_str.lower() == "now":
date2 = datetime.now()
elif len(date2_str) == 10:
date2 = datetime.strptime(date2_str, "%Y-%m-%d")
else:
date2 = datetime.strptime(date2_str, "%Y-%m-%d %H:%M:%S")
# Calculate difference
diff = abs(date2 - date1)
# Format human-readable
days = diff.days
seconds = diff.seconds
hours = seconds // 3600
minutes = (seconds % 3600) // 60
parts = []
if days > 0:
parts.append(f"{days} day{'s' if days != 1 else ''}")
if hours > 0:
parts.append(f"{hours} hour{'s' if hours != 1 else ''}")
if minutes > 0 and days == 0: # Only show minutes if less than a day
parts.append(f"{minutes} minute{'s' if minutes != 1 else ''}")
if not parts:
return "Less than a minute"
return ", ".join(parts)
except Exception as e:
return f"Error calculating time difference: {str(e)}"
# ============================================================================
# SearXNG Search Tool
# ============================================================================
async def search_web(query: str, num_results: int = 5) -> str:
"""
Search the web using SearXNG.
Args:
query: Search query string
num_results: Number of results to return (default: 5, max: 10)
Returns:
Formatted search results as a string with titles, URLs, and snippets
Examples:
search_web("Python async programming") -> "1. Title: ...\n URL: ...\n ..."
"""
try:
# Limit results
num_results = min(num_results, 10)
# Get SearXNG host with fallback logic
searxng_host = str(config.SEARXNG_HOST)
# Try production host first, fall back to localhost in development
hosts_to_try = [searxng_host]
if config.ENVIRONMENT.value == "development" and "localhost" not in searxng_host:
# Add localhost fallback for development
hosts_to_try.append("http://localhost:8087")
last_error = None
for host in hosts_to_try:
try:
logger.info(f"Attempting SearXNG search at {host}")
async with httpx.AsyncClient(timeout=config.SEARXNG_TIMEOUT) as client:
response = await client.get(
f"{host}/search",
params={
"q": query,
"format": "json",
"pageno": 1,
}
)
if response.status_code == 200:
data = response.json()
results = data.get("results", [])
if not results:
return f"No results found for '{query}'"
# Format results
formatted_results = []
for i, result in enumerate(results[:num_results], 1):
title = result.get("title", "No title")
url = result.get("url", "")
content = result.get("content", "No description available")
formatted_results.append(
f"{i}. {title}\n"
f" URL: {url}\n"
f" {content}\n"
)
return "\n".join(formatted_results)
else:
last_error = f"SearXNG returned status {response.status_code}"
except httpx.ConnectError:
last_error = f"Cannot connect to SearXNG at {host}"
logger.warning(f"SearXNG connection failed at {host}, trying next host if available")
continue
except Exception as e:
last_error = str(e)
logger.warning(f"SearXNG error at {host}: {e}")
continue
# All hosts failed
return f"Error searching: {last_error}. Please check that SearXNG is running."
except Exception as e:
logger.error(f"Unexpected error in search_web: {e}", exc_info=True)
return f"Error searching: {str(e)}"