simple tools for https://github.com/cameroncking/llm-sandbox

sysdata_tools.py

"""Tools for memory management and utilities."""

import os
import urllib.request
import urllib.parse
import json
import datetime
import zoneinfo


# Files are stored in /userdata which is the writable directory
MEMORY_PATH = "/userdata/MEMORY.md"


# WMO weather codes to descriptions
_WEATHER_CODES = {
    0: "Clear sky",
    1: "Mainly clear",
    2: "Partly cloudy",
    3: "Overcast",
    45: "Foggy",
    48: "Depositing rime fog",
    51: "Light drizzle",
    53: "Moderate drizzle",
    55: "Dense drizzle",
    61: "Slight rain",
    63: "Moderate rain",
    65: "Heavy rain",
    66: "Light freezing rain",
    67: "Heavy freezing rain",
    71: "Slight snow",
    73: "Moderate snow",
    75: "Heavy snow",
    77: "Snow grains",
    80: "Slight rain showers",
    81: "Moderate rain showers",
    82: "Violent rain showers",
    85: "Slight snow showers",
    86: "Heavy snow showers",
    95: "Thunderstorm",
    96: "Thunderstorm with slight hail",
    99: "Thunderstorm with heavy hail",
}


# =============================================================================
# Memory Tools
# =============================================================================

def read_memory() -> str:
    """Read the contents of MEMORY.md.
    
    Returns:
        The contents of MEMORY.md, or a message if it doesn't exist.
    """
    if os.path.exists(MEMORY_PATH):
        with open(MEMORY_PATH, "r") as f:
            return f.read()
    return "No memories yet."


def write_memory(content: str) -> str:
    """Write content to MEMORY.md.
    
    Args:
        content: The full content to write to MEMORY.md
    
    Returns:
        A confirmation message.
    """
    with open(MEMORY_PATH, "w") as f:
        f.write(content)
    return "MEMORY.md has been updated."


def append_memory(content: str) -> str:
    """Append content to the end of MEMORY.md.
    
    Args:
        content: The content to append to MEMORY.md
    
    Returns:
        A confirmation message.
    """
    with open(MEMORY_PATH, "a") as f:
        f.write(content)
    return "Content appended to MEMORY.md."


def patch_memory(search: str, replace: str) -> str:
    """Replace a unique occurrence of search string with replace string in MEMORY.md.
    
    Args:
        search: The text to find (must appear exactly once)
        replace: The text to replace it with
    
    Returns:
        A confirmation message, or an error if search is not found or ambiguous.
    """
    if not os.path.exists(MEMORY_PATH):
        return "Error: MEMORY.md does not exist."
    
    with open(MEMORY_PATH, "r") as f:
        content = f.read()
    
    count = content.count(search)
    
    if count == 0:
        return "Error: Search string not found in MEMORY.md."
    if count > 1:
        return f"Error: Search string matches {count} locations. Must be unique."
    
    new_content = content.replace(search, replace, 1)
    
    with open(MEMORY_PATH, "w") as f:
        f.write(new_content)
    
    return "MEMORY.md has been patched."


def clear_memory() -> str:
    """Erase MEMORY.md completely.
    
    Returns:
        A confirmation message.
    """
    if os.path.exists(MEMORY_PATH):
        os.remove(MEMORY_PATH)
        return "MEMORY.md has been cleared."
    return "MEMORY.md did not exist."


# =============================================================================
# Utility Tools
# =============================================================================

def _geocode(location: str) -> tuple:
    """Convert location name to coordinates using Open-Meteo geocoding."""
    encoded = urllib.parse.quote(location)
    url = f"https://geocoding-api.open-meteo.com/v1/search?name={encoded}&count=1"
    
    req = urllib.request.Request(url)
    with urllib.request.urlopen(req, timeout=15) as response:
        data = json.loads(response.read().decode("utf-8"))
    
    if not data.get("results"):
        raise ValueError(f"Location not found: {location}")
    
    result = data["results"][0]
    return (
        result["latitude"],
        result["longitude"],
        result["name"],
        result.get("admin1", ""),
        result.get("country", "")
    )


def get_weather(location: str) -> str:
    """Get current weather for a location.
    
    Args:
        location: City name or place (e.g., "London", "New York", "Tokyo")
    
    Returns:
        Current weather information including temperature, conditions, and wind.
    """
    try:
        # Geocode the location
        lat, lon, name, region, country = _geocode(location)
        
        # Get weather from Open-Meteo
        url = (
            f"https://api.open-meteo.com/v1/forecast?"
            f"latitude={lat}&longitude={lon}"
            f"&current=temperature_2m,relative_humidity_2m,apparent_temperature,weather_code,wind_speed_10m,wind_direction_10m,is_day"
        )
        
        req = urllib.request.Request(url)
        with urllib.request.urlopen(req, timeout=15) as response:
            data = json.loads(response.read().decode("utf-8"))
        
        current = data["current"]
        
        temp_c = current["temperature_2m"]
        temp_f = round(temp_c * 9/5 + 32, 1)
        humidity = current["relative_humidity_2m"]
        feels_like_c = current["apparent_temperature"]
        wind_kmh = current["wind_speed_10m"]
        wind_mph = round(wind_kmh * 0.621371, 1)
        wind_dir = current["wind_direction_10m"]
        weather_code = current["weather_code"]
        is_day = current["is_day"]
        
        conditions = _WEATHER_CODES.get(weather_code, f"Unknown ({weather_code})")
        day_night = "Day" if is_day else "Night"
        
        # Build location string
        location_str = name
        if region:
            location_str += f", {region}"
        if country:
            location_str += f", {country}"
        
        # Convert feels like to F
        feels_like_f = round(feels_like_c * 9/5 + 32, 1)
        feels_like_str = f"{feels_like_c}°C ({feels_like_f}°F)"
        
        report = f"""Weather for {location_str}:

Conditions: {conditions}
Temperature: {temp_c}°C ({temp_f}°F)
Feels like: {feels_like_str}
Humidity: {humidity}%
Wind: {wind_kmh} km/h ({wind_mph} mph) from {wind_dir}°
Time of day: {day_night}"""
        
        return report
        
    except ValueError as e:
        return str(e)
    except urllib.error.HTTPError as e:
        return f"Error fetching weather: HTTP {e.code}"
    except urllib.error.URLError as e:
        return f"Error fetching weather: Network error - {e.reason}"
    except (KeyError, IndexError, json.JSONDecodeError) as e:
        return f"Error parsing weather data: {e}"
    except Exception as e:
        return f"Error fetching weather: {e}"


