mims-harvard__tooluniverse/docs/dev_docs/Adding_Tools_Tutorial.md

Adding Tools to ToolUniverse - Complete Tutorial

This tutorial covers everything you need to know about adding custom tools to ToolUniverse using the decorator-based auto-registration system.

Table of Contents

1. :ref:overview
2. :ref:development-environment-setup
3. :ref:quick-start
4. :ref:method-1-decorator-registration-recommended
5. :ref:method-2-manual-registration
6. :ref:tool-configuration
7. :ref:parameter-validation-and-error-handling
8. :ref:real-world-examples
9. :ref:best-practices
10. :ref:troubleshooting

.. _overview:

Overview

ToolUniverse supports automatic tool discovery through decorators. When you add a new tool, it's automatically registered and available without needing to manually edit core files.

What You Can Add

• Custom API wrappers
• Data processing tools
• File manipulation utilities
• External service integrations
• Analysis and computation tools

What Changed

• Before: Manual imports and mappings in execute_function.py
• Now: Simple decorator registration with auto-discovery

.. _development-environment-setup:

Development Environment Setup

Before adding tools, ensure you have the proper development environment set up:

1. Clone and Install ToolUniverse

# Clone the repository
git clone https://github.com/mims-harvard/ToolUniverse.git
cd ToolUniverse

# Install in development mode
pip install -e ".[dev]"

2. Set Up Pre-commit Hooks (Recommended)

Automatic Setup:

# Auto-install pre-commit hooks
./setup_precommit.sh

Manual Setup:

# Install pre-commit if not already installed
pip install pre-commit

# Install hooks
pre-commit install

# Update to latest versions
pre-commit autoupdate

Pre-commit Benefits:

• Automatic code formatting with Black
• Code linting with flake8 and ruff
• Import cleanup with autoflake
• File validation (YAML, TOML, AST checks)
• Runs on every commit to ensure code quality

3. Verify Installation

from tooluniverse import ToolUniverse

tu = ToolUniverse()
tu.load_tools()
print(f"✅ Loaded {len(tu.all_tools)} tools!")

.. _quick-start:

Quick Start

Here's the fastest way to add a new tool:

# my_custom_tool.py
from tooluniverse.tool_registry import register_tool

@register_tool('MyAwesomeTool', config={
   "name": "my_awesome_tool",
   "type": "MyAwesomeTool",
   "description": "A simple example tool",
   "parameter": {
       "type": "object",
       "properties": {
           "message": {"type": "string", "description": "Message to process"}
       },
       "required": ["message"]
   }
})
class MyAwesomeTool:
   def __init__(self, tool_config=None):
       self.tool_config = tool_config

   def run(self, arguments):
       message = arguments.get('message', '')
       return {
           'processed_message': f"Hello! You said: '{message}'",
           'original': message,
           'tool_name': 'MyAwesomeTool'
       }

# Usage
from tooluniverse import ToolUniverse

tu = ToolUniverse()
tu.load_tools()  # Auto-discovers and loads your tool

result = tu.run_one_function({
   "name": "my_awesome_tool",
   "arguments": {"message": "This is a test"}
})
print(result)

That's it! Your tool is automatically discovered and ready to use.

.. _method-1-decorator-registration-recommended:

Method 1: Decorator Registration (Recommended)

The decorator approach is the simplest and most powerful way to add tools.

Basic Structure

from tooluniverse.tool_registry import register_tool

@register_tool('YourToolType', config={...})
class YourTool:
   def __init__(self, tool_config=None):
       # Initialize your tool
       pass

   def run(self, arguments):
       # Process arguments and return results
       return {...}

Step-by-Step Tutorial

1. Create Your Tool File

Create a new Python file ending with _tool.py (e.g., weather_tool.py):

# weather_tool.py
import requests
from tooluniverse.tool_registry import register_tool

