What are Tools?

Tools in Slide are the capabilities that agents can use to interact with the world. They bridge the gap between an agent’s decision-making and actual actions. The Lye package provides a comprehensive set of pre-built tools, and you can easily create custom ones.

Built-in Tools

Slide comes with a rich set of tools organized by category:
from lye import AUDIO_TOOLS
from lye.audio import transcribe, text_to_speech

# Use all audio tools
agent = Agent(tools=AUDIO_TOOLS)

# Or specific tools
agent = Agent(tools=[transcribe, text_to_speech])

Tool Structure

Every tool in Slide follows the OpenAI function calling format:
# Example of a Lye tool definition
def custom_tool_implementation(param1: str, param2: int = 10) -> str:
    """Implementation of the tool."""
    # Your logic here
    return f"Processed {param1} with value {param2}"

# Tool definition
custom_tool = {
    "definition": {
        "type": "function",
        "function": {
            "name": "custom_tool",
            "description": "Does something useful with the input",
            "parameters": {
                "type": "object",
                "properties": {
                    "param1": {
                        "type": "string",
                        "description": "The main input parameter"
                    },
                    "param2": {
                        "type": "integer",
                        "description": "Optional configuration value",
                        "default": 10
                    }
                },
                "required": ["param1"]
            }
        }
    },
    "implementation": custom_tool_implementation
}

Creating custom tools

Simple function tool

The easiest way to create a tool:
def word_counter_implementation(text: str) -> str:
    """Count words in the provided text."""
    words = text.split()
    result = {
        "word_count": len(words),
        "character_count": len(text),
        "unique_words": len(set(words))
    }
    return str(result)

# Create tool definition
word_counter = {
    "definition": {
        "type": "function",
        "function": {
            "name": "word_counter",
            "description": "Count words in text",
            "parameters": {
                "type": "object",
                "properties": {
                    "text": {
                        "type": "string",
                        "description": "The text to analyze"
                    }
                },
                "required": ["text"]
            }
        }
    },
    "implementation": word_counter_implementation
}

# Use with an agent
from tyler import Agent

agent = Agent(
    name="text-analyzer",
    model_name="gpt-4",
    purpose="To analyze text",
    tools=[word_counter]
)

Advanced tool example

For more complex tools with external dependencies:
import aiohttp
import json

async def weather_tool_implementation(city: str, units: str = "celsius") -> str:
    """Get current weather for a city."""
    api_key = "your-api-key"  # In practice, use environment variables
    
    async with aiohttp.ClientSession() as session:
        url = f"https://api.weather.com/v1/current"
        params = {
            "city": city,
            "units": units,
            "api_key": api_key
        }
        
        try:
            async with session.get(url, params=params) as response:
                if response.status == 200:
                    data = await response.json()
                    return json.dumps({
                        "temperature": data["temp"],
                        "conditions": data["conditions"],
                        "humidity": data["humidity"]
                    })
                else:
                    return json.dumps({"error": f"API returned status {response.status}"})
        except Exception as e:
            return json.dumps({"error": str(e)})

# Tool definition
weather_tool = {
    "definition": {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name"
                    },
                    "units": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "default": "celsius"
                    }
                },
                "required": ["city"]
            }
        }
    },
    "implementation": weather_tool_implementation
}

Tool patterns

1. Validation Pattern

Always validate inputs:
def safe_calculator_implementation(expression: str) -> str:
    # Validate input
    allowed_chars = set("0123456789+-*/()., ")
    if not all(c in allowed_chars for c in expression):
        return json.dumps({"error": "Invalid characters in expression"})
    
    try:
        # Safe evaluation after validation
        result = eval(expression)
        return json.dumps({"result": result})
    except Exception as e:
        return json.dumps({"error": str(e)})

safe_calculator = {
    "definition": {
        "type": "function",
        "function": {
            "name": "safe_calculator",
            "description": "Safe math calculations",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "Mathematical expression to evaluate"
                    }
                },
                "required": ["expression"]
            }
        }
    },
    "implementation": safe_calculator_implementation
}

2. Resource Management Pattern

Properly manage external resources:
import asyncpg
import json

async def database_query_implementation(query: str, database: str = "main") -> str:
    """Execute a database query safely."""
    conn = None
    try:
        # Connect to database
        conn = await asyncpg.connect(f"postgresql://localhost/{database}")
        
        # Execute query
        result = await conn.fetch(query)
        
        # Convert to JSON-serializable format
        data = [dict(record) for record in result]
        return json.dumps({"data": data, "count": len(data)})
        
    except Exception as e:
        return json.dumps({"error": str(e)})
    finally:
        if conn:
            await conn.close()  # Always cleanup

