Devtu Create Tool

ToolUniverse Tool Creator

Create new scientific tools following established patterns.

Top 7 Mistakes (90% of Failures)

1. Missing `default_config.py` Entry — tools silently won't load 2. Non-nullable Mutually Exclusive Parameters — validation errors (#1 issue in 2026) 3. Fake test_examples — tests fail, agents get bad examples 4. Single-level Testing — misses registration bugs 5. Skipping `test_new_tools.py` — misses schema/API issues 6. Tool Names > 55 chars — breaks MCP compatibility 7. Raising Exceptions — should return error dicts instead

---

Two-Stage Architecture

Stage 1: Tool Class              Stage 2: Wrappers (Auto-Generated)
@register_tool("MyTool")         MyAPI_list_items()
class MyTool(BaseTool):          MyAPI_search()
    def run(arguments):          MyAPI_get_details()

One class handles multiple operations. JSON defines individual wrappers. Need BOTH.

Three-Step Registration

Step 1: Class registration via @register_tool("MyAPITool")

Step 2 (MOST COMMONLY MISSED): Config registration in default_config.py:

TOOLS_CONFIGS = {
    "my_category": os.path.join(current_dir, "data", "my_category_tools.json"),
}

Step 3: Automatic wrapper generation on tu.load_tools()

---

Implementation Guide

Files to Create

• src/tooluniverse/my_api_tool.py — implementation
• src/tooluniverse/data/my_api_tools.json — tool definitions
• tests/tools/test_my_api_tool.py — tests

Python Tool Class (Multi-Operation Pattern)

from typing import Dict, Any
from tooluniverse.tool import BaseTool
from tooluniverse.tool_utils import register_tool
import requests