def get_current_datetime(tz: str = None) -> str:
    """Get the current date and time.
    
    Args:
        tz: Optional timezone name (e.g., "America/New_York", "Europe/London", "Asia/Tokyo").
            When the user's timezone is known, provide it to return localized time.
            If not provided, returns UTC time.
    
    Returns:
        Current date and time as a formatted string with timezone info.
    """
    try:
        if tz:
            zone = zoneinfo.ZoneInfo(tz)
            now = datetime.datetime.now(zone)
            tz_label = tz
        else:
            now = datetime.datetime.now(datetime.timezone.utc)
            tz_label = "UTC"
        
        return f"{now.strftime('%Y-%m-%d %H:%M:%S')} ({tz_label})"
    except Exception as e:
        return f"Error getting time: {e}"


# =============================================================================
# Web Search
# =============================================================================

def web_search(query: str, count: int = 5) -> str:
    """Search the web.
    
    Args:
        query: The search query string
        count: Number of results to return (default 5, max 20)
    
    Returns:
        Formatted search results with titles, URLs, and descriptions.
    """
    api_key = os.environ.get("BRAVE_SEARCH_API_KEY", "")
    if not api_key:
        return "Error: web_search tool is not available"
    
    count = min(max(1, count), 20)  # Clamp between 1 and 20
    
    try:
        encoded_query = urllib.parse.quote(query)
        url = f"https://api.search.brave.com/res/v1/web/search?q={encoded_query}&count={count}"
        
        req = urllib.request.Request(
            url,
            headers={
                "Accept": "application/json",
                "X-Subscription-Token": api_key
            }
        )
        
        with urllib.request.urlopen(req, timeout=15) as response:
            data = json.loads(response.read().decode("utf-8"))
        
        results = []
        web_results = data.get("web", {}).get("results", [])
        
        if not web_results:
            return f"No results found for: {query}"
        
        for i, result in enumerate(web_results[:count], 1):
            title = result.get("title", "No title")
            url = result.get("url", "")
            description = result.get("description", "No description")
            
            results.append(f"{i}. {title}\n   URL: {url}\n   {description}")
        
        return f"Search results for: {query}\n\n" + "\n\n".join(results)
        
    except urllib.error.HTTPError as e:
        return f"Error searching: HTTP {e.code}"
    except urllib.error.URLError as e:
        return f"Error searching: Network error - {e.reason}"
    except (KeyError, json.JSONDecodeError) as e:
        return f"Error parsing search results: {e}"
    except Exception as e:
        return f"Error searching: {e}"


# =============================================================================
# MCP Image Generation
# =============================================================================

_MCP_IMAGE_GEN_URL = os.environ.get("MCP_IMAGE_GEN_URL", "")
_MCP_IMAGE_GEN_TOKEN = os.environ.get("MCP_IMAGE_GEN_TOKEN", "")


def _call_mcp_tool(url: str, token: str, tool_name: str, arguments: dict, timeout: float = 120.0) -> str:
    """Internal helper to call an MCP tool via HTTP transport."""
    import asyncio
    import httpx
    from mcp.client.session import ClientSession
    from mcp.client.streamable_http import streamable_http_client
    
    async def _do_call():
        headers = {}
        if token:
            headers["Authorization"] = f"Bearer {token}"
        
        async with httpx.AsyncClient(headers=headers, timeout=httpx.Timeout(timeout)) as http_client:
            async with streamable_http_client(url, http_client=http_client) as (read, write, _):
                async with ClientSession(read, write) as session:
                    await session.initialize()
                    result = await session.call_tool(tool_name, arguments)
                    if result.content:
                        texts = []
                        for item in result.content:
                            if hasattr(item, 'text'):
                                texts.append(item.text)
                        return "\n".join(texts) if texts else ""
                    return ""
    
    return asyncio.run(_do_call())


def generate_image(prompt: str) -> str:
    """Generate an image from a text description and return the download URL.
    
    Args:
        prompt: A detailed text description of the image to generate.
                Be specific about style, colors, composition, and subject matter.
    
    Returns:
        A URL where the generated image can be downloaded.
    """
    if not _MCP_IMAGE_GEN_URL:
        return "Error: MCP_IMAGE_GEN_URL environment variable not set"
    
    return _call_mcp_tool(
        _MCP_IMAGE_GEN_URL,
        _MCP_IMAGE_GEN_TOKEN,
        "generate_image_url_from_prompt",
        {"prompt": prompt},
        timeout=120.0
    )