diff --git a/src/agents/tools.py b/src/agents/tools.py new file mode 100644 index 0000000..9f8a28a --- /dev/null +++ b/src/agents/tools.py @@ -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)}" diff --git a/tests/agents/test_tools.py b/tests/agents/test_tools.py new file mode 100644 index 0000000..ae38500 --- /dev/null +++ b/tests/agents/test_tools.py @@ -0,0 +1,371 @@ +""" +Tests for Tatlock's permanent tools (calculator, date/time, search). +""" + +import pytest +from datetime import datetime +from unittest.mock import AsyncMock, patch + +from src.agents.tools import ( + calculate, + get_current_datetime, + calculate_time_offset, + time_difference, + search_web, +) + + +# ============================================================================ +# Calculator Tests +# ============================================================================ + +class TestCalculator: + """Tests for the calculator tool.""" + + def test_basic_arithmetic(self): + """Test basic arithmetic operations.""" + assert calculate("2 + 2") == "4" + assert calculate("10 - 3") == "7" + assert calculate("5 * 6") == "30" + assert calculate("20 / 4") == "5" # Integer result, no decimal + + def test_complex_expressions(self): + """Test complex mathematical expressions.""" + assert calculate("(2 + 3) * 4") == "20" + assert calculate("10 ** 2") == "100" + assert calculate("17 % 5") == "2" + + def test_math_functions(self): + """Test mathematical functions.""" + assert calculate("sqrt(16)") == "4" # Integer result + assert calculate("abs(-5)") == "5" + assert calculate("round(3.7)") == "4" + + # Test with constants + result = calculate("pi * 2") + assert "6.28" in result # Approximately 6.283... + + def test_trigonometry(self): + """Test trigonometric functions.""" + result = calculate("sin(0)") + assert result == "0" # Integer result + + # cos(0) should be 1 + result = calculate("cos(0)") + assert result == "1" # Integer result + + def test_logarithms(self): + """Test logarithmic functions.""" + result = calculate("log10(100)") + assert result == "2" # Integer result + + result = calculate("exp(0)") + assert result == "1" # Integer result + + def test_error_handling(self): + """Test error handling for invalid expressions.""" + result = calculate("1 / 0") + assert "Error: Division by zero" in result + + result = calculate("invalid_function(5)") + assert "Error calculating" in result + + def test_integer_results(self): + """Test that integer results don't show unnecessary decimals.""" + assert calculate("4.0 + 6.0") == "10" + assert calculate("sqrt(9)") == "3" + + +# ============================================================================ +# Date/Time Tests +# ============================================================================ + +class TestDateTime: + """Tests for date/time toolkit.""" + + def test_get_current_datetime_full(self): + """Test getting full current datetime.""" + result = get_current_datetime("full") + # Should match format YYYY-MM-DD HH:MM:SS + assert len(result) == 19 + assert result[4] == "-" + assert result[7] == "-" + assert result[10] == " " + assert result[13] == ":" + assert result[16] == ":" + + def test_get_current_datetime_date(self): + """Test getting current date only.""" + result = get_current_datetime("date") + # Should match format YYYY-MM-DD + assert len(result) == 10 + assert result[4] == "-" + assert result[7] == "-" + + # Verify it's a valid date + datetime.strptime(result, "%Y-%m-%d") + + def test_get_current_datetime_time(self): + """Test getting current time only.""" + result = get_current_datetime("time") + # Should match format HH:MM:SS + assert len(result) == 8 + assert result[2] == ":" + assert result[5] == ":" + + def test_get_current_datetime_iso(self): + """Test getting ISO format.""" + result = get_current_datetime("iso") + # Should be parseable as ISO format + datetime.fromisoformat(result) + + def test_calculate_time_offset_days(self): + """Test calculating time offsets in days.""" + result = calculate_time_offset("1 day ago") + assert len(result) == 19 # YYYY-MM-DD HH:MM:SS + + result = calculate_time_offset("2 days from now") + assert len(result) == 19 + + def test_calculate_time_offset_weeks(self): + """Test calculating time offsets in weeks.""" + result = calculate_time_offset("1 week ago") + assert len(result) == 19 + + result = calculate_time_offset("2 weeks from now") + assert len(result) == 19 + + def test_calculate_time_offset_months(self): + """Test calculating time offsets in months.""" + result = calculate_time_offset("1 month ago") + assert len(result) == 19 + + result = calculate_time_offset("3 months from now") + assert len(result) == 19 + + def test_calculate_time_offset_years(self): + """Test calculating time offsets in years.""" + result = calculate_time_offset("1 year ago") + assert len(result) == 19 + + result = calculate_time_offset("2 years from now") + assert len(result) == 19 + + def test_calculate_time_offset_hours(self): + """Test calculating time offsets in hours.""" + result = calculate_time_offset("5 hours ago") + assert len(result) == 19 + + result = calculate_time_offset("3 hours from now") + assert len(result) == 19 + + def test_calculate_time_offset_invalid(self): + """Test error handling for invalid time offsets.""" + result = calculate_time_offset("invalid input") + assert "Error" in result + assert "Cannot parse" in result + + def test_time_difference(self): + """Test calculating time difference.""" + result = time_difference("2024-01-01", "2024-01-15") + assert "14 day" in result + + def test_time_difference_with_now(self): + """Test time difference with 'now'.""" + # Get today's date + today = datetime.now().strftime("%Y-%m-%d") + result = time_difference(today, "now") + # Should be less than a day + assert "Less than" in result or "hour" in result or "minute" in result + + def test_time_difference_with_times(self): + """Test time difference with full timestamps.""" + result = time_difference("2024-01-01 10:00:00", "2024-01-01 14:30:00") + assert "4 hour" in result + assert "30 minute" in result + + def test_time_difference_error(self): + """Test error handling for invalid dates.""" + result = time_difference("invalid-date", "now") + assert "Error" in result + + +# ============================================================================ +# Search Tests +# ============================================================================ + +class TestSearch: + """Tests for web search tool.""" + + @pytest.mark.asyncio + async def test_search_web_success(self): + """Test successful web search.""" + mock_response = { + "results": [ + { + "title": "Test Result 1", + "url": "https://example.com/1", + "content": "This is a test result" + }, + { + "title": "Test Result 2", + "url": "https://example.com/2", + "content": "Another test result" + } + ] + } + + with patch("src.agents.tools.httpx.AsyncClient") as mock_client_class: + # Create mock response + mock_response_obj = type('MockResponse', (), { + 'status_code': 200, + 'json': lambda *args, **kwargs: mock_response + })() + + # Create mock client with async get method + async def mock_get(*args, **kwargs): + return mock_response_obj + + mock_client_instance = type('MockClient', (), { + 'get': mock_get + })() + + # Setup async context manager + async def mock_aenter(*args, **kwargs): + return mock_client_instance + + async def mock_aexit(*args, **kwargs): + return None + + mock_client_class.return_value.__aenter__ = mock_aenter + mock_client_class.return_value.__aexit__ = mock_aexit + + result = await search_web("test query", num_results=2) + + assert "Test Result 1" in result + assert "https://example.com/1" in result + assert "Test Result 2" in result + assert "https://example.com/2" in result + + @pytest.mark.asyncio + async def test_search_web_no_results(self): + """Test web search with no results.""" + mock_response_data = {"results": []} + + with patch("src.agents.tools.httpx.AsyncClient") as mock_client_class: + mock_response_obj = type('MockResponse', (), { + 'status_code': 200, + 'json': lambda *args, **kwargs: mock_response_data + })() + + async def mock_get(*args, **kwargs): + return mock_response_obj + + mock_client_instance = type('MockClient', (), { + 'get': mock_get + })() + + async def mock_aenter(*args, **kwargs): + return mock_client_instance + + async def mock_aexit(*args, **kwargs): + return None + + mock_client_class.return_value.__aenter__ = mock_aenter + mock_client_class.return_value.__aexit__ = mock_aexit + + result = await search_web("test query") + + assert "No results found" in result + + @pytest.mark.asyncio + async def test_search_web_connection_error(self): + """Test web search with connection error.""" + with patch("httpx.AsyncClient") as mock_client: + mock_client_instance = AsyncMock() + mock_client_instance.get.side_effect = Exception("Connection failed") + mock_client.return_value.__aenter__.return_value = mock_client_instance + + result = await search_web("test query") + + assert "Error searching" in result + + @pytest.mark.asyncio + async def test_search_web_limits_results(self): + """Test that search limits results to max 10.""" + mock_response_data = { + "results": [ + {"title": f"Result {i}", "url": f"https://example.com/{i}", "content": "Test"} + for i in range(20) + ] + } + + with patch("src.agents.tools.httpx.AsyncClient") as mock_client_class: + mock_response_obj = type('MockResponse', (), { + 'status_code': 200, + 'json': lambda *args, **kwargs: mock_response_data + })() + + async def mock_get(*args, **kwargs): + return mock_response_obj + + mock_client_instance = type('MockClient', (), { + 'get': mock_get + })() + + async def mock_aenter(*args, **kwargs): + return mock_client_instance + + async def mock_aexit(*args, **kwargs): + return None + + mock_client_class.return_value.__aenter__ = mock_aenter + mock_client_class.return_value.__aexit__ = mock_aexit + + result = await search_web("test query", num_results=15) + + # Should only return 10 results (max limit) + result_count = result.count("URL:") + assert result_count == 10 + + @pytest.mark.asyncio + async def test_search_web_formats_results(self): + """Test that search results are properly formatted.""" + mock_response_data = { + "results": [ + { + "title": "Test Title", + "url": "https://example.com", + "content": "Test content description" + } + ] + } + + with patch("src.agents.tools.httpx.AsyncClient") as mock_client_class: + mock_response_obj = type('MockResponse', (), { + 'status_code': 200, + 'json': lambda *args, **kwargs: mock_response_data + })() + + async def mock_get(*args, **kwargs): + return mock_response_obj + + mock_client_instance = type('MockClient', (), { + 'get': mock_get + })() + + async def mock_aenter(*args, **kwargs): + return mock_client_instance + + async def mock_aexit(*args, **kwargs): + return None + + mock_client_class.return_value.__aenter__ = mock_aenter + mock_client_class.return_value.__aexit__ = mock_aexit + + result = await search_web("test query") + + # Check formatting + assert "1. Test Title" in result + assert "URL: https://example.com" in result + assert "Test content description" in result