Loading...
Loading...
Create new scientific tools for ToolUniverse framework with proper structure, validation, and testing. Use when users need to add tools to ToolUniverse, implement new API integrations, create tool wrappers for scientific databases/services, expand ToolUniverse capabilities, or follow ToolUniverse contribution guidelines. Supports creating tool classes, JSON configurations, validation, error handling, and test examples.
npx skill4agent add mims-harvard/tooluniverse devtu-create-tooldefault_config.pytest_new_tools.py| SDK User (Using) | Tool Creator (Building) |
|---|---|
| |
| Handle responses | Design schemas |
| One-level usage | Three-step registration |
Stage 1: Tool Class Stage 2: Wrappers (Auto-Generated)
Python Implementation From JSON Configs
↓ ↓
@register_tool("MyTool") MyAPI_list_items()
class MyTool(BaseTool): MyAPI_search()
def run(arguments): MyAPI_get_details()@register_tool("MyAPITool") # Decorator registers class
class MyAPITool(BaseTool):
pass# In src/tooluniverse/default_config.py
TOOLS_CONFIGS = {
"my_category": os.path.join(current_dir, "data", "my_category_tools.json"),
}tu = ToolUniverse()
tu.load_tools() # Auto-generates wrappers in tools/import sys
sys.path.insert(0, 'src')
# Step 1: Check class registered
from tooluniverse.tool_registry import get_tool_registry
import tooluniverse.your_tool_module
registry = get_tool_registry()
assert "YourToolClass" in registry, "❌ Step 1 FAILED"
print("✅ Step 1: Class registered")
# Step 2: Check config registered
from tooluniverse.default_config import TOOLS_CONFIGS
assert "your_category" in TOOLS_CONFIGS, "❌ Step 2 FAILED"
print("✅ Step 2: Config registered")
# Step 3: Check wrappers generated
from tooluniverse import ToolUniverse
tu = ToolUniverse()
tu.load_tools()
assert hasattr(tu.tools, 'YourCategory_operation1'), "❌ Step 3 FAILED"
print("✅ Step 3: Wrappers generated")
print(f"✅ All steps complete!"){
"status": "success" | "error",
"data": {...}, // On success
"error": "message" // On failure
}src/tooluniverse/my_api_tool.pysrc/tooluniverse/data/my_api_tools.jsontests/unit/test_my_api_tool.pyexamples/my_api_examples.pysrc/tooluniverse/tools/MyAPI_*.pyfrom typing import Dict, Any
from tooluniverse.tool import BaseTool
from tooluniverse.tool_utils import register_tool
import requests
@register_tool("MyAPITool")
class MyAPITool(BaseTool):
"""Tool for MyAPI database."""
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]:
"""Route to operation handler."""
operation = arguments.get("operation")
if not operation:
return {"status": "error", "error": "Missing: operation"}
if operation == "list_items":
return self._list_items(arguments)
elif operation == "search":
return self._search(arguments)
else:
return {"status": "error", "error": f"Unknown: {operation}"}
def _list_items(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""List items with pagination."""
try:
params = {}
if "limit" in arguments:
params["limit"] = arguments["limit"]
response = requests.get(
f"{self.BASE_URL}/items",
params=params,
timeout=30
)
response.raise_for_status()
data = response.json()
return {
"status": "success",
"data": data.get("items", []),
"total": data.get("total", 0)
}
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)}
def _search(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Search items by query."""
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",
"results": data.get("results", []),
"count": data.get("count", 0)
}
except requests.exceptions.RequestException as e:
return {"status": "error", "error": f"API failed: {str(e)}"}[
{
"name": "MyAPI_list_items",
"class": "MyAPITool",
"description": "List items from database with pagination. Returns item IDs and names. Supports filtering by status and type. Example: limit=10 returns first 10 items.",
"parameter": {
"type": "object",
"required": ["operation"],
"properties": {
"operation": {
"const": "list_items",
"description": "Operation type (fixed)"
},
"limit": {
"type": "integer",
"description": "Max results (1-100)",
"minimum": 1,
"maximum": 100
}
}
},
"return": {
"type": "object",
"properties": {
"status": {"type": "string", "enum": ["success", "error"]},
"data": {"type": "array"},
"total": {"type": "integer"},
"error": {"type": "string"}
},
"required": ["status"]
},
"test_examples": [
{
"operation": "list_items",
"limit": 10
}
]
}
]import time
def _submit_job(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Submit job and poll for results."""
try:
# Submit
submit_response = requests.post(
f"{self.BASE_URL}/jobs/submit",
json={"data": arguments.get("data")},
timeout=30
)
submit_response.raise_for_status()
job_id = submit_response.json().get("job_id")
# Poll
for attempt in range(60): # 2 min max
status_response = requests.get(
f"{self.BASE_URL}/jobs/{job_id}/status",
timeout=30
)
status_response.raise_for_status()
result = status_response.json()
if result.get("status") == "completed":
return {
"status": "success",
"data": result.get("results"),
"job_id": job_id
}
elif result.get("status") == "failed":
return {
"status": "error",
"error": result.get("error"),
"job_id": job_id
}
time.sleep(2) # Poll every 2s
return {"status": "error", "error": "Timeout after 2 min"}
except requests.exceptions.RequestException as e:
return {"status": "error", "error": str(e)}required_api_keys{
"name": "NVIDIA_ESMFold_predict",
"required_api_keys": ["NVIDIA_API_KEY"],
...
}optional_api_keys{
"name": "PubMed_search_articles",
"optional_api_keys": ["NCBI_API_KEY"],
"description": "Search PubMed. Rate limits: 3 req/sec without key, 10 req/sec with NCBI_API_KEY.",
...
}optional_api_keysdef __init__(self, tool_config):
super().__init__(tool_config)
# Read from environment variable only (not as parameter)
self.api_key = os.environ.get("NCBI_API_KEY", "")
def run(self, arguments):
# Adjust behavior based on key availability
has_key = bool(self.api_key)
rate_limit = 0.1 if has_key else 0.4 # Faster with key
...api_key{API}_{action}_{target}FDA_get_drug_infoFDA_get_detailed_drug_information_with_history{
"description": "Search database for items. Returns up to 100 results with scores. Supports wildcards (* ?) and Boolean operators (AND, OR, NOT). Example: 'protein AND membrane' finds membrane proteins."
}{
"test_examples": [
{
"operation": "search",
"query": "protein", // ✅ Real, common term
"limit": 10
}
]
}"id": "XXXXX""placeholder": "example_123""type"@register_toolreturn_schemareturntest_examplesexamplesparameterreturn_schemadescriptionenumsrc/tooluniverse/__init__.pyvalidate_parameters(arguments)For the full implementation plan, maintenance checklist, and large API expansion guidance, see references/implementation-guide.md
scripts/test_new_tools.pytest_examplesreturn_schema# Test your specific tools
python scripts/test_new_tools.py your_tool_name
# Test with verbose output
python scripts/test_new_tools.py your_tool_name -v
# Test all tools (for full validation)
python scripts/test_new_tools.py
# Stop on first failure
python scripts/test_new_tools.py your_tool_name --fail-fastreturn_schema| Failure | Cause | Fix |
|---|---|---|
| 404 ERROR | Invalid ID in test_examples | Use real IDs from API docs |
| Schema Mismatch | Response doesn't match return_schema | Update schema or fix response format |
| Exception | Code bug or missing dependency | Check error message, fix implementation |
import json
from tooluniverse.your_tool_module import YourToolClass
def test_direct_class():
"""Test implementation logic."""
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"
assert "data" in resultimport pytest
from tooluniverse import ToolUniverse
class TestYourTools:
@pytest.fixture
def tu(self):
tu = ToolUniverse()
tu.load_tools() # CRITICAL
return tu
def test_tools_load(self, tu):
"""Verify registration."""
assert hasattr(tu.tools, 'YourTool_operation1')
def test_execution(self, tu):
"""Test via ToolUniverse (how users call it)."""
result = tu.tools.YourTool_operation1(**{
"operation": "operation1",
"param": "value"
})
assert result["status"] == "success"
def test_error_handling(self, tu):
"""Test missing params."""
result = tu.tools.YourTool_operation1(**{
"operation": "operation1"
# Missing required param
})
assert result["status"] == "error"def test_real_api():
"""Verify actual API integration."""
tu = ToolUniverse()
tu.load_tools()
result = tu.tools.YourTool_operation1(**{
"operation": "operation1",
"param": "real_value_from_docs"
})
if result["status"] == "success":
assert "data" in result
print("✅ Real API works")
else:
print(f"⚠️ API error (may be down): {result['error']}")curl -s https://pypi.org/pypi/PACKAGE/json | python3 -c "
import json, sys
data = json.load(sys.stdin)
print(f'Dependencies: {len(data[\"info\"][\"requires_dist\"] or [])}')
"[project.dependencies][project.optional-dependencies]try:
import optional_package
except ImportError:
return {
"status": "error",
"error": "Install with: pip install optional_package"
}def _list_items(self, arguments):
params = {}
if "page" in arguments:
params["page"] = arguments["page"]
if "limit" in arguments:
params["limit"] = arguments["limit"]
response = requests.get(url, params=params, timeout=30)
data = response.json()
return {
"status": "success",
"data": data.get("items", []),
"page": data.get("page", 0),
"total_pages": data.get("total_pages", 1),
"total_items": data.get("total", 0)
}AttributeError# Edit src/tooluniverse/default_config.py
TOOLS_CONFIGS = {
# ... existing ...
"your_category": os.path.join(current_dir, "data", "your_category_tools.json"),
}grep "your_category" src/tooluniverse/default_config.py
ls src/tooluniverse/tools/YourCategory_*.py
python3 -c "from tooluniverse import ToolUniverse; tu = ToolUniverse(); tu.load_tools(); print(hasattr(tu.tools, 'YourCategory_op1'))"@register_toolreturn_schemadefault_config.pytu.load_tools()python scripts/test_new_tools.py your_tool -vpython scripts/check_tool_name_lengths.py --test-shorteningFor the full development checklist and maintenance phases, see references/implementation-guide.md
# Validate JSON
python3 -m json.tool src/tooluniverse/data/your_tools.json
# Check Python syntax
python3 -m py_compile src/tooluniverse/your_tool.py
# Verify registration
grep "your_category" src/tooluniverse/default_config.py
# Generate wrappers
PYTHONPATH=src python3 -m tooluniverse.generate_tools --force
# List wrappers
ls src/tooluniverse/tools/YourCategory_*.py
# Run unit tests
pytest tests/unit/test_your_tool.py -v
# MANDATORY: Run test_new_tools.py validation
python scripts/test_new_tools.py your_tool -v
# Count tools
python3 << 'EOF'
from tooluniverse import ToolUniverse
tu = ToolUniverse()
tu.load_tools()
print(f"Total: {len([t for t in dir(tu.tools) if 'YourCategory' in t])} tools")
EOFpython scripts/test_new_tools.py your_tool -voptional_api_keysapi_keytest_new_tools.py