@register_tool("MyAPITool")
class MyAPITool(BaseTool):
    BASE_URL = "https://api.example.com/v1"

    def __init__(self, tool_config):
        super().__init__(tool_config)
        self.parameter = tool_config.get("parameter", {})
        self.required = self.parameter.get("required", [])

    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        operation = arguments.get("operation")
        if not operation:
            return {"status": "error", "error": "Missing: operation"}
        if operation == "search":
            return self._search(arguments)
        return {"status": "error", "error": f"Unknown: {operation}"}

    def _search(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        query = arguments.get("query")
        if not query:
            return {"status": "error", "error": "Missing: query"}
        try:
            response = requests.get(
                f"{self.BASE_URL}/search",
                params={"q": query}, timeout=30
            )
            response.raise_for_status()
            data = response.json()
            return {"status": "success", "data": data.get("results", [])}
        except requests.exceptions.Timeout:
            return {"status": "error", "error": "Timeout after 30s"}
        except requests.exceptions.HTTPError as e:
            return {"status": "error", "error": f"HTTP {e.response.status_code}"}
        except Exception as e:
            return {"status": "error", "error": str(e)}

JSON Configuration

[
  {
    "name": "MyAPI_search",
    "class": "MyAPITool",
    "description": "Search items. Returns array of results. Supports Boolean operators. Example: 'protein AND membrane'.",
    "parameter": {
      "type": "object",
      "required": ["operation", "query"],
      "properties": {
        "operation": {"const": "search", "description": "Operation (fixed)"},
        "query": {"type": "string", "description": "Search term"},
        "limit": {"type": ["integer", "null"], "description": "Max results (1-100)"}
      }
    },
    "return_schema": {
      "oneOf": [
        {"type": "object", "properties": {"data": {"type": "array"}}},
        {"type": "object", "properties": {"error": {"type": "string"}}, "required": ["error"]}
      ]
    },
    "test_examples": [{"operation": "search", "query": "protein", "limit": 10}]
  }
]

Critical Requirements

• return_schema MUST have oneOf: success + error schemas
• test_examples MUST use real IDs: NO "TEST", "DUMMY", "PLACEHOLDER"
• Tool name <= 55 chars: {API}_{action}_{target} template
• Description 150-250 chars: what, format, example, notes
• NEVER raise in run(): return {"status": "error", "error": "..."}
• Set timeout on all HTTP requests (30s)
• Standard response: {"status": "success|error", "data": {...}}

---

Parameter Design

Mutually Exclusive Parameters (CRITICAL — #1 issue)

When tool accepts EITHER id OR name, BOTH must be nullable:

{
  "id": {"type": ["integer", "null"], "description": "Numeric ID"},
  "name": {"type": ["string", "null"], "description": "Name (alternative to id)"}
}

Without "null", validation fails when user provides only one parameter.

Common cases: id OR name, gene_id OR gene_symbol, any optional filters.

API Key Configuration

Optional keys (tool works without, better with):

{"optional_api_keys": ["NCBI_API_KEY"]}

self.api_key = os.environ.get("NCBI_API_KEY", "")  # Read from env only

Required keys (tool won't work without):

{"required_api_keys": ["NVIDIA_API_KEY"]}

Rules: Never add api_key as tool parameter for optional keys. Use env vars only.

---

Testing (MANDATORY)

Full guide: references/testing-guide.md

Quick Testing Checklist

1. Level 1 — Direct class test: import class, call run(), check response 2. Level 2 — ToolUniverse test: tu.tools.YourTool_op1(...), check registration 3. Level 3 — Real API test: use real IDs, verify actual responses 4. MANDATORY — Run python scripts/test_new_tools.py your_tool -v → 0 failures

Verification Script

# Check all 3 registration steps
python3 -c "
import sys; sys.path.insert(0, 'src')
from tooluniverse.tool_registry import get_tool_registry
import tooluniverse.your_tool_module
assert 'YourToolClass' in get_tool_registry(), 'Step 1 FAILED'
from tooluniverse.default_config import TOOLS_CONFIGS
assert 'your_category' in TOOLS_CONFIGS, 'Step 2 FAILED'
from tooluniverse import ToolUniverse
tu = ToolUniverse(); tu.load_tools()
assert hasattr(tu.tools, 'YourCategory_op1'), 'Step 3 FAILED'
print('All 3 steps verified!')
"

---

Quick Commands

python3 -m json.tool src/tooluniverse/data/your_tools.json     # Validate JSON
python3 -m py_compile src/tooluniverse/your_tool.py             # Check syntax
grep "your_category" src/tooluniverse/default_config.py         # Verify config
python scripts/test_new_tools.py your_tool -v                   # MANDATORY test

References

• Testing guide: references/testing-guide.md
• Advanced patterns (async, SOAP, pagination): references/advanced-patterns.md
• Implementation guide (full checklist): references/implementation-guide.md
• Tool improvement checklist: references/tool-improvement-checklist.md

Advanced Tool Implementation Patterns

This reference covers advanced patterns and techniques for ToolUniverse tool development.

Offline / Computational Tools

Some tools require no network access at all — they perform purely local calculations using mathematical formulas, lookup tables, or parametric models. These tools are faster, more reliable, and easier to test than API-backed tools.

Minimal Offline Tool Skeleton

# my_calculator_tool.py
import math
from typing import Dict, Any, Optional
from tooluniverse.base_tool import BaseTool
from tooluniverse.tool_registry import register_tool

@register_tool("MyCalculatorTool")
class MyCalculatorTool(BaseTool):
    """
    Brief description.  Runs entirely offline — no network requests.
    """

    def __init__(self, tool_config: Dict[str, Any]):
        super().__init__(tool_config)
        fields = tool_config.get("fields", {})
        self.operation = fields.get("operation", "default_op")

    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        try:
            # operation can come from fields (set in JSON) or runtime arg
            op = arguments.get("operation") or self.operation

            def _get(key: str) -> Optional[float]:
                val = arguments.get(key)
                return float(val) if val is not None else None

            x = _get("x")
            if x is None:
                return {"status": "error", "message": "Missing required parameter: x"}

            result_value = x * 2          # your formula here

            return {
                "status": "success",
                "data": {
                    "operation": op,
                    "result": result_value,
                    "result_formatted": f"{result_value:.4g}",
                },
                "metadata": {
                    "note": "Runs offline — no network request.",
                    "formula": "result = x × 2",
                },
            }
        except ValueError as e:
            return {"status": "error", "message": str(e)}
        except Exception as e:
            return {"status": "error", "message": f"Unexpected error: {str(e)}"}

Key rules for offline tools:

• Return {"status": "error", "message": "..."} — use "message" (not "error") to match the circuit plugin convention.
• Include a "metadata" block that names the formula(s) used and notes "Runs offline".
• Always float(val) when extracting numeric arguments from the dict (they may arrive as strings).
• Use ValueError for domain validation; let the outer except Exception catch everything else.

SI Formatting Helper

Include a local _fmt_si() helper in every offline calculation tool for consistent output:

def _fmt_si(value: float, unit: str) -> str:
    """Format a numeric value with SI prefix and unit label."""
    abs_v = abs(value)
    if abs_v == 0:
        return f"0 {unit}"
    if abs_v >= 1e9:  return f"{value / 1e9:.4g} G{unit}"
    if abs_v >= 1e6:  return f"{value / 1e6:.4g} M{unit}"
    if abs_v >= 1e3:  return f"{value / 1e3:.4g} k{unit}"
    if abs_v >= 1:    return f"{value:.4g} {unit}"
    if abs_v >= 1e-3: return f"{value * 1e3:.4g} m{unit}"
    if abs_v >= 1e-6: return f"{value * 1e6:.4g} µ{unit}"
    if abs_v >= 1e-9: return f"{value * 1e9:.4g} n{unit}"
    return f"{value * 1e12:.4g} p{unit}"

Always include both the raw numeric field and a _formatted string sibling so LLMs can display human-readable values without reformatting:

"data": {
    "delay_ps": 34.0,
    "delay_formatted": "34 ps",        # human-readable sibling
    "frequency_Hz": 4.66e9,
    "frequency_formatted": "4.66 GHz", # human-readable sibling
}

Technology-Node Lookup Tables

Many chip-design quantities are tabulated per process node. Use a Dict[int, ...] keyed by node in nm with a _nearest_node() helper that falls back to the closest entry:

# Empirical FO4 delay values in ps, one per node
_FO4_TABLE: Dict[int, float] = {
    180: 250.0,
    130: 170.0,
    90:  115.0,
    65:   80.0,
    45:   55.0,
    32:   40.0,
    28:   34.0,
    20:   26.0,
    16:   20.0,
    10:   15.0,
    7:    11.0,
}

_SORTED_NODES = sorted(_FO4_TABLE.keys())

def _nearest_node(node_nm: float) -> int:
    """Return the nearest supported technology node for a given nm value."""
    return min(_SORTED_NODES, key=lambda n: abs(n - node_nm))

When the user's requested node is not in the table, include a "warning" field (not an error) in the data dict and continue with the nearest match:

nearest = _nearest_node(node_nm)
fo4_ps  = _FO4_TABLE[nearest]

warning = None
if nearest != int(node_nm):
    warning = (
        f"Node {node_nm} nm not in table; "
        f"using nearest supported node {nearest} nm."
    )

result = {
    "node_nm": nearest,
    "fo4_delay_ps": fo4_ps,
    # ...
}
if warning:
    result["warning"] = warning      # add only when needed

Also expose "supported_nodes_nm": _SORTED_NODES in the response so callers know which values are available without reading the source code.

Warning Fields vs Error Returns

Use these two patterns consistently:

Example — threshold warning pattern:

THRESHOLD_V = 0.100  # 100 mV

v_noise = L * n * di_dt   # computation
exceeds = v_noise > THRESHOLD_V

result = {
    "ground_bounce_V":    v_noise,
    "ground_bounce_mV":   v_noise * 1e3,
    "exceeds_threshold":  exceeds,
}
if exceeds:
    result["warning"] = (
        f"Ground bounce {v_noise * 1e3:.1f} mV exceeds "
        f"{THRESHOLD_V * 1e3:.0f} mV threshold."
    )

Multi-Operation Offline Tool (fields.operation dispatch)

For tools that share the same physics domain but have multiple distinct computations, use the fields.operation dispatch pattern. The JSON config stores the default operation in "fields": { "operation": "op_name" }, and the runtime argument can override it:

# In JSON config (one entry per operation):
{
  "name": "Circuit_fo4_delay",
  "type": "FO4DelayTool",
  "fields": { "operation": "lookup" },    # default for this tool entry
  ...
}

# In __init__:
def __init__(self, tool_config):
    super().__init__(tool_config)
    self.operation = tool_config.get("fields", {}).get("operation", "lookup")

# In run():
op = arguments.get("operation") or self.operation   # runtime arg takes priority
if op == "lookup":
    ...
elif op == "estimate_path":
    ...
else:
    return {"status": "error", "message": f"Unknown operation '{op}'.  Valid: ..."}

This lets you expose a single Python class and multiple JSON entries (one per default operation) — or a single JSON entry that accepts the operation as a runtime argument.

Caching Strategies

Simple LRU Cache

from functools import lru_cache
import json

@register_tool("CachedAPITool")
class CachedAPITool(BaseTool):
    """Tool with response caching."""
    
    @lru_cache(maxsize=128)
    def _cached_request(self, url: str, params_json: str):
        """Cache API responses using LRU cache."""
        params = json.loads(params_json)
        response = requests.get(url, params=params)
        return response.json()
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        url = "https://api.example.com/data"
        
        # Convert params to JSON string for cache key
        params_json = json.dumps(arguments, sort_keys=True)
        
        try:
            data = self._cached_request(url, params_json)
            return {"status": "success", "data": data}
        except Exception as e:
            return {"status": "error", "error": str(e)}

Time-Based Cache

from datetime import datetime, timedelta
from typing import Optional

@register_tool("TimeCachedTool")
class TimeCachedTool(BaseTool):
    """Tool with time-based caching."""
    
    def __init__(self):
        super().__init__()
        self._cache = {}
        self._cache_timeout = timedelta(minutes=15)
    
    def _get_cached(self, key: str) -> Optional[Dict]:
        """Get cached value if not expired."""
        if key in self._cache:
            data, timestamp = self._cache[key]
            if datetime.now() - timestamp < self._cache_timeout:
                return data
            else:
                del self._cache[key]
        return None
    
    def _set_cached(self, key: str, data: Dict):
        """Store value in cache with timestamp."""
        self._cache[key] = (data, datetime.now())
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        cache_key = json.dumps(arguments, sort_keys=True)
        
        # Check cache first
        cached = self._get_cached(cache_key)
        if cached:
            return {
                "status": "success",
                "data": cached,
                "cached": True
            }
        
        # Fetch fresh data
        try:
            response = requests.get(url, params=arguments)
            data = response.json()
            
            # Store in cache
            self._set_cached(cache_key, data)
            
            return {
                "status": "success",
                "data": data,
                "cached": False
            }
        except Exception as e:
            return {"status": "error", "error": str(e)}

Batch Processing

Batch Request Tool

from typing import List

@register_tool("BatchTool")
class BatchTool(BaseTool):
    """Process multiple requests in a single call."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        ids = arguments.get('ids', [])
        batch_size = arguments.get('batch_size', 10)
        
        if not ids:
            return {
                "status": "error",
                "error": "ids parameter is required"
            }
        
        results = []
        errors = []
        
        # Process in batches
        for i in range(0, len(ids), batch_size):
            batch = ids[i:i + batch_size]
            
            try:
                batch_results = self._fetch_batch(batch)
                results.extend(batch_results)
            except Exception as e:
                errors.append({
                    "batch": batch,
                    "error": str(e)
                })
        
        return {
            "status": "success" if not errors else "partial",
            "count": len(results),
            "results": results,
            "errors": errors if errors else None
        }
    
    def _fetch_batch(self, ids: List[str]) -> List[Dict]:
        """Fetch data for a batch of IDs."""
        response = requests.post(
            "https://api.example.com/batch",
            json={"ids": ids},
            timeout=60
        )
        response.raise_for_status()
        return response.json()

Parallel Requests

from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List

@register_tool("ParallelTool")
class ParallelTool(BaseTool):
    """Execute multiple requests in parallel."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        ids = arguments.get('ids', [])
        max_workers = min(arguments.get('max_workers', 5), 10)
        
        results = []
        errors = []
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            # Submit all requests
            future_to_id = {
                executor.submit(self._fetch_single, id_): id_
                for id_ in ids
            }
            
            # Collect results as they complete
            for future in as_completed(future_to_id):
                id_ = future_to_id[future]
                try:
                    result = future.result()
                    results.append(result)
                except Exception as e:
                    errors.append({
                        "id": id_,
                        "error": str(e)
                    })
        
        return {
            "status": "success" if not errors else "partial",
            "count": len(results),
            "results": results,
            "errors": errors if errors else None
        }
    
    def _fetch_single(self, id_: str) -> Dict:
        """Fetch data for a single ID."""
        response = requests.get(
            f"https://api.example.com/items/{id_}",
            timeout=30
        )
        response.raise_for_status()
        return response.json()

Streaming and Pagination

Auto-Pagination Tool

@register_tool("AutoPaginationTool")
class AutoPaginationTool(BaseTool):
    """Automatically fetch all pages."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        query = arguments.get('query')
        max_pages = arguments.get('max_pages', 10)
        
        all_results = []
        page = 1
        
        while page <= max_pages:
            try:
                response = requests.get(
                    "https://api.example.com/search",
                    params={
                        'query': query,
                        'page': page,
                        'page_size': 100
                    },
                    timeout=30
                )
                response.raise_for_status()
                data = response.json()
                
                results = data.get('results', [])
                all_results.extend(results)
                
                # Stop if no more results
                if not data.get('next'):
                    break
                
                page += 1
                
            except Exception as e:
                return {
                    "status": "error",
                    "error": f"Failed on page {page}: {str(e)}",
                    "partial_results": all_results
                }
        
        return {
            "status": "success",
            "count": len(all_results),
            "pages": page,
            "results": all_results
        }

Cursor-Based Pagination

@register_tool("CursorPaginationTool")
class CursorPaginationTool(BaseTool):
    """Handle cursor-based pagination."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        cursor = arguments.get('cursor')
        page_size = arguments.get('page_size', 20)
        
        params = {'page_size': page_size}
        if cursor:
            params['cursor'] = cursor
        
        try:
            response = requests.get(
                "https://api.example.com/items",
                params=params,
                timeout=30
            )
            response.raise_for_status()
            data = response.json()
            
            return {
                "status": "success",
                "count": len(data.get('results', [])),
                "next_cursor": data.get('next_cursor'),
                "has_more": data.get('has_more', False),
                "results": data.get('results', [])
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }

Authentication Patterns

API Key Authentication

import os

@register_tool("APIKeyTool")
class APIKeyTool(BaseTool):
    """Tool with API key authentication."""
    
    def __init__(self):
        super().__init__()
        self.api_key = os.environ.get('API_KEY')
        if not self.api_key:
            raise ValueError("API_KEY environment variable not set")
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        try:
            response = requests.get(
                "https://api.example.com/data",
                headers={"X-API-Key": self.api_key},
                params=arguments,
                timeout=30
            )
            response.raise_for_status()
            
            return {
                "status": "success",
                "data": response.json()
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }

OAuth Token Authentication

@register_tool("OAuthTool")
class OAuthTool(BaseTool):
    """Tool with OAuth token authentication."""
    
    def __init__(self):
        super().__init__()
        self.token = self._get_token()
    
    def _get_token(self) -> str:
        """Get or refresh OAuth token."""
        client_id = os.environ.get('CLIENT_ID')
        client_secret = os.environ.get('CLIENT_SECRET')
        
        response = requests.post(
            "https://api.example.com/oauth/token",
            data={
                'grant_type': 'client_credentials',
                'client_id': client_id,
                'client_secret': client_secret
            }
        )
        response.raise_for_status()
        return response.json()['access_token']
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        try:
            response = requests.get(
                "https://api.example.com/data",
                headers={"Authorization": f"Bearer {self.token}"},
                params=arguments,
                timeout=30
            )
            response.raise_for_status()
            
            return {
                "status": "success",
                "data": response.json()
            }
        except requests.HTTPError as e:
            if e.response.status_code == 401:
                # Token expired, refresh and retry
                self.token = self._get_token()
                return self.run(arguments)
            return {
                "status": "error",
                "error": str(e)
            }

Data Transformation

Field Selection and Projection

@register_tool("ProjectionTool")
class ProjectionTool(BaseTool):
    """Tool with field selection support."""
    
    ALLOWED_FIELDS = {
        'id', 'name', 'description', 'created_date',
        'updated_date', 'status', 'author', 'tags'
    }
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        id_ = arguments.get('id')
        fields = arguments.get('fields', ['id', 'name'])
        
        # Validate fields
        invalid = set(fields) - self.ALLOWED_FIELDS
        if invalid:
            return {
                "status": "error",
                "error": f"Invalid fields: {invalid}",
                "allowed_fields": list(self.ALLOWED_FIELDS)
            }
        
        try:
            # Fetch full data
            response = requests.get(
                f"https://api.example.com/items/{id_}",
                timeout=30
            )
            response.raise_for_status()
            full_data = response.json()
            
            # Project only requested fields
            projected = {
                k: v for k, v in full_data.items()
                if k in fields
            }
            
            return {
                "status": "success",
                "data": projected
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }

Data Filtering

@register_tool("FilterTool")
class FilterTool(BaseTool):
    """Tool with client-side filtering."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        query = arguments.get('query')
        filters = arguments.get('filters', {})
        
        try:
            # Fetch data
            response = requests.get(
                "https://api.example.com/search",
                params={'query': query},
                timeout=30
            )
            response.raise_for_status()
            data = response.json()
            
            # Apply filters
            results = data.get('results', [])
            filtered = self._apply_filters(results, filters)
            
            return {
                "status": "success",
                "count": len(filtered),
                "total": len(results),
                "results": filtered
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }
    
    def _apply_filters(self, results: List[Dict], filters: Dict) -> List[Dict]:
        """Apply filters to results."""
        filtered = results
        
        # Status filter
        if 'status' in filters:
            status = filters['status']
            filtered = [r for r in filtered if r.get('status') == status]
        
        # Date range filter
        if 'date_from' in filters:
            date_from = filters['date_from']
            filtered = [
                r for r in filtered
                if r.get('date', '') >= date_from
            ]
        
        if 'date_to' in filters:
            date_to = filters['date_to']
            filtered = [
                r for r in filtered
                if r.get('date', '') <= date_to
            ]
        
        # Tag filter
        if 'tags' in filters:
            required_tags = set(filters['tags'])
            filtered = [
                r for r in filtered
                if required_tags.issubset(set(r.get('tags', [])))
            ]
        
        return filtered

GraphQL Integration

GraphQL Query Tool

@register_tool("GraphQLTool")
class GraphQLTool(BaseTool):
    """Execute GraphQL queries."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        query = arguments.get('query')
        variables = arguments.get('variables', {})
        
        if not query:
            return {
                "status": "error",
                "error": "query parameter is required"
            }
        
        try:
            response = requests.post(
                "https://api.example.com/graphql",
                json={
                    'query': query,
                    'variables': variables
                },
                timeout=30
            )
            response.raise_for_status()
            result = response.json()
            
            # Check for GraphQL errors
            if 'errors' in result:
                return {
                    "status": "error",
                    "error": "GraphQL query failed",
                    "details": result['errors']
                }
            
            return {
                "status": "success",
                "data": result.get('data')
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }

GraphQL with Predefined Queries

@register_tool("PredefinedGraphQLTool")
class PredefinedGraphQLTool(BaseTool):
    """GraphQL tool with predefined queries."""
    
    QUERIES = {
        'get_drug': """
            query GetDrug($id: ID!) {
                drug(id: $id) {
                    id
                    name
                    description
                    manufacturer
                }
            }
        """,
        'search_drugs': """
            query SearchDrugs($query: String!, $limit: Int) {
                searchDrugs(query: $query, limit: $limit) {
                    id
                    name
                    manufacturer
                }
            }
        """
    }
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        operation = arguments.get('operation')
        variables = arguments.get('variables', {})
        
        if operation not in self.QUERIES:
            return {
                "status": "error",
                "error": f"Unknown operation: {operation}",
                "available_operations": list(self.QUERIES.keys())
            }
        
        query = self.QUERIES[operation]
        
        try:
            response = requests.post(
                "https://api.example.com/graphql",
                json={
                    'query': query,
                    'variables': variables
                },
                timeout=30
            )
            response.raise_for_status()
            result = response.json()
            
            if 'errors' in result:
                return {
                    "status": "error",
                    "error": "GraphQL query failed",
                    "details": result['errors']
                }
            
            return {
                "status": "success",
                "data": result.get('data')
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }

Rate Limiting

Simple Rate Limiter

from time import time, sleep
from collections import deque

@register_tool("RateLimitedTool")
class RateLimitedTool(BaseTool):
    """Tool with rate limiting."""
    
    def __init__(self):
        super().__init__()
        self.requests = deque()
        self.max_requests = 10  # 10 requests
        self.time_window = 60   # per 60 seconds
    
    def _wait_if_needed(self):
        """Wait if rate limit would be exceeded."""
        now = time()
        
        # Remove requests outside time window
        while self.requests and self.requests[0] < now - self.time_window:
            self.requests.popleft()
        
        # Wait if at limit
        if len(self.requests) >= self.max_requests:
            sleep_time = self.requests[0] + self.time_window - now
            if sleep_time > 0:
                sleep(sleep_time)
            self.requests.popleft()
        
        # Record this request
        self.requests.append(now)
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        self._wait_if_needed()
        
        try:
            response = requests.get(
                "https://api.example.com/data",
                params=arguments,
                timeout=30
            )
            response.raise_for_status()
            
            return {
                "status": "success",
                "data": response.json()
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }

Webhook and Async Operations

Async Job Polling Tool

@register_tool("AsyncJobTool")
class AsyncJobTool(BaseTool):
    """Tool that polls for async job completion."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        query = arguments.get('query')
        timeout = arguments.get('timeout', 300)  # 5 minutes default
        poll_interval = arguments.get('poll_interval', 5)  # 5 seconds
        
        try:
            # Start async job
            response = requests.post(
                "https://api.example.com/jobs",
                json={'query': query},
                timeout=30
            )
            response.raise_for_status()
            job_data = response.json()
            job_id = job_data['job_id']
            
            # Poll for completion
            start_time = time()
            while time() - start_time < timeout:
                status_response = requests.get(
                    f"https://api.example.com/jobs/{job_id}",
                    timeout=30
                )
                status_response.raise_for_status()
                status = status_response.json()
                
                if status['state'] == 'completed':
                    return {
                        "status": "success",
                        "job_id": job_id,
                        "data": status['result']
                    }
                elif status['state'] == 'failed':
                    return {
                        "status": "error",
                        "error": "Job failed",
                        "detail": status.get('error')
                    }
                
                sleep(poll_interval)
            
            return {
                "status": "timeout",
                "error": f"Job did not complete within {timeout} seconds",
                "job_id": job_id
            }
            
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }

Multi-Source Aggregation

Aggregation Tool

@register_tool("AggregationTool")
class AggregationTool(BaseTool):
    """Aggregate data from multiple sources."""
    
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        query = arguments.get('query')
        sources = arguments.get('sources', ['source1', 'source2', 'source3'])
        
        results = []
        errors = []
        
        for source in sources:
            try:
                data = self._fetch_from_source(source, query)
                results.extend(data)
            except Exception as e:
                errors.append({
                    "source": source,
                    "error": str(e)
                })
        
        # Deduplicate by ID
        seen_ids = set()
        unique_results = []
        for result in results:
            id_ = result.get('id')
            if id_ not in seen_ids:
                seen_ids.add(id_)
                unique_results.append(result)
        
        return {
            "status": "success" if not errors else "partial",
            "count": len(unique_results),
            "sources_queried": len(sources),
            "sources_failed": len(errors),
            "results": unique_results,
            "errors": errors if errors else None
        }
    
    def _fetch_from_source(self, source: str, query: str) -> List[Dict]:
        """Fetch data from a specific source."""
        url_map = {
            'source1': 'https://api1.example.com/search',
            'source2': 'https://api2.example.com/query',
            'source3': 'https://api3.example.com/find'
        }
        
        url = url_map.get(source)
        if not url:
            raise ValueError(f"Unknown source: {source}")
        
        response = requests.get(url, params={'q': query}, timeout=30)
        response.raise_for_status()
        
        return response.json().get('results', [])

Tool Implementation Guide

Complete guidelines for adding and maintaining tools in ToolUniverse. This is the authoritative reference for tool structure, configuration conventions, and the development checklist.

Guidelines for Adding Tools

Based on docs/expand_tooluniverse/contributing/local_tools.rst and the current codebase structure.

File Structure & Location

• Source Code: src/tooluniverse/xxx_tool.py
• Configuration: src/tooluniverse/data/xxx_tools.json
• Tests: tests/unit/test_xxx_tool.py
• DO NOT manually create files in src/tooluniverse/tools/ — these are auto-generated wrappers.

Implementation Pattern

1. Inheritance: Tool class must inherit from BaseTool. 2. Registration: Use @register_tool decorator with the class name.

from typing import Dict, Any
from .base_tool import BaseTool
from .tool_registry import register_tool

@register_tool("MyNewTool")
class MyNewTool(BaseTool):
    """My new tool description."""
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        return {"result": "success"}

3. Configuration: Use external JSON file (do NOT embed large configs in the decorator).

[
  {
    "name": "my_new_tool",
    "type": "MyNewTool",
    "description": "Convert text to uppercase",
    "parameter": {
      "type": "object",
      "properties": {
        "input": {
          "type": "string",
          "description": "Text to convert"
        }
      },
      "required": ["input"]
    },
    "return_schema": {
      "type": "object",
      "properties": {
        "result": {
          "type": "string",
          "description": "The converted text"
        }
      }
    },
    "test_examples": [
      {
        "input": "hello"
      }
    ]
  }
]

JSON Configuration Conventions

• The "type" field must match the Python class name registered with @register_tool
• Include return_schema to define tool output structure
• Put example inputs in test_examples (and optionally short examples in description)
• Avoid adding JSON Schema examples blocks inside parameter/return_schema — they bloat configs and drift from reality
• For large allow-lists, prefer listing values in the tool description and enforcing in Python at runtime (instead of a huge schema enum)

Auto-Discovery

Modern ToolUniverse uses automated discovery. You generally do not need to modify src/tooluniverse/__init__.py if you place your file correctly in src/tooluniverse/.

Tool Naming Guidelines

• Recommended: ≤ 55 characters (fits in MCP with mcp__tu__ prefix)
• Maximum: 64 characters (MCP hard limit without prefix)
• Automatic shortening: Long names are automatically shortened for MCP exposure
• Best practice: FDA_get_drug_info not FDA_get_detailed_information_about_drug

If your tool name exceeds 55 characters, it will be automatically shortened when exposed via MCP. See docs/mcp_name_shortening.md for details.

Development Checklist

1. [ ] Create src/tooluniverse/xxx_tool.py with @register_tool 2. [ ] Create src/tooluniverse/data/xxx_tools.json including return_schema 3. [ ] Ensure tool names are ≤ 55 characters 4. [ ] Implement run(arguments) method 5. [ ] Implement validate_parameters (optional but recommended) 6. [ ] Write unit tests in tests/unit/ 7. [ ] Verify tool load with tu.load_tools() 8. [ ] Run python scripts/check_tool_name_lengths.py --test-shortening to validate 9. [ ] Run python scripts/test_new_tools.py your_tool -v (MANDATORY)

---

Tool Improvement and Maintenance Checklist

Systematic approach to improving and maintaining existing tools.

Phase 1: Initial Assessment

Step 1.1: Identify Tool Files

• [ ] Locate tool class file: src/tooluniverse/{category}_tool.py
• [ ] Locate JSON config file: src/tooluniverse/data/{category}_tools.json
• [ ] List all tool function files: src/tooluniverse/tools/{category}_*.py (auto-generated wrappers)
• [ ] Check default_config.py for category registration
• [ ] Check tools/__init__.py for imports (auto-generated)

Step 1.2: Verify Basic Structure

• [ ] Tool class registration exists (@register_tool)
• [ ] Class name matches JSON config "type" field
• [ ] JSON file is valid
• [ ] Tool loads without errors
• [ ] Python syntax is valid

Phase 2: Functionality Testing

Step 2.1: Test Tool Execution

• [ ] Load tools and test each tool with sample arguments
• [ ] Verify results contain data (not empty)
• [ ] Check response structure matches return_schema
• [ ] Test error handling with invalid inputs

Step 2.2: Test API Endpoints Directly

• [ ] Test REST/GraphQL endpoints respond correctly
• [ ] Verify status codes are 200 OK (not 404/502/503)
• [ ] Check response format matches tool expectations

Phase 3: Description Improvement

Step 3.1: Review Tool Descriptions

• [ ] Description includes: purpose, input, output, use cases
• [ ] Description is clear to users unfamiliar with API
• [ ] Add usage guidance and example inputs (prefer test_examples; optionally include short examples in description, not in JSON Schema)

Step 3.2: Review Parameter Descriptions

For each parameter:

• [ ] Has clear description (include example values in the description text if helpful)
• [ ] Has default value if optional
• [ ] Has constraints (min/max/enum) if applicable
• [ ] Type is correct

Step 3.3: Review Return Schema

• [ ] return_schema field exists
• [ ] Schema matches actual tool output (test live responses; do not rely on docs alone)
• [ ] Schema is meaningful (avoid data: { additionalProperties: true } as the only definition)
• [ ] Model the common response shapes explicitly:
• [ ] Paginated lists: count, next, previous, results[]
• [ ] Detail objects: required identifiers + key domain fields
• [ ] For nested structures, type the important subfields but allow extra fields with additionalProperties: true
• [ ] Handle real-world type variability (e.g., values sometimes returned as string vs number): use union types like ["string","number","null"]
• [ ] If the tool wraps upstream responses (e.g., adds status, url, error), ensure return_schema reflects the wrapper shape consistently

Phase 4: Error Handling Improvement

Step 4.1: Review Current Error Handling

• [ ] Test error messages with invalid inputs
• [ ] Test HTTP error handling (404, 502, 503)
• [ ] Verify try/except blocks exist
• [ ] Errors return dict with "error" key

Step 4.2: Improve Error Messages

• [ ] Error messages are specific (not generic)
• [ ] Errors suggest actionable solutions
• [ ] Errors include context (status_code, endpoint)
• [ ] Errors are user-friendly
• [ ] If introducing a standardized error envelope, apply it consistently within that tool family

Step 4.3: Add Retry Logic (if needed)

• [ ] Identify transient failures (ConnectionError, Timeout)
• [ ] Implement retry with exponential backoff
• [ ] Set max retries (typically 2-3)
• [ ] Handle final failure appropriately
• [ ] Prefer using a shared retry helper if one exists in the codebase

Phase 5: Finding Missing Tools

Step 5.1: Research API Capabilities

• [ ] Read API Docs: Check official documentation for all endpoints/operations
• [ ] GraphQL Introspection: Use schema introspection to find all queries
• [ ] Test Endpoints: Try different endpoint patterns
• [ ] Check Related Packages: Look at R/Bioconductor or Python packages
• [ ] Web Search: Search for "{API_NAME} API documentation"

Step 5.2: Create Gap Analysis Matrix

• [ ] List current tools from JSON config
• [ ] List all API capabilities
• [ ] Create comparison table (implemented vs available)
• [ ] Prioritize missing tools (HIGH/MEDIUM/LOW)
• [ ] Document findings

Step 5.3: Identify Subset Extraction Opportunities

• [ ] Check Data Size: If full response is large/complex
• [ ] Identify Subsets: Common fields users need (diseases, pathways, etc.)
• [ ] Add Subset Tools: Create tools that extract specific data types
• [ ] Implement Method: Create _extract_subset() helper if needed
• [ ] Add field selection / projection when supported upstream
• [ ] If projection fields are many: document the allowed values in description and enforce in code (avoid giant schema enums)

Phase 6: Fix Common Issues

Issue 6.1: Tool Class Name Mismatch

• Check: Python class name matches @register_tool("ClassName")
• Check: JSON config "type" field matches class name exactly
• Fix: Ensure exact match (case-sensitive)

Issue 6.2: Response Format Mismatch

• Check: Test API response format directly (list vs dict)
• Fix: Check API response format and convert if needed

Issue 6.3: Endpoint URL Issues

• Check: Test endpoint directly, verify URL pattern in API documentation
• Fix: Verify URL building logic and placeholder replacement

Issue 6.4: Missing Error Handling

• Check: Test with invalid inputs and network failures
• Fix: Add try/except blocks with specific error handling

Phase 7: Final Verification

Step 7.1: Comprehensive Testing

• [ ] Test all tools with valid inputs
• [ ] Test error cases with invalid inputs
• [ ] Test edge cases (empty results, null values)
• [ ] Verify results contain meaningful data
• [ ] Check performance is reasonable

Step 7.2: Validation Checks

• [ ] JSON files are valid
• [ ] Python syntax is valid
• [ ] No linting errors
• [ ] All tools load without errors
• [ ] Tool functions imported in tools/__init__.py (auto-generated, verify they exist)
• [ ] Category registered in default_config.py

python3 -m json.tool src/tooluniverse/data/{category}_tools.json
python3 -m py_compile src/tooluniverse/{category}_tool.py

Step 7.3: Documentation

• [ ] Tool descriptions are clear and complete
• [ ] Parameter descriptions include examples
• [ ] Return schemas match actual output
• [ ] Create example script in examples/
• [ ] Document findings and fixes

Guidance for Large API Expansions

When covering many endpoints in one category:

• [ ] Use a generic REST tool + JSON configs to cover multiple endpoints
• [ ] Verify real API behavior with live requests (prefer working patterns over docs)
• [ ] Make return_schema match the tool's wrapper and validate upstream payload structure at a useful depth
• [ ] Use real IDs from search/list endpoints in test_examples
• [ ] Remove tools for endpoints that have no working or replacement API
• [ ] Design "research-first": include discovery tools (search/list), detail tools (get by ID), and version/release tools for reproducibility
• [ ] Keep schemas and parameter surfaces LLM-friendly: avoid enormous enums; keep descriptions explicit; enforce strict validation in code

---

Quick Reference: Common Commands

Validation

python3 -m json.tool src/tooluniverse/data/{category}_tools.json
python3 -m py_compile src/tooluniverse/{category}_tool.py

Testing

python3 examples/{category}_tools_example.py
python scripts/test_new_tools.py {tool_name} -v

Finding Tools

ls src/tooluniverse/tools/{category}_*.py
grep -c "\"name\":" src/tooluniverse/data/{category}_tools.json
grep "@register_tool" src/tooluniverse/{category}_tool.py

Quick Reference Guide

Fast lookup for common ToolUniverse tool development tasks.

File Locations

src/tooluniverse/{category}_tool.py          # Tool class
src/tooluniverse/data/{category}_tools.json  # Configuration
tests/unit/test_{category}_tool.py           # Unit tests
examples/{category}_tools_example.py         # Example usage
src/tooluniverse/tools/{category}_*.py       # Auto-generated (don't edit)

Basic Tool Class

from typing import Dict, Any
from .base_tool import BaseTool
from .tool_registry import register_tool

@register_tool("ToolName")
class ToolName(BaseTool):
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        try:
            # Your logic here
            return {"status": "success", "data": result}
        except Exception as e:
            return {"status": "error", "error": str(e)}

Minimal JSON Config

{
  "name": "tool_name",
  "type": "ToolClassName",
  "description": "What it does, inputs, outputs, use cases",
  "parameter": {
    "type": "object",
    "properties": {
      "param": {"type": "string", "description": "Parameter description"}
    },
    "required": ["param"]
  },
  "return_schema": {
    "type": "object",
    "properties": {
      "status": {"type": "string"},
      "data": {"type": "object", "additionalProperties": true}
    }
  },
  "test_examples": [{"param": "value"}]
}

Common Return Patterns

Simple Success/Error

# Success
return {"status": "success", "data": result}

# Error
return {"status": "error", "error": "Error message"}

Paginated Results

return {
    "status": "success",
    "count": len(results),
    "total": total_count,
    "next": next_url,
    "previous": prev_url,
    "results": results
}

Detail Object

return {
    "status": "success",
    "data": {
        "id": "123",
        "name": "Item name",
        "description": "Details...",
        # ... other fields
    }
}

Error with Details

return {
    "status": "error",
    "error": "Request failed",
    "detail": error_details,
    "suggestion": "Try this instead",
    "url": request_url,
    "status_code": 404
}

HTTP Requests

Basic GET Request

import requests

response = requests.get(
    "https://api.example.com/endpoint",
    params={"param": "value"},
    timeout=30
)
response.raise_for_status()
data = response.json()

GET with Headers

response = requests.get(
    url,
    params=params,
    headers={
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    },
    timeout=30
)

POST Request

response = requests.post(
    url,
    json={"key": "value"},
    headers=headers,
    timeout=30
)

With Retry Logic

import time

for attempt in range(3):
    try:
        response = requests.get(url, params=params, timeout=30)
        response.raise_for_status()
        return response
    except (requests.ConnectionError, requests.Timeout):
        if attempt == 2:
            raise
        time.sleep(2 ** attempt)

Error Handling

Comprehensive Try-Except

try:
    response = requests.get(url, params=params, timeout=30)
    response.raise_for_status()
    return {"status": "success", "data": response.json()}

except requests.Timeout:
    return {
        "status": "error",
        "error": "Request timed out",
        "suggestion": "Try again or use more specific query"
    }

except requests.ConnectionError as e:
    return {
        "status": "error",
        "error": "Failed to connect to API",
        "detail": str(e)
    }

except requests.HTTPError as e:
    return {
        "status": "error",
        "error": f"API error: {e.response.status_code}",
        "detail": e.response.text,
        "url": e.response.url
    }

except Exception as e:
    return {
        "status": "error",
        "error": f"{type(e).__name__}: {str(e)}"
    }

Validation

Parameter Validation

def validate_parameters(self, arguments: Dict[str, Any]) -> None:
    param = arguments.get('param', '')
    
    if not param:
        raise ValueError("param cannot be empty")
    
    if len(param) < 2:
        raise ValueError("param must be at least 2 characters")
    
    if not param.replace(' ', '').isalnum():
        raise ValueError("param contains invalid characters")

In-Method Validation

def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
    param = arguments.get('param')
    
    if not param:
        return {
            "status": "error",
            "error": "Missing required parameter: param"
        }
    
    # Continue with logic...

Common Patterns

Search Tool

def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
    query = arguments.get('query')
    limit = arguments.get('limit', 20)
    
    response = requests.get(
        f"{self.base_url}/search",
        params={'q': query, 'limit': limit}
    )
    data = response.json()
    
    return {
        "status": "success",
        "count": len(data['results']),
        "results": data['results']
    }

Detail Tool

def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
    id_ = arguments.get('id')
    
    response = requests.get(f"{self.base_url}/items/{id_}")
    data = response.json()
    
    return {
        "status": "success",
        "data": data
    }

List Tool with Pagination

def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
    page = arguments.get('page', 1)
    page_size = arguments.get('page_size', 20)
    
    response = requests.get(
        f"{self.base_url}/items",
        params={'page': page, 'page_size': page_size}
    )
    data = response.json()
    
    return {
        "status": "success",
        "count": len(data['results']),
        "total": data['total'],
        "next": data.get('next'),
        "previous": data.get('previous'),
        "results": data['results']
    }

Testing Commands

# Validate JSON syntax
python3 -m json.tool src/tooluniverse/data/{category}_tools.json

# Check Python syntax
python3 -m py_compile src/tooluniverse/{category}_tool.py

# Run tests
pytest tests/unit/test_{category}_tool.py -v

# Check tool loads
python3 -c "from tooluniverse import ToolUniverse; tu = ToolUniverse(); print('Loaded:', len(tu.list_tools()), 'tools')"

# Check tool name lengths
python scripts/check_tool_name_lengths.py --test-shortening

# List auto-generated wrappers
ls src/tooluniverse/tools/{category}_*.py

Quick Test Script

from tooluniverse import ToolUniverse

tu = ToolUniverse()

# Test tool
result = tu.run_tool("tool_name", {"param": "value"})
print(result)

# List all tools
print(f"Total tools: {len(tu.list_tools())}")

JSON Schema Types

{
  "type": "string",           // Text
  "type": "integer",          // Whole numbers
  "type": "number",           // Decimals
  "type": "boolean",          // true/false
  "type": "array",            // Lists
  "type": "object",           // Dictionaries
  "type": ["string", "null"], // Union types
  "type": "object",
  "additionalProperties": true  // Allow extra fields
}

Common Schema Patterns

String Parameter

{
  "param_name": {
    "type": "string",
    "description": "Description with examples: 'example1', 'example2'"
  }
}

Integer with Constraints

{
  "limit": {
    "type": "integer",
    "description": "Max results. Range: 1-100. Default: 20",
    "default": 20,
    "minimum": 1,
    "maximum": 100
  }
}

Optional Boolean

{
  "include_details": {
    "type": "boolean",
    "description": "Include detailed information. Default: false",
    "default": false
  }
}

Enum

{
  "sort_by": {
    "type": "string",
    "description": "Sort order. Options: 'relevance', 'date', 'name'",
    "enum": ["relevance", "date", "name"],
    "default": "relevance"
  }
}

Array of Strings

{
  "tags": {
    "type": "array",
    "description": "List of tags to filter by",
    "items": {
      "type": "string"
    }
  }
}

Environment Variables

import os

# Get with default
api_key = os.environ.get('API_KEY', 'default_key')

# Get required
api_key = os.environ['API_KEY']  # Raises KeyError if missing

# Check existence
if 'API_KEY' in os.environ:
    api_key = os.environ['API_KEY']

URL Building

# Simple concatenation
url = f"{base_url}/endpoint"

# With path parameter
url = f"{base_url}/items/{item_id}"

# With multiple segments
url = f"{base_url}/api/v1/items/{item_id}/details"

# Build with urllib
from urllib.parse import urljoin
url = urljoin(base_url, f"/items/{item_id}")

Common Mistakes to Avoid

❌ Don't

# Don't edit auto-generated files
src/tooluniverse/tools/category_tool_name.py

# Don't use bare except
except:
    pass

# Don't ignore errors
result = some_function()  # No error checking

# Don't hardcode URLs
response = requests.get("http://example.com/api")

# Don't forget timeouts
response = requests.get(url)  # No timeout

# Don't create tools longer than 55 chars
"very_long_tool_name_that_exceeds_the_mcp_compatibility_limit"

✅ Do

# Do create tool class files
src/tooluniverse/category_tool.py

# Do catch specific exceptions
except ValueError as e:
    return {"error": str(e)}

# Do check errors
if not result:
    return {"error": "Failed"}

# Do use configurable URLs
self.base_url = "https://api.example.com"

# Do set timeouts
response = requests.get(url, timeout=30)

# Do keep names concise
"get_drug_info"  # 13 chars

Debugging Tips

# Print arguments
print(f"Arguments: {arguments}")

# Print response
print(f"Status: {response.status_code}")
print(f"Data: {response.json()}")

# Check data type
print(f"Type: {type(data)}")

# Pretty print JSON
import json
print(json.dumps(data, indent=2))

# Log errors
import logging
logging.error(f"Failed: {str(e)}")

Return Schema Anti-Patterns

❌ Bad (Too Vague)

{
  "return_schema": {
    "type": "object",
    "properties": {
      "data": {"type": "object", "additionalProperties": true}
    }
  }
}

✅ Good (Specific)

{
  "return_schema": {
    "type": "object",
    "properties": {
      "status": {"type": "string"},
      "count": {"type": "integer"},
      "results": {
        "type": "array",
        "items": {
          "type": "object",
          "properties": {
            "id": {"type": "string"},
            "name": {"type": "string"}
          },
          "additionalProperties": true
        }
      }
    }
  }
}

Time Savers

# Create all files at once
mkdir -p src/tooluniverse/data tests/unit examples
touch src/tooluniverse/my_tool.py
touch src/tooluniverse/data/my_tools.json
touch tests/unit/test_my_tool.py

# Validate everything
python3 -m json.tool src/tooluniverse/data/*.json
python3 -m py_compile src/tooluniverse/*_tool.py

# Count tools
grep -c '"name":' src/tooluniverse/data/*_tools.json

# Find tool registration
grep -r "@register_tool" src/tooluniverse/

# Check for long names
python scripts/check_tool_name_lengths.py

Testing Guide for New Tools

Three-Level Testing (MANDATORY)

Level 1: Direct Class Testing

import json
from tooluniverse.your_tool_module import YourToolClass

with open("src/tooluniverse/data/your_tools.json") as f:
    tools = json.load(f)
    config = next(t for t in tools if t["name"] == "YourTool_operation1")

tool = YourToolClass(config)
result = tool.run({"operation": "operation1", "param": "value"})
assert result["status"] == "success"

Level 2: ToolUniverse Interface Testing

from tooluniverse import ToolUniverse
tu = ToolUniverse()
tu.load_tools()

# Verify registration
assert hasattr(tu.tools, 'YourTool_operation1')

# Test execution
result = tu.tools.YourTool_operation1(operation="operation1", param="value")
assert result["status"] == "success"

# Test error handling
result = tu.tools.YourTool_operation1(operation="operation1")  # Missing required param
assert result["status"] == "error"

Level 3: Real API Testing

result = tu.tools.YourTool_operation1(operation="operation1", param="real_value_from_docs")
if result["status"] == "success":
    assert "data" in result
else:
    print(f"API error (may be down): {result['error']}")

Mandatory: test_new_tools.py

python scripts/test_new_tools.py your_tool_name -v
python scripts/test_new_tools.py your_tool_name --fail-fast

Validates: execution succeeds, response matches return_schema, 404s indicate bad test_examples.

Systematic Testing for Multiple Tools

1. Sample test 1-2 tools per API to catch common issues 2. Identify patterns: group errors by type (param validation, API errors, schema errors) 3. Fix systematically: fix all tools with same issue together 4. Verify all: python scripts/test_new_tools.py MyAPI -v 5. Verify parameters: print param names from JSON, don't assume

Common Failures

Verification Script

import sys
sys.path.insert(0, 'src')

# Step 1: Class registered
from tooluniverse.tool_registry import get_tool_registry
import tooluniverse.your_tool_module
assert "YourToolClass" in get_tool_registry()

# Step 2: Config registered
from tooluniverse.default_config import TOOLS_CONFIGS
assert "your_category" in TOOLS_CONFIGS

# Step 3: Wrappers generated
from tooluniverse import ToolUniverse
tu = ToolUniverse()
tu.load_tools()
assert hasattr(tu.tools, 'YourCategory_operation1')

print("All 3 registration steps verified!")

Tool Improvement and Maintenance Checklist

This reference provides a systematic approach to improving and maintaining existing ToolUniverse tools.

Phase 1: Initial Assessment

Step 1.1: Identify Tool Files

Locate all relevant files for the tool category:

# Tool class file
src/tooluniverse/{category}_tool.py

# JSON configuration
src/tooluniverse/data/{category}_tools.json

# Auto-generated wrappers (DO NOT EDIT)
src/tooluniverse/tools/{category}_*.py

# Check registration
grep "@register_tool" src/tooluniverse/{category}_tool.py

# Check imports (auto-generated)
grep "{category}" src/tooluniverse/tools/__init__.py

Step 1.2: Verify Basic Structure

• [ ] Tool class registration exists (@register_tool)
• [ ] Class name matches JSON config "type" field
• [ ] JSON file is valid: python3 -m json.tool {file}.json
• [ ] Tool loads without errors: tu.load_tools()
• [ ] Python syntax is valid: python3 -m py_compile {file}.py

Phase 2: Functionality Testing

Step 2.1: Test Tool Execution

from tooluniverse import ToolUniverse

tu = ToolUniverse()

# List tools
tools = [t for t in tu.list_tools() if t.startswith('category_')]

# Test each tool
for tool_name in tools:
    print(f"\nTesting {tool_name}...")
    
    # Get test example from JSON
    result = tu.run_tool(tool_name, test_arguments)
    
    # Verify results
    assert result is not None, "Result is None"
    assert result != {}, "Result is empty"
    assert "error" not in result or result.get("status") != "error"
    
    print(f"✓ {tool_name} passed")

Checklist:

• [ ] Tool executes without errors
• [ ] Results contain data (not empty)
• [ ] Response structure matches return_schema
• [ ] Error handling works with invalid inputs

Step 2.2: Test API Endpoints Directly

import requests

# Test endpoint directly
url = "https://api.example.com/endpoint"
response = requests.get(url, params={"id": "test123"})

print(f"Status Code: {response.status_code}")
print(f"Response: {response.json()}")

Checklist:

• [ ] REST/GraphQL endpoints respond correctly
• [ ] Status codes are 200 OK (not 404/502/503)
• [ ] Response format matches tool expectations
• [ ] Authentication works if required

Phase 3: Description Improvement

Step 3.1: Review Tool Descriptions

Good description template:

{
  "description": "[ACTION] [WHAT] from [SOURCE]. [INPUT DETAILS]. Returns [OUTPUT DETAILS] including [KEY FIELDS]. Use for: [USE CASE 1], [USE CASE 2], [USE CASE 3]. Example: [BRIEF EXAMPLE]."
}

Example:

{
  "description": "Search for clinical trials by condition or intervention. Accepts disease names, drug names, or medical terms. Returns trial details including status, phase, locations, and eligibility criteria. Use for: drug development research, patient recruitment, competitive analysis. Example: Search 'diabetes' to find all diabetes-related trials."
}

Checklist:

• [ ] Description includes purpose
• [ ] Description explains inputs
• [ ] Description explains outputs
• [ ] Description lists use cases
• [ ] Description includes brief example
• [ ] Description is clear to users unfamiliar with API

Step 3.2: Review Parameter Descriptions

For each parameter:

{
  "parameter": {
    "properties": {
      "query": {
        "type": "string",
        "description": "Search term for drug name or condition. Examples: 'aspirin', 'hypertension', 'cancer therapy'. Case-insensitive, supports partial matches."
      },
      "max_results": {
        "type": "integer",
        "description": "Maximum number of results to return. Default: 20. Range: 1-100.",
        "default": 20
      },
      "sort_by": {
        "type": "string",
        "description": "Sort order for results. Options: 'relevance', 'date', 'name'. Default: 'relevance'.",
        "default": "relevance"
      }
    }
  }
}

Checklist:

• [ ] Has clear description with example values
• [ ] Has default value if optional
• [ ] Has constraints (min/max/enum) if applicable
• [ ] Type is correct (string, integer, boolean, array, object)
• [ ] Enums list all valid options (or describe in text if many)

Step 3.3: Review Return Schema

Anti-pattern (avoid this):

{
  "return_schema": {
    "type": "object",
    "properties": {
      "data": {
        "type": "object",
        "additionalProperties": true
      }
    }
  }
}

Good pattern (do this):

{
  "return_schema": {
    "type": "object",
    "properties": {
      "status": {
        "type": "string",
        "description": "Request status: 'success' or 'error'"
      },
      "count": {
        "type": "integer",
        "description": "Total number of results"
      },
      "next": {
        "type": ["string", "null"],
        "description": "URL for next page of results, null if last page"
      },
      "previous": {
        "type": ["string", "null"],
        "description": "URL for previous page of results, null if first page"
      },
      "results": {
        "type": "array",
        "description": "Array of result objects",
        "items": {
          "type": "object",
          "properties": {
            "id": {
              "type": "string",
              "description": "Unique identifier"
            },
            "name": {
              "type": "string",
              "description": "Display name"
            },
            "description": {
              "type": "string",
              "description": "Detailed description"
            }
          },
          "additionalProperties": true
        }
      }
    }
  }
}

Checklist:

• [ ] return_schema field exists
• [ ] Schema matches actual tool output (test live responses)
• [ ] Schema is meaningful (not just additionalProperties: true)
• [ ] Common patterns modeled explicitly:
• [ ] Paginated lists: count, next, previous, results[]
• [ ] Detail objects: required identifiers + key domain fields
• [ ] Nested structures type important subfields
• [ ] Use additionalProperties: true for flexibility
• [ ] Handle type variability with unions: ["string", "number", "null"]
• [ ] Wrapper fields included if tool adds them (status, url, error)

Phase 4: Error Handling Improvement

Step 4.1: Review Current Error Handling

Test error scenarios:

# Test missing required parameter
result = tu.run_tool("tool_name", {})

# Test invalid parameter value
result = tu.run_tool("tool_name", {"id": "invalid"})

# Test network error (mock or use bad URL)

Checklist:

• [ ] Error messages are specific (not generic)
• [ ] Try/except blocks exist around risky operations
• [ ] Errors return dict with "error" key
• [ ] HTTP errors handled (404, 502, 503)

Step 4.2: Improve Error Messages

Bad error message:

return {"error": "Invalid input"}

Good error message:

return {
    "status": "error",
    "error": "Invalid parameter: 'drug_name' must be a non-empty string",
    "detail": f"Received: {drug_name}",
    "suggestion": "Provide a valid drug name, e.g., 'aspirin'"
}

Error message checklist:

• [ ] Specific: States exactly what went wrong
• [ ] Actionable: Suggests how to fix the problem
• [ ] Context: Includes relevant details (status_code, endpoint, values)
• [ ] User-friendly: Written for end users, not developers
• [ ] Consistent: Uses same error envelope across tool family

Step 4.3: Add Retry Logic

When to add retries:

• Network connection errors
• Timeout errors
• 502/503 service unavailable errors
• Rate limit errors (with longer backoff)

When NOT to retry:

• 400 Bad Request (client error)
• 401 Unauthorized (auth error)
• 404 Not Found (resource doesn't exist)
• Validation errors

Implementation:

import time
import requests

def _request_with_retry(self, url: str, params: dict, max_retries: int = 3):
    """Make request with exponential backoff."""
    for attempt in range(max_retries):
        try:
            response = requests.get(url, params=params, timeout=30)
            response.raise_for_status()
            return response
            
        except (requests.ConnectionError, requests.Timeout) as e:
            if attempt == max_retries - 1:
                raise
            wait_time = 2 ** attempt  # 1s, 2s, 4s
            time.sleep(wait_time)
            
        except requests.HTTPError as e:
            # Don't retry 4xx client errors
            if 400 <= e.response.status_code < 500:
                raise
            # Retry 5xx server errors
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)

Checklist:

• [ ] Identify transient failures (ConnectionError, Timeout, 5xx)
• [ ] Implement exponential backoff (1s, 2s, 4s, ...)
• [ ] Set max retries (typically 2-3)
• [ ] Don't retry 4xx client errors
• [ ] Handle final failure appropriately
• [ ] Use shared retry helper if available in codebase

Phase 5: Finding Missing Tools

Step 5.1: Research API Capabilities

Methods to discover all API capabilities:

1. Read Official API Documentation

# Search for API docs
https://example.com/api/docs
https://example.com/developers
https://docs.example.com

2. GraphQL Introspection

import requests

query = """
query {
  __schema {
    queryType {
      fields {
        name
        description
      }
    }
  }
}
"""

response = requests.post(
    'https://api.example.com/graphql',
    json={'query': query}
)
print(response.json())

3. Test Endpoint Patterns

# Try common patterns
endpoints = [
    "/search",
    "/list",
    "/get/{id}",
    "/details/{id}",
    "/query",
    "/find"
]

for endpoint in endpoints:
    url = f"https://api.example.com{endpoint}"
    response = requests.get(url)
    print(f"{endpoint}: {response.status_code}")

4. Check Related Packages

# R/Bioconductor packages
https://bioconductor.org/packages/

# Python packages
https://pypi.org/search/?q=example

# Look at package source code for API calls

5. Web Search

"{API_NAME} API documentation"
"{API_NAME} API endpoints"
"{API_NAME} API reference"
site:github.com "{API_NAME} API"

Step 5.2: Create Gap Analysis Matrix

Create a comparison table:

Prioritization criteria:

• HIGH: Common use case, core functionality, fills major gap
• MEDIUM: Useful but not critical, nice-to-have
• LOW: Niche use case, edge functionality

Step 5.3: Identify Subset Extraction Opportunities

When full responses are large/complex, create focused subset tools:

Example: Large drug object with many fields

# Main tool returns everything
def get_drug_details(drug_id):
    return {
        "id": "...",
        "name": "...",
        "manufacturer": "...",
        "approval_date": "...",
        "ingredients": [...],
        "adverse_events": [...],
        "clinical_trials": [...],
        # ... 50+ more fields
    }

# Subset tools extract specific data
def get_drug_adverse_events(drug_id):
    """Extract only adverse events from drug data."""
    full_data = get_drug_details(drug_id)
    return {
        "drug_id": drug_id,
        "drug_name": full_data["name"],
        "adverse_events": full_data["adverse_events"]
    }

def get_drug_ingredients(drug_id):
    """Extract only ingredients from drug data."""
    full_data = get_drug_details(drug_id)
    return {
        "drug_id": drug_id,
        "drug_name": full_data["name"],
        "ingredients": full_data["ingredients"]
    }

Checklist:

• [ ] Check if full response is large (>1000 lines) or complex (>5 nested levels)
• [ ] Identify common subsets users need (diseases, pathways, events, etc.)
• [ ] Create focused tools that extract specific data types
• [ ] Implement helper method: _extract_subset()
• [ ] Add field selection parameter when supported upstream
• [ ] Document allowed fields in description (not schema enum if many)

Phase 6: Fix Common Issues

Issue 6.1: Tool Class Name Mismatch

Symptoms:

• Tool doesn't load
• "Tool not found" errors
• Registration errors

Check:

# Check Python class name
grep "class.*BaseTool" src/tooluniverse/{category}_tool.py

# Check registration
grep "@register_tool" src/tooluniverse/{category}_tool.py

# Check JSON type
grep '"type":' src/tooluniverse/data/{category}_tools.json

Fix: Ensure exact match (case-sensitive):

@register_tool("DrugSearchTool")  # Must match exactly
class DrugSearchTool(BaseTool):
    ...

{
  "type": "DrugSearchTool"  // Must match exactly
}

Issue 6.2: Response Format Mismatch

Symptoms:

• 'list' object has no attribute 'get'
• TypeError: 'NoneType' object is not subscriptable
• Unexpected data structure errors

Check:

# Test API response directly
import requests
response = requests.get(url)
data = response.json()
print(f"Type: {type(data)}")
print(f"Data: {data}")

Fix: Convert as needed:

def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
    response = requests.get(url)
    data = response.json()
    
    # API returns list, but tool expects dict
    if isinstance(data, list):
        return {
            "status": "success",
            "count": len(data),
            "results": data
        }
    
    # API returns None
    if data is None:
        return {
            "status": "error",
            "error": "No data returned from API"
        }
    
    # API returns dict as expected
    return {
        "status": "success",
        "data": data
    }

Issue 6.3: Endpoint URL Issues

Symptoms:

• 404 Not Found errors
• "Endpoint does not exist" errors

Check:

# Test endpoint directly
import requests

url = "https://api.example.com/v1/drugs/search"
response = requests.get(url, params={"query": "aspirin"})
print(f"Status: {response.status_code}")
print(f"URL: {response.url}")
print(f"Response: {response.text}")

Common URL issues:

# Wrong: Missing API version
url = "https://api.example.com/drugs"  # 404

# Right: Include API version
url = "https://api.example.com/v1/drugs"  # 200

# Wrong: Incorrect placeholder replacement
url = f"https://api.example.com/drugs/{drug_id}"  # drug_id = None

# Right: Validate before replacing
if not drug_id:
    return {"error": "drug_id is required"}
url = f"https://api.example.com/drugs/{drug_id}"

# Wrong: Missing trailing slash
url = "https://api.example.com/drugs"  # Some APIs require this

# Right: Check API documentation
url = "https://api.example.com/drugs/"

Issue 6.4: Missing Error Handling

Symptoms:

• Tool crashes on API errors
• Unhandled exceptions
• No error messages returned

Fix: Add comprehensive error handling:

def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
    try:
        # Validate parameters
        if 'query' not in arguments:
            return {
                "status": "error",
                "error": "Missing required parameter: query"
            }
        
        # Make API request
        response = requests.get(url, params=arguments, timeout=30)
        response.raise_for_status()
        
        # Parse response
        data = response.json()
        
        return {
            "status": "success",
            "data": data
        }
        
    except requests.Timeout:
        return {
            "status": "error",
            "error": "Request timed out after 30 seconds",
            "suggestion": "Try again or use a more specific query"
        }
    
    except requests.ConnectionError as e:
        return {
            "status": "error",
            "error": "Failed to connect to API",
            "detail": str(e),
            "suggestion": "Check network connection and API availability"
        }
    
    except requests.HTTPError as e:
        return {
            "status": "error",
            "error": f"API request failed: {e.response.status_code}",
            "detail": e.response.text,
            "url": e.response.url
        }
    
    except ValueError as e:
        return {
            "status": "error",
            "error": "Invalid response from API",
            "detail": str(e)
        }
    
    except Exception as e:
        return {
            "status": "error",
            "error": f"Unexpected error: {type(e).__name__}",
            "detail": str(e)
        }

Phase 7: Final Verification

Step 7.1: Comprehensive Testing

Test matrix:

Step 7.2: Validation Checks

# Validate JSON
python3 -m json.tool src/tooluniverse/data/{category}_tools.json

# Check Python syntax
python3 -m py_compile src/tooluniverse/{category}_tool.py

# Check linting
pylint src/tooluniverse/{category}_tool.py

# Verify tool loads
python3 -c "from tooluniverse import ToolUniverse; tu = ToolUniverse(); print(tu.list_tools())"

# Check auto-generated wrappers exist
ls src/tooluniverse/tools/{category}_*.py

# Check tool name lengths
python scripts/check_tool_name_lengths.py --test-shortening

Checklist:

• [ ] JSON files are valid
• [ ] Python syntax is valid
• [ ] No linting errors
• [ ] All tools load without errors
• [ ] Tool functions exist in tools/ directory (auto-generated)
• [ ] Category registered if needed

Step 7.3: Documentation

Create example script in examples/{category}_tools_example.py:

"""
Example usage of {category} tools.
"""

from tooluniverse import ToolUniverse

def main():
    # Initialize ToolUniverse
    tu = ToolUniverse()
    
    # Example 1: Search for drugs
    print("Searching for aspirin...")
    result = tu.run_tool("search_drugs", {"query": "aspirin"})
    print(f"Found {result['count']} results")
    
    # Example 2: Get drug details
    if result['results']:
        drug_id = result['results'][0]['id']
        print(f"\nGetting details for drug {drug_id}...")
        details = tu.run_tool("get_drug_details", {"drug_id": drug_id})
        print(f"Name: {details['name']}")
        print(f"Manufacturer: {details['manufacturer']}")
    
    # Example 3: Error handling
    print("\nTesting error handling...")
    error_result = tu.run_tool("get_drug_details", {"drug_id": "invalid"})
    print(f"Error: {error_result.get('error')}")

if __name__ == "__main__":
    main()

Documentation checklist:

• [ ] Tool descriptions are clear and complete
• [ ] Parameter descriptions include examples
• [ ] Return schemas match actual output
• [ ] Example script created in examples/
• [ ] Document findings and fixes in commit message

Summary Checklist

Quick reference for complete tool improvement:

✓ Phase 1: Initial Assessment

• Identify all tool files
• Verify basic structure

✓ Phase 2: Functionality Testing

• Test tool execution
• Test API endpoints directly

✓ Phase 3: Description Improvement

• Review tool descriptions
• Review parameter descriptions
• Review return schemas

✓ Phase 4: Error Handling

• Review current error handling
• Improve error messages
• Add retry logic if needed

✓ Phase 5: Finding Missing Tools

• Research API capabilities
• Create gap analysis matrix
• Identify subset extraction opportunities

✓ Phase 6: Fix Common Issues

• Fix tool class name mismatches
• Fix response format mismatches
• Fix endpoint URL issues
• Add missing error handling

✓ Phase 7: Final Verification

• Comprehensive testing
• Validation checks
• Documentation updates

For large API expansions (many endpoints in one category):

• [ ] Use a generic REST tool + JSON configs to cover multiple endpoints
• [ ] Verify real API behavior with live requests (prefer working patterns over docs)
• [ ] Make return_schema match the tool's wrapper and validate upstream payload structure at a useful depth (paging shapes, required IDs, important nested fields)
• [ ] Use real IDs from search/list endpoints in test_examples
• [ ] Remove tools for endpoints that have no working or replacement API
• [ ] Design "research-first": include discovery tools (search/list), detail tools (get by ID), and version/release tools for reproducibility
• [ ] Keep schemas and parameter surfaces LLM-friendly: avoid enormous enums; keep descriptions explicit; enforce strict validation in code