@register_tool('WeatherTool', config={
   "name": "get_weather",
   "type": "WeatherTool",
   "description": "Get current weather for a city",
   "parameter": {
       "type": "object",
       "properties": {
           "city": {
               "type": "string",
               "description": "City name to get weather for"
           },
           "units": {
               "type": "string",
               "description": "Temperature units (metric/imperial)",
               "default": "metric"
           }
       },
       "required": ["city"]
   }
})
class WeatherTool:
   def __init__(self, tool_config=None):
       self.tool_config = tool_config
       self.api_key = "your_api_key_here"  # In practice, use env vars

   def run(self, arguments):
       city = arguments.get('city')
       units = arguments.get('units', 'metric')

       # Mock API call (replace with real API)
       url = f"http://api.openweathermap.org/data/2.5/weather"
       params = {
           "q": city,
           "appid": self.api_key,
           "units": units
       }

       try:
           response = requests.get(url, params=params)
           data = response.json()

           return {
               "city": city,
               "temperature": data["main"]["temp"],
               "description": data["weather"][0]["description"],
               "units": units,
               "success": True
           }
       except Exception as e:
           return {
               "error": f"Failed to get weather: {str(e)}",
               "success": False
           }

2. Place in ToolUniverse Package

Move your file to the ToolUniverse source directory:

cp weather_tool.py /path/to/ToolUniverse/src/tooluniverse/

3. Use Your Tool

Your tool is automatically discovered:

from tooluniverse import ToolUniverse

tu = ToolUniverse()
tu.load_tools()

# Your tool is now available!
result = tu.run_one_function({
   "name": "get_weather",
   "arguments": {"city": "San Francisco", "units": "metric"}
})

print(result)

Configuration Options

The config parameter in the decorator supports all standard ToolUniverse configuration:

@register_tool('AdvancedTool', config={
   "name": "advanced_tool",
   "type": "AdvancedTool",
   "description": "An advanced tool with complex parameters",
   "parameter": {
       "type": "object",
       "properties": {
           "input_data": {
               "type": "object",
               "description": "Complex input data",
               "properties": {
                   "values": {"type": "array", "items": {"type": "number"}},
                   "operation": {"type": "string", "enum": ["sum", "avg", "max"]}
               }
           },
           "options": {
               "type": "object",
               "description": "Optional settings",
               "properties": {
                   "precision": {"type": "integer", "default": 2},
                   "format": {"type": "string", "default": "json"}
               }
           }
       },
       "required": ["input_data"]
   },
   "settings": {
       "timeout": 30,
       "max_retries": 3,
       "custom_field": "value"
   }
})
class AdvancedTool:
   ...

.. _method-2-manual-registration:

Method 2: Manual Registration

For dynamic tools or special cases, you can register tools at runtime:

from tooluniverse import ToolUniverse

class DynamicTool:
   def __init__(self, tool_config=None):
       self.tool_config = tool_config
       self.dynamic_data = self._load_dynamic_data()

   def run(self, arguments):
       return {"result": f"Processed {arguments}"}

   def _load_dynamic_data(self):
       # Load configuration from database, API, etc.
       return {}

# Create ToolUniverse instance
tu = ToolUniverse()
tu.load_tools()

# Manual registration
tool_config = {
   "name": "dynamic_tool",
   "type": "DynamicTool",
   "description": "A dynamically configured tool",
   "parameter": {
       "type": "object",
       "properties": {
           "input": {"type": "string", "description": "Input data"}
       },
       "required": ["input"]
   }
}

tu.register_custom_tool(DynamicTool, 'DynamicTool', tool_config)

# Use the manually registered tool
result = tu.run_one_function({
   "name": "dynamic_tool",
   "arguments": {"input": "test"}
})

.. _tool-configuration:

Tool Configuration

Required Fields

• name: Unique identifier for your tool
• type: Class name (must match decorator parameter)
• description: Human-readable description
• parameter: JSON Schema defining input parameters

Optional Fields

• settings: Tool-specific configuration
• version: Tool version
• tags: Categorization tags
• examples: Usage examples

Parameter Schema Examples

Simple Parameters

{
   "type": "object",
   "properties": {
       "text": {"type": "string", "description": "Input text"},
       "count": {"type": "integer", "minimum": 1, "maximum": 100}
   },
   "required": ["text"]
}

Complex Parameters

{
   "type": "object",
   "properties": {
       "data": {
           "type": "array",
           "items": {
               "type": "object",
               "properties": {
                   "id": {"type": "string"},
                   "value": {"type": "number"}
               }
           }
       },
       "options": {
           "type": "object",
           "properties": {
               "format": {"type": "string", "enum": ["json", "csv", "xml"]},
               "include_metadata": {"type": "boolean", "default": false}
           }
       }
   },
   "required": ["data"]
}