database_tool = {
    "definition": {
        "type": "function",
        "function": {
            "name": "query_database",
            "description": "Execute a database query",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "SQL query to execute"
                    },
                    "database": {
                        "type": "string",
                        "description": "Database name",
                        "default": "main"
                    }
                },
                "required": ["query"]
            }
        }
    },
    "implementation": database_query_implementation
}

3. Error Handling Pattern

Provide meaningful error messages:
import aiohttp
import json

async def api_caller_implementation(url: str, method: str = "GET", data: dict = None) -> str:
    """Make API calls with proper error handling."""
    try:
        async with aiohttp.ClientSession() as session:
            kwargs = {"url": url}
            if data and method in ["POST", "PUT", "PATCH"]:
                kwargs["json"] = data
                
            async with session.request(method, **kwargs) as response:
                response_text = await response.text()
                
                if response.status >= 400:
                    return json.dumps({
                        "error": f"API error: {response.status}",
                        "details": response_text
                    })
                
                try:
                    response_data = json.loads(response_text)
                    return json.dumps({"data": response_data})
                except json.JSONDecodeError:
                    return json.dumps({"data": response_text})
                    
    except aiohttp.ClientError as e:
        return json.dumps({"error": f"Network error: {str(e)}"})
    except Exception as e:
        return json.dumps({"error": f"Unexpected error: {str(e)}"})

api_tool = {
    "definition": {
        "type": "function",
        "function": {
            "name": "api_caller",
            "description": "Make API calls",
            "parameters": {
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "API endpoint URL"
                    },
                    "method": {
                        "type": "string",
                        "enum": ["GET", "POST", "PUT", "DELETE", "PATCH"],
                        "default": "GET"
                    },
                    "data": {
                        "type": "object",
                        "description": "Request body data"
                    }
                },
                "required": ["url"]
            }
        }
    },
    "implementation": api_caller_implementation
}

4. Rate Limiting

Implement rate limiting for external services:
import asyncio
from datetime import datetime, timedelta
import json

class RateLimiter:
    def __init__(self, max_calls: int, time_window: timedelta):
        self.max_calls = max_calls
        self.time_window = time_window
        self.calls = []
        self.lock = asyncio.Lock()
    
    async def acquire(self):
        async with self.lock:
            now = datetime.now()
            # Remove old calls outside the time window
            self.calls = [call_time for call_time in self.calls 
                         if now - call_time < self.time_window]
            
            if len(self.calls) >= self.max_calls:
                # Calculate wait time
                oldest_call = min(self.calls)
                wait_time = (oldest_call + self.time_window - now).total_seconds()
                if wait_time > 0:
                    await asyncio.sleep(wait_time)
                    return await self.acquire()  # Retry
            
            self.calls.append(now)

# Global rate limiter for API calls
api_limiter = RateLimiter(max_calls=60, time_window=timedelta(minutes=1))

async def rate_limited_api_implementation(endpoint: str) -> str:
    """Make rate-limited API calls."""
    await api_limiter.acquire()
    
    # Make the actual API call
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/{endpoint}") as response:
            data = await response.json()
            return json.dumps(data)

Tool Best Practices

Testing tools

Always test your custom tools:
import pytest
import json

@pytest.mark.asyncio
async def test_weather_tool():
    # Mock the implementation for testing
    async def mock_weather_implementation(city: str, units: str = "celsius") -> str:
        if city == "London":
            return json.dumps({
                "temperature": 15,
                "conditions": "Cloudy",
                "humidity": 70
            })
        else:
            return json.dumps({"error": "City not found"})
    
    # Replace implementation for testing
    weather_tool["implementation"] = mock_weather_implementation
    
    # Test successful call
    result = await weather_tool["implementation"]("London")
    data = json.loads(result)
    
    assert "temperature" in data
    assert "conditions" in data
    assert data["temperature"] == 15

@pytest.mark.asyncio
async def test_weather_tool_error():
    result = await weather_tool["implementation"]("InvalidCity")
    data = json.loads(result)
    
    assert "error" in data

Tool Composition

Combine multiple tools for complex operations:
from tyler import Agent, Thread, Message
from lye import WEB_TOOLS, FILES_TOOLS, IMAGE_TOOLS

# Create a research agent with multiple tool categories
agent = Agent(
    name="researcher",
    model_name="gpt-4",
    purpose="To conduct comprehensive research",
    tools=[
        *WEB_TOOLS,      # search, fetch
        *FILES_TOOLS,    # read_file, write_file
        *IMAGE_TOOLS     # analyze_image, extract_text_from_image
    ]
)

# Example usage
thread = Thread()
message = Message(
    role="user",
    content="""
    1. Search for information about the James Webb telescope
    2. Fetch detailed content from NASA's website
    3. Analyze any images found
    4. Save a comprehensive report
    """
)
thread.add_message(message)

result = await agent.go(thread)

Next steps