.. _parameter-validation-and-error-handling:

Parameter Validation and Error Handling

ToolUniverse provides built-in support for parameter validation and structured error handling to help you create robust tools.

Parameter Validation

You can implement custom parameter validation by overriding the validate_parameters() method:

from tooluniverse.exceptions import ToolValidationError

class MyTool:
   def validate_parameters(self, arguments):
       # First run base validation
       base_error = super().validate_parameters(arguments)
       if base_error:
           return base_error

       # Add custom validation
       if "email" in arguments:
           email = arguments["email"]
           if "@" not in email:
               return ToolValidationError(
                   "Invalid email format",
                   next_steps=["Provide a valid email address"],
                   details={"field": "email", "value": email}
               )

       return None  # Validation passed

Error Handling

Override handle_error() to provide tool-specific error classification:

from tooluniverse.exceptions import ToolAuthError, ToolRateLimitError

class APITool:
   def handle_error(self, exception):
       error_str = str(exception).lower()

       if "api key" in error_str:
           return ToolAuthError(
               "API authentication failed",
               next_steps=["Check API key", "Verify environment variables"]
           )
       elif "rate limit" in error_str:
           return ToolRateLimitError(
               "API rate limit exceeded",
               next_steps=["Wait and retry", "Check quota limits"]
           )

       # Fall back to base error handling
       return super().handle_error(exception)

Exception System Migration

If you're using older exception classes, migrate to the new structured system:

# Old way (deprecated)
from tooluniverse.base_tool import ValidationError
raise ValidationError("Invalid input")

# New way (recommended)
from tooluniverse.exceptions import ToolValidationError
raise ToolValidationError(
   "Invalid input",
   next_steps=["Check parameter format"],
   details={"field": "email"}
)

For more information about the exception system, see the API documentation.

.. _real-world-examples:

Real-World Examples

Example 1: Database Query Tool

# database_tool.py
import sqlite3
from typing import Dict, Any, List
from tooluniverse.tool_registry import register_tool

@register_tool('DatabaseQueryTool', config={
   "name": "query_database",
   "type": "DatabaseQueryTool",
   "description": "Execute SQL queries on a database",
   "parameter": {
       "type": "object",
       "properties": {
           "query": {"type": "string", "description": "SQL query to execute"},
           "database": {"type": "string", "description": "Database name"},
           "limit": {"type": "integer", "default": 100, "maximum": 1000}
       },
       "required": ["query", "database"]
   },
   "settings": {
       "db_path": "/path/to/databases/",
       "readonly": true
   }
})
class DatabaseQueryTool:
   def __init__(self, tool_config: Dict[str, Any]):
       self.tool_config = tool_config
       self.db_path = tool_config.get("settings", {}).get("db_path", ".")
       self.readonly = tool_config.get("settings", {}).get("readonly", True)

   def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
       query = arguments.get('query')
       database = arguments.get('database')
       limit = arguments.get('limit', 100)

       try:
           db_file = f"{self.db_path}/{database}.db"
           conn = sqlite3.connect(db_file)
           conn.row_factory = sqlite3.Row  # Enable column access by name

           cursor = conn.cursor()

           # Add LIMIT if not present in query
           if 'LIMIT' not in query.upper():
               query += f" LIMIT {limit}"

           cursor.execute(query)
           rows = cursor.fetchall()

           # Convert to list of dictionaries
           results = [dict(row) for row in rows]

           conn.close()

           return {
               "results": results,
               "count": len(results),
               "query": query,
               "database": database,
               "success": True
           }

       except Exception as e:
           return {
               "error": str(e),
               "query": query,
               "database": database,
               "success": False
           }

Example 2: File Processing Tool

# file_processor_tool.py
import os
import json
import csv
from pathlib import Path
from tooluniverse.tool_registry import register_tool

@register_tool('FileProcessorTool', config={
   "name": "process_file",
   "type": "FileProcessorTool",
   "description": "Process and transform files between different formats",
   "parameter": {
       "type": "object",
       "properties": {
           "input_file": {"type": "string", "description": "Path to input file"},
           "output_format": {"type": "string", "enum": ["json", "csv", "txt"]},
           "operation": {"type": "string", "enum": ["convert", "analyze", "validate"]}
       },
       "required": ["input_file", "operation"]
   }
})
class FileProcessorTool:
   def __init__(self, tool_config=None):
       self.tool_config = tool_config
       self.supported_formats = ['.json', '.csv', '.txt', '.xml']

   def run(self, arguments):
       input_file = arguments.get('input_file')
       operation = arguments.get('operation')
       output_format = arguments.get('output_format')

       if not os.path.exists(input_file):
           return {"error": f"File not found: {input_file}", "success": False}

       try:
           if operation == "analyze":
               return self._analyze_file(input_file)
           elif operation == "convert":
               return self._convert_file(input_file, output_format)
           elif operation == "validate":
               return self._validate_file(input_file)
           else:
               return {"error": f"Unknown operation: {operation}", "success": False}

       except Exception as e:
           return {"error": str(e), "success": False}

   def _analyze_file(self, file_path):
       file_stats = os.stat(file_path)
       file_ext = Path(file_path).suffix.lower()

       analysis = {
           "file_path": file_path,
           "size_bytes": file_stats.st_size,
           "extension": file_ext,
           "readable": os.access(file_path, os.R_OK),
           "success": True
       }

       # Format-specific analysis
       if file_ext == '.json':
           try:
               with open(file_path, 'r') as f:
                   data = json.load(f)
               analysis["json_valid"] = True
               analysis["json_type"] = type(data).__name__
               if isinstance(data, list):
                   analysis["items_count"] = len(data)
               elif isinstance(data, dict):
                   analysis["keys_count"] = len(data.keys())
           except:
               analysis["json_valid"] = False

       return analysis

   def _convert_file(self, input_file, output_format):
       # Implementation for file conversion
       return {
           "message": f"Would convert {input_file} to {output_format}",
           "input_file": input_file,
           "output_format": output_format,
           "success": True
       }

   def _validate_file(self, file_path):
       # Implementation for file validation
       return {
           "file_path": file_path,
           "valid": True,
           "issues": [],
           "success": True
       }

Example 3: API Integration Tool

# api_client_tool.py
import requests
from typing import Dict, Any, Optional
from tooluniverse.tool_registry import register_tool

@register_tool('APIClientTool', config={
   "name": "call_api",
   "type": "APIClientTool",
   "description": "Make HTTP API calls with built-in error handling",
   "parameter": {
       "type": "object",
       "properties": {
           "url": {"type": "string", "description": "API endpoint URL"},
           "method": {"type": "string", "enum": ["GET", "POST", "PUT", "DELETE"], "default": "GET"},
           "headers": {"type": "object", "description": "HTTP headers"},
           "data": {"type": "object", "description": "Request body data"},
           "timeout": {"type": "integer", "default": 30, "minimum": 1, "maximum": 300}
       },
       "required": ["url"]
   },
   "settings": {
       "max_retries": 3,
       "default_headers": {
           "User-Agent": "ToolUniverse-APIClient/1.0"
       }
   }
})
class APIClientTool:
   def __init__(self, tool_config: Dict[str, Any]):
       self.tool_config = tool_config
       self.max_retries = tool_config.get("settings", {}).get("max_retries", 3)
       self.default_headers = tool_config.get("settings", {}).get("default_headers", {})

   def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
       url = arguments.get('url')
       method = arguments.get('method', 'GET').upper()
       headers = {**self.default_headers, **arguments.get('headers', {})}
       data = arguments.get('data')
       timeout = arguments.get('timeout', 30)

       for attempt in range(self.max_retries + 1):
           try:
               response = requests.request(
                   method=method,
                   url=url,
                   headers=headers,
                   json=data if data else None,
                   timeout=timeout
               )

               result = {
                   "status_code": response.status_code,
                   "url": url,
                   "method": method,
                   "success": response.status_code < 400,
                   "attempt": attempt + 1
               }

               # Try to parse JSON response
               try:
                   result["data"] = response.json()
               except:
                   result["data"] = response.text

               if response.status_code < 400:
                   return result
               else:
                   result["error"] = f"HTTP {response.status_code}"
                   if attempt == self.max_retries:
                       return result

           except Exception as e:
               if attempt == self.max_retries:
                   return {
                       "error": str(e),
                       "url": url,
                       "method": method,
                       "success": False,
                       "attempt": attempt + 1
                   }

       return {"error": "Max retries exceeded", "success": False}

.. _best-practices:

Best Practices

1. File Naming

• Name your tool files with _tool.py suffix
• Use descriptive names: weather_tool.py, database_tool.py

2. Class Design

class MyTool:
   def __init__(self, tool_config=None):
       """Always accept tool_config parameter"""
       self.tool_config = tool_config or {}
       # Initialize from config
       self.setting_value = tool_config.get("settings", {}).get("key", "default")

   def run(self, arguments):
       """Main entry point - always called 'run'"""
       # Validate inputs
       required_param = arguments.get('required_param')
       if not required_param:
           return {"error": "required_param is missing", "success": False}

       # Process and return results
       return {"result": "success", "success": True}

3. Error Handling

def run(self, arguments):
   try:
       # Your logic here
       result = self.process_data(arguments)
       return {"result": result, "success": True}

   except ValueError as e:
       return {"error": f"Invalid input: {str(e)}", "success": False}

   except Exception as e:
       return {"error": f"Unexpected error: {str(e)}", "success": False}

4. Configuration Management

@register_tool('ConfigurableTool', config={
   "name": "my_tool",
   "type": "ConfigurableTool",
   "description": "A tool that uses configuration",
   "parameter": {...},
   "settings": {
       "api_endpoint": "https://api.example.com",
       "timeout": 30,
       "retries": 3
   }
})
class ConfigurableTool:
   def __init__(self, tool_config=None):
       settings = tool_config.get("settings", {})
       self.api_endpoint = settings.get("api_endpoint")
       self.timeout = settings.get("timeout", 30)
       self.retries = settings.get("retries", 3)

5. Return Format Consistency

Always return dictionaries with consistent structure:

# Success
return {
   "result": actual_result,
   "success": True,
   "metadata": {...}  # optional
}

# Error
return {
   "error": "Description of what went wrong",
   "success": False,
   "details": {...}  # optional error details
}

.. _troubleshooting:

Troubleshooting

Tool Not Found

Problem: Your tool isn't being discovered Solutions:

1. Ensure file ends with _tool.py
2. Check file is in the correct directory
3. Verify the import doesn't fail:

python -c "from your_tool_module import YourTool"

Import Errors

Problem: Module import fails during auto-discovery Solutions:

1. Check all dependencies are installed
2. Ensure relative imports use proper syntax
3. Test import manually:

import sys
sys.path.append('/path/to/tooluniverse/src')
from tooluniverse import YourTool

Configuration Issues

Problem: Tool config not loading properly Solutions:

1. Validate JSON syntax in config
2. Check required fields are present
3. Test config separately:

import json
config = {...}  # your config
print(json.dumps(config, indent=2))  # Should not error

Runtime Errors

Problem: Tool fails during execution Solutions:

1. Add comprehensive error handling
2. Validate all inputs in run() method
3. Return error details:

try:
    result = tool.run(arguments)
except Exception as e:
    return {
        "error": str(e),
        "traceback": traceback.format_exc(),  # For debugging
        "success": False
    }

Testing Your Tool

Create a simple test script:

# test_my_tool.py
from tooluniverse import ToolUniverse

def test_my_tool():
   tu = ToolUniverse()
   tu.load_tools()

   # Test your tool
   result = tu.run_one_function({
       "name": "your_tool_name",
       "arguments": {"test_param": "test_value"}
   })

   print("Result:", result)
   assert result.get("success") is True, f"Tool failed: {result}"
   print("✅ Tool test passed!")

if __name__ == "__main__":
   test_my_tool()

Next Steps

1. Start Simple: Begin with a basic tool using the quick start example
2. Add Complexity: Gradually add more features and configuration
3. Test Thoroughly: Create test cases for different scenarios
4. Document: Add clear descriptions and examples
5. Share: Consider contributing useful tools back to the community

Need Help?

• Check existing tools in the ToolUniverse codebase for examples
• Review the configuration schemas of similar tools
• Test incrementally - start with basic functionality
• Use the troubleshooting section for common issues

The decorator-based system makes adding tools straightforward while maintaining all the power and flexibility of ToolUniverse. Happy tool building! ️